gRPC 负载均衡

LoadBalancerRegistry.getDefaultRegistry().register(new HealthCheckingRoundRobinLoadBalancerProvider());

ManagedChannelBuilder.forTarget("server")
                     .defaultLoadBalancingPolicy("round_robin")
                     .build();

this.loadBalancerFactory = new AutoConfiguredLoadBalancerFactory(builder.defaultLbPolicy);

// 构建新的 lbHelper
LbHelperImpl lbHelper = new LbHelperImpl();

// 自动配置负载均衡
lbHelper.lb = loadBalancerFactory.newLoadBalancer(lbHelper);

this.lbHelper = lbHelper;

public AutoConfiguredLoadBalancer newLoadBalancer(Helper helper) {
  return new AutoConfiguredLoadBalancer(helper);
}

AutoConfiguredLoadBalancer(Helper helper) {
  this.helper = helper;
  // 从注册器中获取默认的负载均衡策略提供器
  delegateProvider = registry.getProvider(defaultPolicy);
  if (delegateProvider == null) {
    throw new IllegalStateException("Could not find policy '" + defaultPolicy
            + "'. Make sure its implementation is either registered to LoadBalancerRegistry or"
            + " included in META-INF/services/io.grpc.LoadBalancerProvider from your jar files.");
  }
  // 创建新的
  delegate = delegateProvider.newLoadBalancer(helper);
}

public LoadBalancer newLoadBalancer(Helper helper) {
    return HealthCheckingLoadBalancerUtil.newHealthCheckingLoadBalancer(rrProvider, helper);
}

public static LoadBalancer newHealthCheckingLoadBalancer(Factory factory, Helper helper) {
    // 创建工厂
    HealthCheckingLoadBalancerFactory hcFactory = new HealthCheckingLoadBalancerFactory(factory,
            new ExponentialBackoffPolicy.Provider(),
            GrpcUtil.STOPWATCH_SUPPLIER);

    // 使用工厂创建 LoadBalancer
    return hcFactory.newLoadBalancer(helper);
}

public LoadBalancer newLoadBalancer(Helper helper) {
    // 代理 Helper
    HelperImpl wrappedHelper = new HelperImpl(helper);
    // 创建 LoadBalancer
    LoadBalancer delegateBalancer = delegateFactory.newLoadBalancer(wrappedHelper);
    return new HealthCheckingLoadBalancer(wrappedHelper, delegateBalancer);
}

// 服务发现监听器
NameResolverListener listener = new NameResolverListener(lbHelper, nameResolver);
nameResolver.start(listener);

public void run() {

  List<EquivalentAddressGroup> servers = resolutionResult.getAddresses();

  nameResolverBackoffPolicy = null;
  // ...
  ManagedChannelServiceConfig effectiveServiceConfig;
    effectiveServiceConfig = defaultServiceConfig == null ? EMPTY_SERVICE_CONFIG : defaultServiceConfig;
        
  // 获取属性
  Attributes effectiveAttrs = resolutionResult.getAttributes();
  // 如果服务发现没有关闭
  if (NameResolverListener.this.helper == ManagedChannelImpl.this.lbHelper) {
    // 获取健康检查
    Map<String, ?> healthCheckingConfig = effectiveServiceConfig.getHealthCheckingConfig();
    // 构建健康检查配置
    if (healthCheckingConfig != null) {
      effectiveAttrs = effectiveAttrs.toBuilder()
                                     .set(LoadBalancer.ATTR_HEALTH_CHECKING_CONFIG, healthCheckingConfig)
                                     .build();
    }

    // 更新负载均衡算法，处理未处理的请求
    Status handleResult = helper.lb.tryHandleResolvedAddresses(
            ResolvedAddresses.newBuilder()
                             .setAddresses(servers)
                             .setAttributes(effectiveAttrs)
                             .setLoadBalancingPolicyConfig(effectiveServiceConfig.getLoadBalancingConfig())
                             .build());

    if (!handleResult.isOk()) {
      handleErrorInSyncContext(handleResult.augmentDescription(resolver + " was used"));
    }
  }
}

Status tryHandleResolvedAddresses(ResolvedAddresses resolvedAddresses) {
  List<EquivalentAddressGroup> servers = resolvedAddresses.getAddresses();
  Attributes attributes = resolvedAddresses.getAttributes();

  // 负载均衡选择
  PolicySelection policySelection = (PolicySelection) resolvedAddresses.getLoadBalancingPolicyConfig();

  if (policySelection == null) {
    LoadBalancerProvider defaultProvider;
    // 更新负载均衡提供器
    defaultProvider = getProviderOrThrow(defaultPolicy, "using default policy");
    policySelection = new PolicySelection(defaultProvider, null, null);
  }

  Object lbConfig = policySelection.config;

  // 设置负载均衡算法参数
  if (lbConfig != null) {
    attributes = attributes.toBuilder()
                           .set(ATTR_LOAD_BALANCING_CONFIG, policySelection.rawConfig)
                           .build();
  }

  // 负载均衡器
  LoadBalancer delegate = getDelegate();
  // 如果地址是空的，或者处理失败，则返回错误
  if (resolvedAddresses.getAddresses().isEmpty() && !delegate.canHandleEmptyAddressListFromNameResolution()) {
    return Status.UNAVAILABLE.withDescription("NameResolver returned no usable address. addrs=" + servers + ", attrs=" + attributes);
  } else {
    // 返回处理成功
    delegate.handleResolvedAddresses(
            ResolvedAddresses.newBuilder()
                             .setAddresses(resolvedAddresses.getAddresses())
                             .setAttributes(attributes)
                             .setLoadBalancingPolicyConfig(lbConfig)
                             .build());
    return Status.OK;
  }
}

