Java 类io.grpc.util.RoundRobinLoadBalancerFactory 实例源码

@Test
public void testRoundRobinLoadBalancerFactory() throws Exception {

  try (Client client = Client.builder()
          .endpoints(TestConstants.endpoints)
          .loadBalancerFactory(RoundRobinLoadBalancerFactory.getInstance())
          .build();
       KV kv = client.getKVClient()) {
    PutResponse response;
    long lastMemberId = 0;
    long differences = 0;

    for (int i = 0; i < TestConstants.endpoints.length; i++) {
      response = kv.put(TestUtil.randomByteSequence(), TestUtil.randomByteSequence()).get();

      if (i > 0 && lastMemberId != response.getHeader().getMemberId()) {
        differences++;
      }

      lastMemberId = response.getHeader().getMemberId();
    }

    assertThat(differences).isNotEqualTo(lastMemberId);
  }
}

public static void main(String[] args) throws InterruptedException, UnknownHostException {
  String target = System.getenv("ECHO_SERVICE_TARGET");
  if (target == null || target.isEmpty()) {
    target = "localhost:8080";
  }
  final ManagedChannel channel = ManagedChannelBuilder.forTarget(target)
      .nameResolverFactory(new KubernetesNameResolverProvider())  // this is on by default
      .loadBalancerFactory(RoundRobinLoadBalancerFactory.getInstance())
      .usePlaintext(true)
      .build();

  final String self = InetAddress.getLocalHost().getHostName();

  ExecutorService executorService = Executors.newFixedThreadPool(THREADS);
  for (int i = 0; i < THREADS; i++) {
    EchoServiceGrpc.EchoServiceBlockingStub stub = EchoServiceGrpc.newBlockingStub(channel);
    executorService.submit(() -> {
      while (true) {
        EchoResponse response = stub.echo(EchoRequest.newBuilder()
            .setMessage(self + ": " + Thread.currentThread().getName())
            .build());
        System.out.println(response.getFrom() + " echoed");

        Thread.sleep(RANDOM.nextInt(700));
      }
    });
  }
}

public static void main(String[] args) throws InterruptedException, UnknownHostException {
  String target = System.getenv("ECHO_SERVICE_TARGET");
  if (target == null || target.isEmpty()) {
    target = "localhost:8080";
  }
  final ManagedChannel channel = ManagedChannelBuilder.forTarget(target)
      .nameResolverFactory(new DnsNameResolverProvider())  // this is on by default
      .loadBalancerFactory(RoundRobinLoadBalancerFactory.getInstance())
      .usePlaintext(true)
      .build();

  final String self = InetAddress.getLocalHost().getHostName();

  ExecutorService executorService = Executors.newFixedThreadPool(THREADS);
  for (int i = 0; i < THREADS; i++) {
    EchoServiceGrpc.EchoServiceBlockingStub stub = EchoServiceGrpc.newBlockingStub(channel);
    executorService.submit(() -> {
      while (true) {
        EchoResponse response = stub.echo(EchoRequest.newBuilder()
            .setMessage(self + ": " + Thread.currentThread().getName())
            .build());
        System.out.println(response.getFrom() + " echoed");

        Thread.sleep(RANDOM.nextInt(700));
      }
    });
  }
}

@Override
public Channel createChannel(String name) {
 RoundRobinLoadBalancerFactory instance = RoundRobinLoadBalancerFactory.getInstance();
  return ManagedChannelBuilder.forTarget(name)
      .loadBalancerFactory(instance)
      .nameResolverFactory(new DiscoveryClientResolverFactory(client))
      .usePlaintext(channels.getChannels().get(name).isPlaintext()).build();
}

/**
 * Creates the NameResolver instance.
 *
 * @param authority         The authority string used to create the URI.
 * @param bootstrapServers  The initial set of controller endpoints.
 * @param enableDiscovery   Whether to use the controller's discovery API.
 */
ControllerNameResolver(final String authority, final List<InetSocketAddress> bootstrapServers,
                       final boolean enableDiscovery) {
    this.authority = authority;
    this.bootstrapServers = ImmutableList.copyOf(bootstrapServers);
    this.enableDiscovery = enableDiscovery;
    if (this.enableDiscovery) {
        // We will use the direct scheme to send the discovery RPC request to the controller bootstrap servers.
        String connectString = "tcp://";
        final List<String> strings = this.bootstrapServers.stream()
                .map(server -> server.getHostString() + ":" + server.getPort())
                .collect(Collectors.toList());
        connectString = connectString + String.join(",", strings);

        this.client = ControllerServiceGrpc.newBlockingStub(ManagedChannelBuilder
                .forTarget(connectString)
                .nameResolverFactory(new ControllerResolverFactory())
                .loadBalancerFactory(RoundRobinLoadBalancerFactory.getInstance())
                .usePlaintext(true)
                .build());
    } else {
        this.client = null;
    }

    // We enable the periodic refresh only if controller discovery is enabled or if DNS resolution is required.
    if (this.enableDiscovery || this.bootstrapServers.stream().anyMatch(
            inetSocketAddress -> !InetAddresses.isInetAddress(inetSocketAddress.getHostString()))) {
        this.scheduledExecutor = ExecutorServiceHelpers.newScheduledThreadPool(1, "fetch-controllers");
    } else {
        this.scheduledExecutor = null;
    }
}

/**
 * Create a new gRPC {@link ManagedChannel}.
 *
 * @throws IOException in case the channel can't be constructed.
 */
