大纲
10.gRPC客户端初始化分析
11.gRPC客户端的心跳机制(健康检查)
12.gRPC服务端如何处理客户端的建立连接请求
13.gRPC服务端如何映射各种请求与对应的Handler处理类
14.gRPC简单介绍
10.gRPC客户端初始化分析
(1)gRPC客户端代理初始化的源码
(2)gRPC客户端启动的源码
(3)gRPC客户端发起与服务端建立连接请求的源码
(1)gRPC客户端代理初始化的源码
Nacos客户端注册服务实例时会调用NacosNamingService的registerInstance()方法，接着会调用NamingClientProxyDelegate的registerService()方法，然后判断注册的服务实例是不是临时的。如果注册的服务实例是临时的，那么就使用gRPC客户端代理去进行注册。如果注册的服务实例不是临时的，那么就使用HTTP客户端代理去进行注册。
NacosNamingService的init()方法在创建客户端代理，也就是执行NamingClientProxyDelegate的构造方法时，便会创建和初始化gRPC客户端代理NamingGrpcClientProxy。
创建和初始化gRPC客户端代理NamingGrpcClientProxy时，首先会由RpcClientFactory的createClient()方法创建一个RpcClient对象，并将GrpcClient对象赋值给NamingGrpcClientProxy的rpcClient属性，然后调用NamingGrpcClientProxy的start()方法启动RPC客户端连接。
在NamingGrpcClientProxy的start()方法中，会先注册一个用于处理服务端推送请求的NamingPushRequestHandler，然后调用RpcClient的start()方法启动RPC客户端即RpcClient对象，最后将NamingGrpcClientProxy自己作为订阅者向通知中心进行注册。
public class NacosNamingService implements NamingService {
...
private NamingClientProxy clientProxy;
private void init(Properties properties) throws NacosException {
...
this.clientProxy = new NamingClientProxyDelegate(this.namespace, serviceInfoHolder, properties, changeNotifier);
}
...
@Override
public void registerInstance(String serviceName, Instance instance) throws NacosException {
registerInstance(serviceName, Constants.DEFAULT_GROUP, instance);
}
@Override
public void registerInstance(String serviceName, String groupName, Instance instance) throws NacosException {
NamingUtils.checkInstanceIsLegal(instance);
//调用NamingClientProxy的注册方法registerService()，其实就是NamingClientProxyDelegate.registerService()方法
clientProxy.registerService(serviceName, groupName, instance);
}
...
}
//客户端代理
public class NamingClientProxyDelegate implements NamingClientProxy {
private final NamingHttpClientProxy httpClientProxy;
private final NamingGrpcClientProxy grpcClientProxy;
public NamingClientProxyDelegate(String namespace, ServiceInfoHolder serviceInfoHolder, Properties properties, InstancesChangeNotifier changeNotifier) throws NacosException {
...
//初始化HTTP客户端代理
this.httpClientProxy = new NamingHttpClientProxy(namespace, securityProxy, serverListManager, properties, serviceInfoHolder);
//初始化gRPC客户端代理
this.grpcClientProxy = new NamingGrpcClientProxy(namespace, securityProxy, serverListManager, properties, serviceInfoHolder);
}
...
@Override
public void registerService(String serviceName, String groupName, Instance instance) throws NacosException {
getExecuteClientProxy(instance).registerService(serviceName, groupName, instance);
}
private NamingClientProxy getExecuteClientProxy(Instance instance) {
return instance.isEphemeral() ? grpcClientProxy : httpClientProxy;
}
...
}
//gRPC客户端代理
public class NamingGrpcClientProxy extends AbstractNamingClientProxy {
private final String namespaceId;
private final String uuid;
private final Long requestTimeout;
private final RpcClient rpcClient;
private final NamingGrpcRedoService redoService;
//初始化gRPC客户端代理
public NamingGrpcClientProxy(String namespaceId, SecurityProxy securityProxy, ServerListFactory serverListFactory, Properties properties, ServiceInfoHolder serviceInfoHolder) throws NacosException {
super(securityProxy);
this.namespaceId = namespaceId;
this.uuid = UUID.randomUUID().toString();
this.requestTimeout = Long.parseLong(properties.getProperty(CommonParams.NAMING_REQUEST_TIMEOUT, "-1"));
Map<String, String> labels = new HashMap<String, String>();
labels.put(RemoteConstants.LABEL_SOURCE, RemoteConstants.LABEL_SOURCE_SDK);
labels.put(RemoteConstants.LABEL_MODULE, RemoteConstants.LABEL_MODULE_NAMING);
//1.通过RpcClientFactory.createClient()方法创建一个GrpcSdkClient对象实例，然后赋值给rpcClient属性
this.rpcClient = RpcClientFactory.createClient(uuid, ConnectionType.GRPC, labels);
this.redoService = new NamingGrpcRedoService(this);
//2.启动gRPC客户端代理NamingGrpcClientProxy
start(serverListFactory, serviceInfoHolder);
}
private void start(ServerListFactory serverListFactory, ServiceInfoHolder serviceInfoHolder) throws NacosException {
rpcClient.serverListFactory(serverListFactory);
//注册连接监听器
rpcClient.registerConnectionListener(redoService);
//1.注册一个用于处理服务端推送请求的NamingPushRequestHandler
rpcClient.registerServerRequestHandler(new NamingPushRequestHandler(serviceInfoHolder));
//2.启动RPC客户端RpcClient
rpcClient.start();
//3.将NamingGrpcClientProxy自己作为订阅者向通知中心进行注册
NotifyCenter.registerSubscriber(this);
}
...
@Override
public void registerService(String serviceName, String groupName, Instance instance) throws NacosException {
NAMING_LOGGER.info("[REGISTER-SERVICE] {} registering service {} with instance {}", namespaceId, serviceName, instance);
redoService.cacheInstanceForRedo(serviceName, groupName, instance);
//执行服务实例的注册
doRegisterService(serviceName, groupName, instance);
}
//Execute register operation.
public void doRegisterService(String serviceName, String groupName, Instance instance) throws NacosException {
//创建请求参数对象
InstanceRequest request = new InstanceRequest(namespaceId, serviceName, groupName, NamingRemoteConstants.REGISTER_INSTANCE, instance);
//向服务端发起请求
requestToServer(request, Response.class);
redoService.instanceRegistered(serviceName, groupName);
}
private <T extends Response> T requestToServer(AbstractNamingRequest request, Class<T> responseClass) throws NacosException {
try {
request.putAllHeader(getSecurityHeaders(request.getNamespace(), request.getGroupName(), request.getServiceName()));
//实际会调用RpcClient.request()方法发起gRPC请求
Response response = requestTimeout < 0 ? rpcClient.request(request) : rpcClient.request(request, requestTimeout);
if (ResponseCode.SUCCESS.getCode() != response.getResultCode()) {
throw new NacosException(response.getErrorCode(), response.getMessage());
}
if (responseClass.isAssignableFrom(response.getClass())) {
return (T) response;
}
NAMING_LOGGER.error("Server return unexpected response '{}', expected response should be '{}'", response.getClass().getName(), responseClass.getName());
} catch (Exception e) {
throw new NacosException(NacosException.SERVER_ERROR, "Request nacos server failed: ", e);
}
throw new NacosException(NacosException.SERVER_ERROR, "Server return invalid response");
}
...
}
public class RpcClientFactory {
private static final Map<String, RpcClient> CLIENT_MAP = new ConcurrentHashMap<>();
...
//create a rpc client.
public static RpcClient createClient(String clientName, ConnectionType connectionType, Integer threadPoolCoreSize, Integer threadPoolMaxSize, Map<String, String> labels) {
if (!ConnectionType.GRPC.equals(connectionType)) {
throw new UnsupportedOperationException("unsupported connection type :" + connectionType.getType());
}
return CLIENT_MAP.computeIfAbsent(clientName, clientNameInner -> {
LOGGER.info("[RpcClientFactory] create a new rpc client of " + clientName);
try {
//创建GrpcClient对象
GrpcClient client = new GrpcSdkClient(clientNameInner);
//设置线程核心数和最大数
client.setThreadPoolCoreSize(threadPoolCoreSize);
client.setThreadPoolMaxSize(threadPoolMaxSize);
client.labels(labels);
return client;
} catch (Throwable throwable) {
LOGGER.error("Error to init GrpcSdkClient for client name :" + clientName, throwable);
throw throwable;
}
});
}
...
}(2)gRPC客户端启动的源码
NamingGrpcClientProxy的start()方法会通过调用RpcClient的start()方法，来启动RPC客户端即RpcClient对象。
在RpcClient的start()方法中，首先会利用CAS来修改RPC客户端(RpcClient)的状态，也就是将RpcClient.rpcClientStatus属性从INITIALIZED更新为STARTING。
然后会创建一个核心线程数为2的线程池，并提交两个任务。任务一是处理连接成功或连接断开时的线程，任务二是处理重连或健康检查的线程。
接着会创建Connection连接对象，也就是在while循环中调用GrpcClient的connectToServer()方法，尝试与服务端建立连接。如果连接失败，则会抛出异常并且进行重试，由于是同步连接，所以最大重试次数是3。
最后当客户端与服务端成功建立连接后，会把对应的Connection连接对象赋值给RpcClient.currentConnection属性，并且修改RpcClient.rpcClientStatus属性即RPC客户端状态为RUNNING。
如果客户端与服务端连接失败，则会通过异步尝试进行连接，也就是调用RpcClient的switchServerAsync()方法，往RpcClient的reconnectionSignal队列中放入一个ReconnectContext对象，reconnectionSignal队列中的元素会交给任务2来处理。
public abstract class RpcClient implements Closeable {
protected volatile AtomicReference<RpcClientStatus> rpcClientStatus = new AtomicReference<>(RpcClientStatus.WAIT_INIT);
protected ScheduledExecutorService clientEventExecutor;
protected BlockingQueue<ConnectionEvent> eventLinkedBlockingQueue = new LinkedBlockingQueue<>();
//在NamingGrpcClientProxy初始化 -> 调用RpcClient.start()方法时，会将GrpcClient.connectToServer()方法的返回值赋值给currentConnection属性
protected volatile Connection currentConnection;
private final BlockingQueue<ReconnectContext> reconnectionSignal = new ArrayBlockingQueue<>(1);
...
public final void start() throws NacosException {
//利用CAS来修改RPC客户端(RpcClient)的状态，从INITIALIZED更新为STARTING
boolean success = rpcClientStatus.compareAndSet(RpcClientStatus.INITIALIZED, RpcClientStatus.STARTING);
if (!success) {
return;
}
//接下来创建调度线程池执行器，并提交两个任务
clientEventExecutor = new ScheduledThreadPoolExecutor(2, r -> {
Thread t = new Thread(r);
t.setName("com.alibaba.nacos.client.remote.worker");
t.setDaemon(true);
return t;
});
//任务1：处理连接成功或连接断开时的线程
clientEventExecutor.submit(() -> {
...
});
//任务2：处理重连或健康检查的线程
clientEventExecutor.submit(() -> {
...
});
//创建连接对象
Connection connectToServer = null;
rpcClientStatus.set(RpcClientStatus.STARTING);
//重试次数为3次
int startUpRetryTimes = RETRY_TIMES;
//在while循环中尝试与服务端建立连接，最多循环3次
while (startUpRetryTimes > 0 && connectToServer == null) {
try {
startUpRetryTimes--;
//获取服务端信息
ServerInfo serverInfo = nextRpcServer();
LoggerUtils.printIfInfoEnabled(LOGGER, "[{}] Try to connect to server on start up, server: {}", name, serverInfo);
//调用GrpcClient.connectToServer()方法建立和服务端的长连接
connectToServer = connectToServer(serverInfo);
} catch (Throwable e) {
LoggerUtils.printIfWarnEnabled(LOGGER, "[{}] Fail to connect to server on start up, error message = {}, start up retry times left: {}", name, e.getMessage(), startUpRetryTimes);
}
}
//如果连接成功，connectToServer对象就不为空
if (connectToServer != null) {
LoggerUtils.printIfInfoEnabled(LOGGER, "[{}] Success to connect to server [{}] on start up, connectionId = {}", name, connectToServer.serverInfo.getAddress(), connectToServer.getConnectionId());
//连接对象赋值，currentConnection其实就是一个在客户端使用的GrpcConnection对象实例
this.currentConnection = connectToServer;
//更改RPC客户端RpcClient的状态
rpcClientStatus.set(RpcClientStatus.RUNNING);
//往eventLinkedBlockingQueue队列放入ConnectionEvent事件
eventLinkedBlockingQueue.offer(new ConnectionEvent(ConnectionEvent.CONNECTED));
} else {
//尝试进行异步连接
switchServerAsync();
}
registerServerRequestHandler(new ConnectResetRequestHandler());
//register client detection request.
registerServerRequestHandler(request -> {
if (request instanceof ClientDetectionRequest) {
return new ClientDetectionResponse();
}
return null;
});
}
protected ServerInfo nextRpcServer() {
String serverAddress = getServerListFactory().genNextServer();
//获取服务端信息
return resolveServerInfo(serverAddress);
}
private ServerInfo resolveServerInfo(String serverAddress) {
Matcher matcher = EXCLUDE_PROTOCOL_PATTERN.matcher(serverAddress);
if (matcher.find()) {
serverAddress = matcher.group(1);
}
String[] ipPortTuple = serverAddress.split(Constants.COLON, 2);
int defaultPort = Integer.parseInt(System.getProperty("nacos.server.port", "8848"));
String serverPort = CollectionUtils.getOrDefault(ipPortTuple, 1, Integer.toString(defaultPort));
return new ServerInfo(ipPortTuple[0], NumberUtils.toInt(serverPort, defaultPort));
}
public void switchServerAsync() {
//异步注册逻辑
switchServerAsync(null, false);
}
protected void switchServerAsync(final ServerInfo recommendServerInfo, boolean onRequestFail) {
//往reconnectionSignal队列里放入一个对象
reconnectionSignal.offer(new ReconnectContext(recommendServerInfo, onRequestFail));
}
...
}(3)gRPC客户端发起与服务端建立连接请求的源码
gRPC客户端与服务端建立连接的方法是GrpcClient的connectToServer()方法。该方法首先会获取进行网络通信的端口号，因为gRPC服务需要额外占用一个端口的，所以这个端口号是在Nacos的8848基础上 + 偏移量1000，变成9848。
在建立连接之前，会先检查一下服务端，如果没问题才发起连接请求，接着就会调用GrpcConnection的sendRequest()方法发起连接请求，最后返回GrpcConnection连接对象。
public abstract class GrpcClient extends RpcClient {
...
@Override
public Connection connectToServer(ServerInfo serverInfo) {
try {
if (grpcExecutor == null) {
this.grpcExecutor = createGrpcExecutor(serverInfo.getServerIp());
}
//获取端口号：gRPC服务需要额外占用一个端口的，这个端口是在Nacos 8848的基础上，+ 偏移量1000，所以是9848
int port = serverInfo.getServerPort() + rpcPortOffset();
RequestGrpc.RequestFutureStub newChannelStubTemp = createNewChannelStub(serverInfo.getServerIp(), port);
if (newChannelStubTemp != null) {
//检查一下服务端，没问题才会发起RPC连接请求
Response response = serverCheck(serverInfo.getServerIp(), port, newChannelStubTemp);
if (response == null || !(response instanceof ServerCheckResponse)) {
shuntDownChannel((ManagedChannel) newChannelStubTemp.getChannel());
return null;
}
BiRequestStreamGrpc.BiRequestStreamStub biRequestStreamStub = BiRequestStreamGrpc.newStub(newChannelStubTemp.getChannel());
//创建连接对象
GrpcConnection grpcConn = new GrpcConnection(serverInfo, grpcExecutor);
grpcConn.setConnectionId(((ServerCheckResponse) response).getConnectionId());
//create stream request and bind connection event to this connection.
//创建流请求并将连接事件绑定到此连接
StreamObserver<Payload> payloadStreamObserver = bindRequestStream(biRequestStreamStub, grpcConn);
//stream observer to send response to server
grpcConn.setPayloadStreamObserver(payloadStreamObserver);
grpcConn.setGrpcFutureServiceStub(newChannelStubTemp);
grpcConn.setChannel((ManagedChannel) newChannelStubTemp.getChannel());
//send a setup request.
ConnectionSetupRequest conSetupRequest = new ConnectionSetupRequest();
conSetupRequest.setClientVersion(VersionUtils.getFullClientVersion());
conSetupRequest.setLabels(super.getLabels());
conSetupRequest.setAbilities(super.clientAbilities);
conSetupRequest.setTenant(super.getTenant());
//发起连接请求
grpcConn.sendRequest(conSetupRequest);
//wait to register connection setup
Thread.sleep(100L);
return grpcConn;
}
return null;
} catch (Exception e) {
LOGGER.error("[{}]Fail to connect to server!,error={}", GrpcClient.this.getName(), e);
}
return null;
}
private Response serverCheck(String ip, int port, RequestGrpc.RequestFutureStub requestBlockingStub) {
try {
if (requestBlockingStub == null) {
return null;
}
ServerCheckRequest serverCheckRequest = new ServerCheckRequest();
Payload grpcRequest = GrpcUtils.convert(serverCheckRequest);
//向服务端发送一个检查请求
ListenableFuture<Payload> responseFuture = requestBlockingStub.request(grpcRequest);
Payload response = responseFuture.get(3000L, TimeUnit.MILLISECONDS);
//receive connection unregister response here,not check response is success.
return (Response) GrpcUtils.parse(response);
} catch (Exception e) {
LoggerUtils.printIfErrorEnabled(LOGGER, "Server check fail, please check server {} ,port {} is available , error ={}", ip, port, e);
return null;
}
}
private StreamObserver<Payload> bindRequestStream(final BiRequestStreamGrpc.BiRequestStreamStub streamStub, final GrpcConnection grpcConn) {
//调用BiRequestStreamStub.requestBiStream()方法连接服务端
return streamStub.requestBiStream(new StreamObserver<Payload>() {
@Override
public void onNext(Payload payload) {
LoggerUtils.printIfDebugEnabled(LOGGER, "[{}]Stream server request receive, original info: {}", grpcConn.getConnectionId(), payload.toString());
try {
Object parseBody = GrpcUtils.parse(payload);
final Request request = (Request) parseBody;
if (request != null) {
try {
Response response = handleServerRequest(request);
if (response != null) {
response.setRequestId(request.getRequestId());
sendResponse(response);
} else {
LOGGER.warn("[{}]Fail to process server request, ackId->{}", grpcConn.getConnectionId(), request.getRequestId());
}
} catch (Exception e) {
LoggerUtils.printIfErrorEnabled(LOGGER, "[{}]Handle server request exception: {}", grpcConn.getConnectionId(), payload.toString(), e.getMessage());
Response errResponse = ErrorResponse.build(NacosException.CLIENT_ERROR, "Handle server request error");
errResponse.setRequestId(request.getRequestId());
sendResponse(errResponse);
}
}
} catch (Exception e) {
LoggerUtils.printIfErrorEnabled(LOGGER, "[{}]Error to process server push response: {}", grpcConn.getConnectionId(), payload.getBody().getValue().toStringUtf8());
}
}
@Override
public void onError(Throwable throwable) {
boolean isRunning = isRunning();
boolean isAbandon = grpcConn.isAbandon();
if (isRunning && !isAbandon) {
LoggerUtils.printIfErrorEnabled(LOGGER, "[{}]Request stream error, switch server,error={}", grpcConn.getConnectionId(), throwable);
if (rpcClientStatus.compareAndSet(RpcClientStatus.RUNNING, RpcClientStatus.UNHEALTHY)) {
switchServerAsync();
}
} else {
LoggerUtils.printIfWarnEnabled(LOGGER, "[{}]Ignore error event,isRunning:{},isAbandon={}", grpcConn.getConnectionId(), isRunning, isAbandon);
}
}
@Override
public void onCompleted() {
boolean isRunning = isRunning();
boolean isAbandon = grpcConn.isAbandon();
if (isRunning && !isAbandon) {
LoggerUtils.printIfErrorEnabled(LOGGER, "[{}]Request stream onCompleted, switch server", grpcConn.getConnectionId());
if (rpcClientStatus.compareAndSet(RpcClientStatus.RUNNING, RpcClientStatus.UNHEALTHY)) {
switchServerAsync();
}
} else {
LoggerUtils.printIfInfoEnabled(LOGGER, "[{}]Ignore complete event,isRunning:{},isAbandon={}", grpcConn.getConnectionId(), isRunning, isAbandon);
}
}
});
}
...
}(4)总结
11.gRPC客户端的心跳机制(健康检查)
(1)线程任务一：处理连接成功或连接断开时的通知
(2)线程任务二：处理重连或健康检查
RpcClient的start()方法会调用GrpcClient的connectToServer()方法连接服务端，不管连接是否成功，最后都会往不同的阻塞队列中添加事件。
如果连接成功，那么就往RpcClient的eventLinkedBlockingQueue添加连接事件。如果连接失败，那么就往RpcClient的reconnectionSignal队列添加重连对象。而这两个阻塞队列中的数据处理，便是由执行RpcClient的start()方法时启动的两个线程任务进行处理的。
(1)线程任务一：处理连接成功或连接断开时的通知
这个任务主要在连接成功或者连接断开时，修改一些属性状态。通过eventLinkedBlockingQueue的take()方法从队列取到连接事件后，会判断连接事件是否建立连接还是断开连接。
如果是建立连接，那么就调用RpcClient的notifyConnected()方法，把执行NamingGrpcClientProxy的start()方法时所注册的NamingGrpcRedoService对象的connected属性设置为true。
如果是断开连接，那么就调用RpcClient的notifyDisConnected()方法，把执行NamingGrpcClientProxy的start()方法时所注册的NamingGrpcRedoService对象的connected属性设置为false。
public abstract class RpcClient implements Closeable {
protected volatile AtomicReference<RpcClientStatus> rpcClientStatus = new AtomicReference<>(RpcClientStatus.WAIT_INIT);
protected ScheduledExecutorService clientEventExecutor;
protected BlockingQueue<ConnectionEvent> eventLinkedBlockingQueue = new LinkedBlockingQueue<>();
private final BlockingQueue<ReconnectContext> reconnectionSignal = new ArrayBlockingQueue<>(1);
//listener called where connection's status changed. 连接状态改变的监听器
protected List<ConnectionEventListener> connectionEventListeners = new ArrayList<>();
...
public final void start() throws NacosException {
//利用CAS来修改RPC客户端(RpcClient)的状态，从INITIALIZED更新为STARTING
boolean success = rpcClientStatus.compareAndSet(RpcClientStatus.INITIALIZED, RpcClientStatus.STARTING);
if (!success) {
return;
}
//接下来创建调度线程池执行器，并提交两个任务
clientEventExecutor = new ScheduledThreadPoolExecutor(2, r -> {
Thread t = new Thread(r);
t.setName("com.alibaba.nacos.client.remote.worker");
t.setDaemon(true);
return t;
});
//任务1：处理连接成功或连接断开时的线程
clientEventExecutor.submit(() -> {
while (!clientEventExecutor.isTerminated() && !clientEventExecutor.isShutdown()) {
ConnectionEvent take;
try {
take = eventLinkedBlockingQueue.take();
if (take.isConnected()) {
notifyConnected();
} else if (take.isDisConnected()) {
notifyDisConnected();
}
} catch (Throwable e) {
// Do nothing
}
}
});
//任务2：向服务端上报心跳或重连的线程
clientEventExecutor.submit(() -> {
...
});
}
...
//Notify when client new connected.
protected void notifyConnected() {
if (connectionEventListeners.isEmpty()) {
return;
}
LoggerUtils.printIfInfoEnabled(LOGGER, "[{}] Notify connected event to listeners.", name);
for (ConnectionEventListener connectionEventListener : connectionEventListeners) {
try {
connectionEventListener.onConnected();
} catch (Throwable throwable) {
LoggerUtils.printIfErrorEnabled(LOGGER, "[{}] Notify connect listener error, listener = {}", name, connectionEventListener.getClass().getName());
}
}
}
//Notify when client disconnected.
protected void notifyDisConnected() {
if (connectionEventListeners.isEmpty()) {
return;
}
LoggerUtils.printIfInfoEnabled(LOGGER, "[{}] Notify disconnected event to listeners", name);
for (ConnectionEventListener connectionEventListener : connectionEventListeners) {
try {
connectionEventListener.onDisConnect();
} catch (Throwable throwable) {
LoggerUtils.printIfErrorEnabled(LOGGER, "[{}] Notify disconnect listener error, listener = {}", name, connectionEventListener.getClass().getName());
}
}
}
...
//Register connection handler. Will be notified when inner connection's state changed.
//在执行NamingGrpcClientProxy.start()方法时会将NamingGrpcRedoService对象注册到connectionEventListeners中
public synchronized void registerConnectionListener(ConnectionEventListener connectionEventListener) {
LoggerUtils.printIfInfoEnabled(LOGGER, "[{}] Registry connection listener to current client:{}", name, connectionEventListener.getClass().getName());
this.connectionEventListeners.add(connectionEventListener);
}
...
}
public class NamingGrpcRedoService implements ConnectionEventListener {
private volatile boolean connected = false;
...
@Override
public void onConnected() {
connected = true;
LogUtils.NAMING_LOGGER.info("Grpc connection connect");
}
@Override
public void onDisConnect() {
connected = false;
LogUtils.NAMING_LOGGER.warn("Grpc connection disconnect, mark to redo");
synchronized (registeredInstances) {
registeredInstances.values().forEach(instanceRedoData -> instanceRedoData.setRegistered(false));
}
synchronized (subscribes) {
subscribes.values().forEach(subscriberRedoData -> subscriberRedoData.setRegistered(false));
}
LogUtils.NAMING_LOGGER.warn("mark to redo completed");
}
...
}(2)线程任务二：处理重连或健康检查
如果RpcClient的start()方法在调用GrpcClient的connectToServer()方法连接服务端时失败了，那么会往RpcClient.reconnectionSignal队列添加重连对象的，而这个任务就会获取reconnectionSignal队列中的重连对象进行重连。
因为reconnectionSignal中的数据是当连接失败时放入的，所以如果从reconnectionSignal中获取不到重连对象，等同于连接成功。
注意：这个任务从reconnectionSignal阻塞队列中获取重连对象时，调用的是阻塞队列的take()方法，而不是阻塞队列的poll()方法。BlockingQueue的take()方法，如果读取不到数据，会一直处于阻塞状态。BlockingQueue的poll()方法，在指定的时间内读取不到数据，会返回null。
情况一：如果从reconnectionSignal队列中获取到的重连对象为null
首先判断存活时间是否大于 5s，如果大于则调用RpcClient.healthCheck()方法发起健康检查的RPC请求。健康检查的触发方法是currentConnection.request()方法，健康检查的请求类型是HealthCheckRequest。
如果健康检查成功，只需刷新存活时间即可。如果健康检查失败，则需要尝试与服务端重新建立连接。
情况二：如果从reconnectionSignal队列中获取到的重连对象不为null
那么就调用RpcClient的reconnect()方法进行重新连接，该方法会通过GrpcClient的connectToServer()方法尝试与服务端建立连接。
public abstract class RpcClient implements Closeable {
protected volatile AtomicReference<RpcClientStatus> rpcClientStatus = new AtomicReference<>(RpcClientStatus.WAIT_INIT);
protected ScheduledExecutorService clientEventExecutor;
protected BlockingQueue<ConnectionEvent> eventLinkedBlockingQueue = new LinkedBlockingQueue<>();
private final BlockingQueue<ReconnectContext> reconnectionSignal = new ArrayBlockingQueue<>(1);
...
public final void start() throws NacosException {
//利用CAS来修改RPC客户端(RpcClient)的状态，从INITIALIZED更新为STARTING
boolean success = rpcClientStatus.compareAndSet(RpcClientStatus.INITIALIZED, RpcClientStatus.STARTING);
if (!success) {
return;
}
//接下来创建调度线程池执行器，并提交两个任务
clientEventExecutor = new ScheduledThreadPoolExecutor(2, r -> {
Thread t = new Thread(r);
t.setName("com.alibaba.nacos.client.remote.worker");
t.setDaemon(true);
return t;
});
//任务1：处理连接成功或连接断开时的线程
clientEventExecutor.submit(() -> {
...
});
//任务2：向服务端上报心跳或重连的线程
clientEventExecutor.submit(() -> {
while (true) {
try {
if (isShutdown()) {
break;
}
//这里从reconnectionSignal阻塞队列中获取任务不是调用take()方法，而是调用poll()方法，并且指定了5s的最大读取时间
//BlockingQueue的take()方法，如果读取不到数据，会一直处于阻塞状态
//BlockingQueue的poll()方法，在指定的时间内读取不到数据，会返回null
ReconnectContext reconnectContext = reconnectionSignal.poll(keepAliveTime, TimeUnit.MILLISECONDS);
//reconnectContext为null，说明从reconnectionSignal中获取不到数据
//由于reconnectionSignal中的数据是当连接失败时放入的
//所以从reconnectionSignal中获取不到数据，等同于连接成功
if (reconnectContext == null) {
//check alive time.
//检查存活时间，默认存活时间为5s，超过5s就需要做健康检查
if (System.currentTimeMillis() - lastActiveTimeStamp >= keepAliveTime) {
//调用RpcClient.healthCheck()方法，发起健康检查请求
boolean isHealthy = healthCheck();
//如果向服务端发起健康检查请求失败，则需要尝试重新建立连接
if (!isHealthy) {
if (currentConnection == null) {
continue;
}
LoggerUtils.printIfInfoEnabled(LOGGER, "[{}] Server healthy check fail, currentConnection = {}", name, currentConnection.getConnectionId());
//判断连接状态是否关闭，如果是则结束异步任务
RpcClientStatus rpcClientStatus = RpcClient.this.rpcClientStatus.get();
if (RpcClientStatus.SHUTDOWN.equals(rpcClientStatus)) {
break;
}
//修改RpcClient的连接状态为不健康
boolean statusFLowSuccess = RpcClient.this.rpcClientStatus.compareAndSet(rpcClientStatus, RpcClientStatus.UNHEALTHY);
//给reconnectContext属性赋值，准备尝试重连
if (statusFLowSuccess) {
//重新赋值，注意这里没有continue，所以逻辑会接着往下执行
reconnectContext = new ReconnectContext(null, false);
} else {
continue;
}
} else {
//如果向服务端发起健康检查请求成功，则刷新RpcClient的存活时间
lastActiveTimeStamp = System.currentTimeMillis();
continue;
}
} else {
continue;
}
}
if (reconnectContext.serverInfo != null) {
//clear recommend server if server is not in server list.
//如果服务器不在服务器列表中，则清除推荐服务器，即设置reconnectContext.serverInfo为null
boolean serverExist = false;
//遍历服务端列表
for (String server : getServerListFactory().getServerList()) {
ServerInfo serverInfo = resolveServerInfo(server);
if (serverInfo.getServerIp().equals(reconnectContext.serverInfo.getServerIp())) {
serverExist = true;
reconnectContext.serverInfo.serverPort = serverInfo.serverPort;
break;
}
}
//reconnectContext.serverInfo不存在服务端列表中，就清除服务器信息，设置reconnectContext.serverInfo为null
if (!serverExist) {
LoggerUtils.printIfInfoEnabled(LOGGER, "[{}] Recommend server is not in server list, ignore recommend server {}", name, reconnectContext.serverInfo.getAddress());
reconnectContext.serverInfo = null;
}
}
//进行重新连接，RpcClient.reconnect()方法中会调用GrpcClient.connectToServer()方法尝试与服务端建立连接
reconnect(reconnectContext.serverInfo, reconnectContext.onRequestFail);
} catch (Throwable throwable) {
//Do nothing
}
}
});
}
private boolean healthCheck() {
HealthCheckRequest healthCheckRequest = new HealthCheckRequest();
if (this.currentConnection == null) {
return false;
}
try {
//利用currentConnection连接对象，发起RPC请求，请求类型是HealthCheckRequest
Response response = this.currentConnection.request(healthCheckRequest, 3000L);
//not only check server is ok, also check connection is register.
return response != null && response.isSuccess();
} catch (NacosException e) {
//ignore
}
return false;
}
...
}(3)总结