public void handleResolvedAddresses(ResolvedAddresses resolvedAddresses) {
    // 获取健康检查配置
    Map<String, ?> healthCheckingConfig = resolvedAddresses.getAttributes()
                                                           .get(LoadBalancer.ATTR_HEALTH_CHECKING_CONFIG);
    // 获取服务的健康检查配置
    String serviceName = ServiceConfigUtil.getHealthCheckedServiceName(healthCheckingConfig);

    // 配置服务健康检查
    helper.setHealthCheckedService(serviceName);
    // 调用被代理的类处理地址
    super.handleResolvedAddresses(resolvedAddresses);
}

public void handleResolvedAddresses(ResolvedAddresses resolvedAddresses) {

    // 获取地址列表
    List<EquivalentAddressGroup> servers = resolvedAddresses.getAddresses();

    // 当前的地址
    Set<EquivalentAddressGroup> currentAddrs = subchannels.keySet();

    // 将地址 List 转为 Map
    Map<EquivalentAddressGroup, EquivalentAddressGroup> latestAddrs = stripAttrs(servers);

    // 根据当前的地址，获取需要移除的地址，返回的地址是现有地址中有，新的地址中没有的
    Set<EquivalentAddressGroup> removedAddrs = setsDifference(currentAddrs, latestAddrs.keySet());


    for (Map.Entry<EquivalentAddressGroup, EquivalentAddressGroup> latestEntry : latestAddrs.entrySet()) {

        // 不含 Attributes 的 EquivalentAddressGroup
        EquivalentAddressGroup strippedAddressGroup = latestEntry.getKey();
        // 包含 Attributes 的 EquivalentAddressGroup
        EquivalentAddressGroup originalAddressGroup = latestEntry.getValue();

        // 根据地址获取对应的已经存在的 Subchannel
        Subchannel existingSubchannel = subchannels.get(strippedAddressGroup);

        // 如果存在已有的 Subchannel，则更新地址并跳出
        if (existingSubchannel != null) {
            // EAG's Attributes may have changed.
            // 更新地址
            existingSubchannel.updateAddresses(Collections.singletonList(originalAddressGroup));
            continue;
        }
        // 根据地址创建新的 Subchannel
        // Create new subchannels for new addresses.

        // 设置新的连接状态是 IDLE
        Attributes.Builder subchannelAttrs = Attributes.newBuilder()
                                                       .set(STATE_INFO, new Ref<>(ConnectivityStateInfo.forNonError(IDLE)));

        // 创建新 Subchannel
        final Subchannel subchannel = checkNotNull(helper.createSubchannel(CreateSubchannelArgs.newBuilder()
                                                                                               .setAddresses(originalAddressGroup)
                                                                                               .setAttributes(subchannelAttrs.build())
                                                                                               .build()),
                "subchannel");

        // 启动 Subchannel 状态监听器
        subchannel.start(new SubchannelStateListener() {
            @Override
            public void onSubchannelState(ConnectivityStateInfo state) {
                // 处理状态变化
                processSubchannelState(subchannel, state);
            }
        });

        // 将新创建的 Subchannel 放在 Subchannel 的 Map 中
        subchannels.put(strippedAddressGroup, subchannel);
        // 要求建立连接
        subchannel.requestConnection();
    }

    // 移除不包含的地址
    ArrayList<Subchannel> removedSubchannels = new ArrayList<>();
    for (EquivalentAddressGroup addressGroup : removedAddrs) {
        removedSubchannels.add(subchannels.remove(addressGroup));
    }

    // 在关闭 Subchannel 之前更新 picker，减少关闭期间的风险
    updateBalancingState();

    // 关闭被移除的 Subchannel
    for (Subchannel removedSubchannel : removedSubchannels) {
        shutdownSubchannel(removedSubchannel);
    }
}

ClientTransport transport = clientTransportProvider.get(new PickSubchannelArgsImpl(method, headers, callOptions));

public ClientTransport get(PickSubchannelArgs args) {
  SubchannelPicker pickerCopy = subchannelPicker;
  // 如果是关闭状态，则停止调用
  if (shutdown.get()) {
    return delayedTransport;
  }
  
  // 如果是 SubchannelPicker 是空的，则退出 idle 模模式，返回 delayedTransport
  if (pickerCopy == null) {
    final class ExitIdleModeForTransport implements Runnable {
      @Override
      public void run() {
        // 退出 idle 模式，将会创建 LoadBalancer,NameResovler
        exitIdleMode();
      }
    }

    syncContext.execute(new ExitIdleModeForTransport());
    return delayedTransport;
  }
  // 选择某个 SubChannel 发起调用，即选择某个服务端
  PickResult pickResult = pickerCopy.pickSubchannel(args);
  ClientTransport transport = GrpcUtil.getTransportFromPickResult(pickResult, args.getCallOptions().isWaitForReady());
  // 如果有 Transport，则返回
  if (transport != null) {
    return transport;
  }
  return delayedTransport;
}

public PickResult pickSubchannel(PickSubchannelArgs args) {
    return PickResult.withSubchannel(nextSubchannel());
}

private Subchannel nextSubchannel() {
    int size = list.size();
    int i = indexUpdater.incrementAndGet(this);
    if (i >= size) {
        int oldi = i;
        i %= size;
        indexUpdater.compareAndSet(this, oldi, i);
    }
    return list.get(i);
}