private CronetTransportFactory(
    StreamBuilderFactory streamFactory,
    Executor executor,
    @Nullable ScheduledExecutorService timeoutService,
    int maxMessageSize,
    boolean alwaysUsePut,
    TransportTracer transportTracer) {
  usingSharedScheduler = timeoutService == null;
  this.timeoutService = usingSharedScheduler
      ? SharedResourceHolder.get(GrpcUtil.TIMER_SERVICE) : timeoutService;
  this.maxMessageSize = maxMessageSize;
  this.alwaysUsePut = alwaysUsePut;
  this.streamFactory = streamFactory;
  this.executor = Preconditions.checkNotNull(executor, "executor");
  this.transportTracer = Preconditions.checkNotNull(transportTracer, "transportTracer");
}
 

NettyTransportFactory(ProtocolNegotiator protocolNegotiator,
    Class<? extends Channel> channelType, Map<ChannelOption<?>, ?> channelOptions,
    EventLoopGroup group, int flowControlWindow, int maxMessageSize, int maxHeaderListSize,
    long keepAliveTimeNanos, long keepAliveTimeoutNanos, boolean keepAliveWithoutCalls,
    TransportTracer transportTracer, LocalSocketPicker localSocketPicker) {
  this.protocolNegotiator = protocolNegotiator;
  this.channelType = channelType;
  this.channelOptions = new HashMap<ChannelOption<?>, Object>(channelOptions);
  this.flowControlWindow = flowControlWindow;
  this.maxMessageSize = maxMessageSize;
  this.maxHeaderListSize = maxHeaderListSize;
  this.keepAliveTimeNanos = new AtomicBackoff("keepalive time nanos", keepAliveTimeNanos);
  this.keepAliveTimeoutNanos = keepAliveTimeoutNanos;
  this.keepAliveWithoutCalls = keepAliveWithoutCalls;
  this.transportTracer = transportTracer;
  this.localSocketPicker =
      localSocketPicker != null ? localSocketPicker : new LocalSocketPicker();

  usingSharedGroup = group == null;
  if (usingSharedGroup) {
    // The group was unspecified, using the shared group.
    this.group = SharedResourceHolder.get(Utils.DEFAULT_WORKER_EVENT_LOOP_GROUP);
  } else {
    this.group = group;
  }
}
 

/**
 * 将当前出错的服务器从备选列表中去除
 *
 * @author sxp
 * @since 2018-6-21
 * @since 2019/12/11 modify by wlh 10分钟（时间可配）后，将服务重新放回至服务提供列表
 */
private static void removeCurrentProvider(NameResolver nameResolver, String providerId, String method) {
  Map<String, ServiceProvider> providersForLoadBalance = nameResolver.getProvidersForLoadBalance();
  if (providersForLoadBalance == null || providersForLoadBalance.size() == 0) {
    logger.info("客户端的备选列表为空", providerId);
    return;
  }

  if (providersForLoadBalance.containsKey(providerId)) {
    logger.error("FATAL ERROR : 服务器节点{}连续调用出错{}次，从客户端备选服务器列表中删除", providerId, switchoverThreshold);
    providersForLoadBalance.remove(providerId);
    nameResolver.reCalculateProvidersCountAfterLoadBalance(method);

    if (timerService == null) {
      timerService = SharedResourceHolder.get(GrpcUtil.TIMER_SERVICE);
    }
    timerService.schedule(new RecoveryServerRunnable(nameResolver, providerId, method), recoveryMilliseconds, TimeUnit.MILLISECONDS);
  }
}
 

@Override
public final synchronized void shutdown() {
  if (shutdown) {
    return;
  }
  shutdown = true;
  if (timerService != null) {
    timerService = SharedResourceHolder.release(timerServiceResource, timerService);
  }
  if (executor != null) {
    executor = SharedResourceHolder.release(executorResource, executor);
  }

  //----begin----自动注销zk中的Consumer信息----dengjq

  if (findZkFuture != null) {
    findZkFuture.cancel(false);
  }
  if (findZkExecutor != null) {
    findZkExecutor.shutdown();
  }

  unRegistry();

  //----end----自动注销zk中的Consumer信息----
}
 

@Override
@GuardedBy("this")
public void refresh() {
  if (refreshing) return;
  try {
    refreshing = true;

    Endpoints endpoints = kubernetesClient.endpoints().inNamespace(namespace)
        .withName(name)
        .get();

    if (endpoints == null) {
      // Didn't find anything, retrying
      ScheduledExecutorService timerService = SharedResourceHolder.get(timerServiceResource);
      timerService.schedule(() -> {
        refresh();
      }, 30, TimeUnit.SECONDS);
      return;
    }

    update(endpoints);
    watch();
  } finally {
    refreshing = false;
  }
}
 