public static ManagedChannel newChannel(String target, AuthAndTLSOptions options)
    throws IOException {
  Preconditions.checkNotNull(target);
  Preconditions.checkNotNull(options);

  final SslContext sslContext =
      options.tlsEnabled ? createSSlContext(options.tlsCertificate) : null;

  try {
    NettyChannelBuilder builder =
        NettyChannelBuilder.forTarget(target)
            .negotiationType(options.tlsEnabled ? NegotiationType.TLS : NegotiationType.PLAINTEXT)
            .loadBalancerFactory(RoundRobinLoadBalancerFactory.getInstance());
    if (sslContext != null) {
      builder.sslContext(sslContext);
      if (options.tlsAuthorityOverride != null) {
        builder.overrideAuthority(options.tlsAuthorityOverride);
      }
    }
    return builder.build();
  } catch (RuntimeException e) {
    // gRPC might throw all kinds of RuntimeExceptions: StatusRuntimeException,
    // IllegalStateException, NullPointerException, ...
    String message = "Failed to connect to '%s': %s";
    throw new IOException(String.format(message, target, e.getMessage()));
  }
}

@Override
public LoadBalancer newLoadBalancer(LoadBalancer.Helper helper) {
  return new GrpclbLoadBalancer(
      helper, PickFirstBalancerFactory.getInstance(),
      RoundRobinLoadBalancerFactory.getInstance(),
      // TODO(zhangkun83): balancer sends load reporting RPCs from it, which also involves
      // channelExecutor thus may also run other tasks queued in the channelExecutor.  If such
      // load should not be on the shared scheduled executor, we should use a combination of the
      // scheduled executor and the default app executor.
      SharedResourcePool.forResource(GrpcUtil.TIMER_SERVICE),
      TIME_PROVIDER);
}

@ConditionalOnMissingBean
@Bean
public LoadBalancer.Factory defaultGrpcLoadBalancerFactory() {
    return RoundRobinLoadBalancerFactory.getInstance();
}

CentralConnection(String collectorAddress, @Nullable String collectorAuthority, File confDir,
        @Nullable File sharedConfDir, AtomicBoolean inConnectionFailure) throws SSLException {
    ParsedCollectorAddress parsedCollectorAddress = parseCollectorAddress(collectorAddress);
    eventLoopGroup = EventLoopGroups.create("Glowroot-GRPC-Worker-ELG");
    channelExecutor =
            Executors.newSingleThreadExecutor(ThreadFactories.create("Glowroot-GRPC-Executor"));
    String authority;
    if (collectorAuthority != null) {
        authority = collectorAuthority;
    } else if (parsedCollectorAddress.addresses().size() == 1) {
        authority = parsedCollectorAddress.addresses().get(0).getHostName();
    } else if (!parsedCollectorAddress.https()) {
        authority = "dummy-service-authority";
    } else {
        throw new IllegalStateException("collector.authority is required when using client"
                + " side load balancing to connect to a glowroot central cluster over HTTPS");
    }
    NettyChannelBuilder builder = NettyChannelBuilder
            .forTarget("dummy-target")
            .nameResolverFactory(new SimpleNameResolverFactory(
                    parsedCollectorAddress.addresses(), authority))
            .loadBalancerFactory(RoundRobinLoadBalancerFactory.getInstance())
            .eventLoopGroup(eventLoopGroup)
            .executor(channelExecutor)
            // aggressive keep alive, shouldn't even be used since gauge data is sent every
            // 5 seconds and keep alive will only kick in after 30 seconds of not hearing back
            // from the server
            .keepAliveTime(30, SECONDS);
    if (parsedCollectorAddress.https()) {
        SslContextBuilder sslContext = GrpcSslContexts.forClient();
        File trustCertCollectionFile = getTrustCertCollectionFile(confDir, sharedConfDir);
        if (trustCertCollectionFile != null) {
            sslContext.trustManager(trustCertCollectionFile);
        }
        channel = builder.sslContext(sslContext.build())
                .negotiationType(NegotiationType.TLS)
                .build();
    } else {
        channel = builder.negotiationType(NegotiationType.PLAINTEXT)
                .build();
    }
    retryExecutor = Executors.newSingleThreadScheduledExecutor(
            ThreadFactories.create("Glowroot-Collector-Retry"));
    this.inConnectionFailure = inConnectionFailure;
    this.collectorAddress = collectorAddress;
}

/**
 * Creates a new instance of the Controller client class.
 *
 * @param controllerURI The controller rpc URI. This can be of 2 types
 *                      1. tcp://ip1:port1,ip2:port2,...
 *                          This is used if the controller endpoints are static and can be directly accessed.
 *                      2. pravega://ip1:port1,ip2:port2,...
 *                          This is used to autodiscovery the controller endpoints from an initial controller list.
 * @param config        The configuration for this client implementation.
 * @param executor      The executor service to be used for handling retries.
 */
public ControllerImpl(final URI controllerURI, final ControllerImplConfig config,
                      final ScheduledExecutorService executor) {
    this(NettyChannelBuilder.forTarget(controllerURI.toString())
            .nameResolverFactory(new ControllerResolverFactory())
            .loadBalancerFactory(RoundRobinLoadBalancerFactory.getInstance())
            .keepAliveTime(DEFAULT_KEEPALIVE_TIME_MINUTES, TimeUnit.MINUTES)
            .usePlaintext(true), config, executor);
    log.info("Controller client connecting to server at {}", controllerURI.getAuthority());
}