private CronetTransportFactory(
    StreamBuilderFactory streamFactory,
    Executor executor,
    @Nullable ScheduledExecutorService timeoutService,
    int maxMessageSize,
    boolean alwaysUsePut,
    TransportTracer transportTracer,
    boolean useGetForSafeMethods,
    boolean usePutForIdempotentMethods) {
  usingSharedScheduler = timeoutService == null;
  this.timeoutService = usingSharedScheduler
      ? SharedResourceHolder.get(GrpcUtil.TIMER_SERVICE) : timeoutService;
  this.maxMessageSize = maxMessageSize;
  this.alwaysUsePut = alwaysUsePut;
  this.streamFactory = streamFactory;
  this.executor = Preconditions.checkNotNull(executor, "executor");
  this.transportTracer = Preconditions.checkNotNull(transportTracer, "transportTracer");
  this.useGetForSafeMethods = useGetForSafeMethods;
  this.usePutForIdempotentMethods = usePutForIdempotentMethods;
}
 

@Test
public void scheduledExecutorService_default() {
  CronetChannelBuilder builder = CronetChannelBuilder.forAddress("address", 1234, mockEngine);
  ClientTransportFactory clientTransportFactory = builder.buildTransportFactory();
  assertSame(
      SharedResourceHolder.get(TIMER_SERVICE),
      clientTransportFactory.getScheduledExecutorService());

  SharedResourceHolder.release(
      TIMER_SERVICE, clientTransportFactory.getScheduledExecutorService());
  clientTransportFactory.close();
}
 

/**
 * When the transport is in goAway state, we should stop it once all active streams finish.
 */
@GuardedBy("lock")
private void stopIfNecessary() {
  if (!(goAwayStatus != null && streams.isEmpty() && pendingStreams.isEmpty())) {
    return;
  }
  if (stopped) {
    return;
  }
  stopped = true;

  if (keepAliveManager != null) {
    keepAliveManager.onTransportTermination();
    // KeepAliveManager should stop using the scheduler after onTransportTermination gets called.
    scheduler = SharedResourceHolder.release(TIMER_SERVICE, scheduler);
  }

  if (ping != null) {
    ping.failed(getPingFailure());
    ping = null;
  }

  if (!goAwaySent) {
    // Send GOAWAY with lastGoodStreamId of 0, since we don't expect any server-initiated
    // streams. The GOAWAY is part of graceful shutdown.
    goAwaySent = true;
    frameWriter.goAway(0, ErrorCode.NO_ERROR, new byte[0]);
  }

  // We will close the underlying socket in the writing thread to break out the reader
  // thread, which will close the frameReader and notify the listener.
  frameWriter.close();
}
 

private OkHttpTransportFactory(Executor executor,
    @Nullable ScheduledExecutorService timeoutService,
    @Nullable SSLSocketFactory socketFactory,
    @Nullable HostnameVerifier hostnameVerifier,
    ConnectionSpec connectionSpec,
    int maxMessageSize,
    boolean enableKeepAlive,
    long keepAliveTimeNanos,
    long keepAliveTimeoutNanos,
    int flowControlWindow,
    boolean keepAliveWithoutCalls,
    int maxInboundMetadataSize,
    TransportTracer.Factory transportTracerFactory) {
  usingSharedScheduler = timeoutService == null;
  this.timeoutService = usingSharedScheduler
      ? SharedResourceHolder.get(GrpcUtil.TIMER_SERVICE) : timeoutService;
  this.socketFactory = socketFactory;
  this.hostnameVerifier = hostnameVerifier;
  this.connectionSpec = connectionSpec;
  this.maxMessageSize = maxMessageSize;
  this.enableKeepAlive = enableKeepAlive;
  this.keepAliveTimeNanos = new AtomicBackoff("keepalive time nanos", keepAliveTimeNanos);
  this.keepAliveTimeoutNanos = keepAliveTimeoutNanos;
  this.flowControlWindow = flowControlWindow;
  this.keepAliveWithoutCalls = keepAliveWithoutCalls;
  this.maxInboundMetadataSize = maxInboundMetadataSize;

  usingSharedExecutor = executor == null;
  this.transportTracerFactory =
      Preconditions.checkNotNull(transportTracerFactory, "transportTracerFactory");
  if (usingSharedExecutor) {
    // The executor was unspecified, using the shared executor.
    this.executor = SharedResourceHolder.get(SHARED_EXECUTOR);
  } else {
    this.executor = executor;
  }
}
 

@Override
public void close() {
  if (closed) {
    return;
  }
  closed = true;

  if (usingSharedScheduler) {
    SharedResourceHolder.release(GrpcUtil.TIMER_SERVICE, timeoutService);
  }

  if (usingSharedExecutor) {
    SharedResourceHolder.release(SHARED_EXECUTOR, executor);
  }
}
 

@Test
public void scheduledExecutorService_default() {
  OkHttpChannelBuilder builder = OkHttpChannelBuilder.forTarget("foo");
  ClientTransportFactory clientTransportFactory = builder.buildTransportFactory();
  assertSame(
      SharedResourceHolder.get(TIMER_SERVICE),
      clientTransportFactory.getScheduledExecutorService());

  SharedResourceHolder.release(
      TIMER_SERVICE, clientTransportFactory.getScheduledExecutorService());
  clientTransportFactory.close();
}
 

@Override
public void close() {
  if (closed) {
    return;
  }
  closed = true;

  protocolNegotiator.close();
  if (usingSharedGroup) {
    SharedResourceHolder.release(Utils.DEFAULT_WORKER_EVENT_LOOP_GROUP, group);
  }
}
 

@Override
public GoogleDefaultProtocolNegotiator buildProtocolNegotiator() {
  TsiHandshakerFactory altsHandshakerFactory =
      new TsiHandshakerFactory() {
        @Override
        public TsiHandshaker newHandshaker(String authority) {
          // Used the shared grpc channel to connecting to the ALTS handshaker service.
          // TODO: Release the channel if it is not used.
          // https://github.com/grpc/grpc-java/issues/4755.
          ManagedChannel channel =
              SharedResourceHolder.get(HandshakerServiceChannel.SHARED_HANDSHAKER_CHANNEL);
          AltsClientOptions handshakerOptions =
              new AltsClientOptions.Builder()
                  .setRpcProtocolVersions(RpcProtocolVersionsUtil.getRpcProtocolVersions())
                  .setTargetName(authority)
                  .build();
          return AltsTsiHandshaker.newClient(
              HandshakerServiceGrpc.newStub(channel), handshakerOptions);
        }
      };
  SslContext sslContext;
  try {
    sslContext = GrpcSslContexts.forClient().build();
  } catch (SSLException ex) {
    throw new RuntimeException(ex);
  }
  return negotiatorForTest =
      new GoogleDefaultProtocolNegotiator(altsHandshakerFactory, sslContext);
}
 

@Override
public final synchronized void start(Listener listener) {
  Preconditions.checkState(this.listener == null, "already started");
  timerService = SharedResourceHolder.get(timerServiceResource);
  executor = SharedResourceHolder.get(executorResource);
  this.listener = Preconditions.checkNotNull(listener, "listener");
  resolve();
}
 

private InProcessClientTransportFactory(
    String name,
    @Nullable ScheduledExecutorService scheduledExecutorService,
    int maxInboundMetadataSize) {
  this.name = name;
  useSharedTimer = scheduledExecutorService == null;
  timerService = useSharedTimer
      ? SharedResourceHolder.get(GrpcUtil.TIMER_SERVICE) : scheduledExecutorService;
  this.maxInboundMetadataSize = maxInboundMetadataSize;
}
 

@Override
public void close() {
  if (closed) {
    return;
  }
  closed = true;
  if (useSharedTimer) {
    SharedResourceHolder.release(GrpcUtil.TIMER_SERVICE, timerService);
  }
}
 

@Test
public void scheduledExecutorService_default() {
  InProcessChannelBuilder builder = InProcessChannelBuilder.forName("foo");
  ClientTransportFactory clientTransportFactory = builder.buildTransportFactory();
  assertSame(
      SharedResourceHolder.get(TIMER_SERVICE),
      clientTransportFactory.getScheduledExecutorService());

  SharedResourceHolder.release(
      TIMER_SERVICE, clientTransportFactory.getScheduledExecutorService());
  clientTransportFactory.close();
}
 

/**
 * Creates a self name resolver with the given properties.
 *
 * @param properties The properties to read the server address from.
 * @param args The arguments for the resolver.
 * @param executorResource The shared executor resource for channels.
 */
public SelfNameResolver(final GrpcServerProperties properties, final Args args,
        final SharedResourceHolder.Resource<Executor> executorResource) {
    this.properties = requireNonNull(properties, "properties");
    this.syncContext = requireNonNull(args.getSynchronizationContext(), "syncContext");
    this.executorResource = requireNonNull(executorResource, "executorResource");
    this.executor = args.getOffloadExecutor();
    this.usingExecutorResource = this.executor == null;
}
 

@Override
public final void start(final Listener2 listener) {
    checkState(this.listener == null, "already started");
    this.listener = checkNotNull(listener, "listener");
    if (this.usingExecutorResource) {
        this.executor = SharedResourceHolder.get(this.executorResource);
    }
    resolve();
}
 

@Override
public void shutdown() {
    this.listener = null;
    if (this.executor != null && this.usingExecutorResource) {
        this.executor = SharedResourceHolder.release(this.executorResource, this.executor);
    }
}
 

/**
 * Creates a new DiscoveryClientNameResolver.
 *
 * @param name The name of the service to look up.
 * @param client The client used to look up the service addresses.
 * @param args The name resolver args.
 * @param executorResource The executor resource.
 * @param externalCleaner The optional cleaner used during {@link #shutdown()}
 */
public DiscoveryClientNameResolver(final String name, final DiscoveryClient client, final Args args,
        final SharedResourceHolder.Resource<Executor> executorResource, final Runnable externalCleaner) {
    this.name = name;
    this.client = client;
    this.syncContext = requireNonNull(args.getSynchronizationContext(), "syncContext");
    this.externalCleaner = externalCleaner;
    this.executor = args.getOffloadExecutor();
    this.usingExecutorResource = this.executor == null;
    this.executorResource = executorResource;
}
 

@Override
public void start(final Listener2 listener) {
    checkState(this.listener == null, "already started");
    if (this.usingExecutorResource) {
        this.executor = SharedResourceHolder.get(this.executorResource);
    }
    this.listener = checkNotNull(listener, "listener");
    resolve();
}
 

@Override
public void shutdown() {
    this.listener = null;
    if (this.executor != null && this.usingExecutorResource) {
        this.executor = SharedResourceHolder.release(this.executorResource, this.executor);
    }
    this.instanceList = Lists.newArrayList();
    if (this.externalCleaner != null) {
        this.externalCleaner.run();
    }
}
 

/**
 * Creates a self name resolver with the given properties.
 *
 * @param properties The properties to read the server address from.
 * @param args The arguments for the resolver.
 * @param executorResource The shared executor resource for channels.
 */
public SelfNameResolver(final GrpcServerProperties properties, final Args args,
        final SharedResourceHolder.Resource<Executor> executorResource) {
    this.properties = requireNonNull(properties, "properties");
    this.syncContext = requireNonNull(args.getSynchronizationContext(), "syncContext");
    this.executorResource = requireNonNull(executorResource, "executorResource");
    this.executor = args.getOffloadExecutor();
    this.usingExecutorResource = this.executor == null;
}
 

@Override
public final void start(final Listener2 listener) {
    checkState(this.listener == null, "already started");
    this.listener = checkNotNull(listener, "listener");
    if (this.usingExecutorResource) {
        this.executor = SharedResourceHolder.get(this.executorResource);
    }
    resolve();
}
 

@Override
public void shutdown() {
    this.listener = null;
    if (this.executor != null && this.usingExecutorResource) {
        this.executor = SharedResourceHolder.release(this.executorResource, this.executor);
    }
}
 

/**
 * Creates a new DiscoveryClientNameResolver.
 *
 * @param name The name of the service to look up.
 * @param client The client used to look up the service addresses.
 * @param args The name resolver args.
 * @param executorResource The executor resource.
 * @param externalCleaner The optional cleaner used during {@link #shutdown()}
 */
public DiscoveryClientNameResolver(final String name, final DiscoveryClient client, final Args args,
        final SharedResourceHolder.Resource<Executor> executorResource, final Runnable externalCleaner) {
    this.name = name;
    this.client = client;
    this.syncContext = requireNonNull(args.getSynchronizationContext(), "syncContext");
    this.externalCleaner = externalCleaner;
    this.executor = args.getOffloadExecutor();
    this.usingExecutorResource = this.executor == null;
    this.executorResource = executorResource;
}
 

@Override
public void start(final Listener2 listener) {
    checkState(this.listener == null, "already started");
    if (this.usingExecutorResource) {
        this.executor = SharedResourceHolder.get(this.executorResource);
    }
    this.listener = checkNotNull(listener, "listener");
    resolve();
}
 

@Override
public void shutdown() {
    this.listener = null;
    if (this.executor != null && this.usingExecutorResource) {
        this.executor = SharedResourceHolder.release(this.executorResource, this.executor);
    }
    this.instanceList = Lists.newArrayList();
    if (this.externalCleaner != null) {
        this.externalCleaner.run();
    }
}
 

public KubernetesNameResolver(String namespace, String name, int port, Attributes params, SharedResourceHolder.Resource<ScheduledExecutorService> timerServiceResource, SharedResourceHolder.Resource<Executor> sharedChannelExecutorResource) {
  this.namespace = namespace;
  this.name = name;
  this.port = port;
  this.params = params;
  this.timerServiceResource = timerServiceResource;
  this.sharedChannelExecutorResource = sharedChannelExecutorResource;
  this.kubernetesClient = new DefaultKubernetesClient();
}