Nacos源码—8.Nacos gRPC升级深入分析与解读第三部分
大纲
7.服务端对服务实例进行健康检查
8.服务下线如何注销注册表和客户端等信息
9.事件驱动架构源码分析
7.服务端对服务实例进行健康检查
(1)服务端对服务实例进行健康检查的设计逻辑
(2)服务端对服务实例进行健康检查的源码
(3)服务端检查服务实例不健康后的注销处理
(1)服务端对服务实例进行健康检查的设计逻辑
一.首先会获取所有客户端的Connection连接对象
Connection连接对象里有个属性叫lastActiveTime，表示的是最后存活时间。
二.然后判断当前时间-最后存活时间是否大于20s
如果大于，则把该Connection连接对象的connectionId放入到一个集合里。这个集合是一个名为outDatedConnections的待移除集合Set，此时该Connection连接对象并不会马上删除。
三.当判断完全部的Connection连接对象后会遍历outDatedConnections集合
向遍历到的Connection连接对象发起一次请求，确认是否真的下线。如果响应成功，则往successConnections集合中添加connectionId，并且刷新Connection连接对象的lastActiveTime属性。这个机制有一个专业的名称叫做：探活机制。
四.遍历待移除集合进行注销并且在注销之前先判断一下是否探活成功
也就是connectionId存在于待移除集合outDatedConnections中，但是不存在于探活成功集合successConnections中，那么这个connectionId对应的客户端就会被注销掉。
(2)服务端对服务实例进行健康检查的源码
对服务实例进行健康检查的源码入口是ConnectionManager的start()方法。
@Service
public class ConnectionManager extends Subscriber<ConnectionLimitRuleChangeEvent> {
Map<String, Connection> connections = new ConcurrentHashMap<>();
…
//Start Task：Expel the connection which active Time expire.@PostConstructpublic void start() { //Start UnHealthy Connection Expel Task. RpcScheduledExecutor.COMMON_SERVER_EXECUTOR.scheduleWithFixedDelay(new Runnable() { @Override public void run() { ... //一.首先获取所有的连接 Set<Map.Entry<String, Connection>> entries = connections.entrySet(); ... //二.然后判断客户端是否超过20s没有发来心跳信息了，如果是则会将clientId加入outDatedConnections集合中 Set<String> outDatedConnections = new HashSet<>(); long now = System.currentTimeMillis(); for (Map.Entry<String, Connection> entry : entries) { Connection client = entry.getValue(); String clientIp = client.getMetaInfo().getClientIp(); AtomicInteger integer = expelForIp.get(clientIp); if (integer != null && integer.intValue() > 0) { integer.decrementAndGet(); expelClient.add(client.getMetaInfo().getConnectionId()); expelCount--; } else if (now - client.getMetaInfo().getLastActiveTime() >= KEEP_ALIVE_TIME) {//判断心跳时间 //添加到待移除列表 outDatedConnections.add(client.getMetaInfo().getConnectionId()); } } ... //client active detection. //三.初次检测完超过20s的Connection连接对象后，并不会立马进行删除，而是进行探活，服务端主动请求客户端，来确认是否真的下线 Loggers.REMOTE_DIGEST.info("Out dated connection ,size={}", outDatedConnections.size()); if (CollectionUtils.isNotEmpty(outDatedConnections)) { Set<String> successConnections = new HashSet<>(); final CountDownLatch latch = new CountDownLatch(outDatedConnections.size()); //遍历超过20s没有心跳的客户端clientId for (String outDateConnectionId : outDatedConnections) { try { Connection connection = getConnection(outDateConnectionId); if (connection != null) { ClientDetectionRequest clientDetectionRequest = new ClientDetectionRequest(); //调用GrpcConnection.asyncRequest()方法异步发送请求 connection.asyncRequest(clientDetectionRequest, new RequestCallBack() { @Override public Executor getExecutor() { return null; } @Override public long getTimeout() { return 1000L; } @Override public void onResponse(Response response) { latch.countDown(); if (response != null && response.isSuccess()) { //响应成功刷新心跳时间 connection.freshActiveTime(); //并且加入到探活成功的集合列表中 successConnections.add(outDateConnectionId); } } @Override public void onException(Throwable e) { latch.countDown(); } }); Loggers.REMOTE_DIGEST.info("[{}]send connection active request ", outDateConnectionId); } else { latch.countDown(); } } catch (ConnectionAlreadyClosedException e) { latch.countDown(); } catch (Exception e) { Loggers.REMOTE_DIGEST.error("[{}]Error occurs when check client active detection ,error={}", outDateConnectionId, e); latch.countDown(); } } latch.await(3000L, TimeUnit.MILLISECONDS); Loggers.REMOTE_DIGEST.info("Out dated connection check successCount={}", successConnections.size()); //经过探活还是不成功的Connection连接对象，就准备进行移除了 //遍历20s没有心跳的客户端，准备移除客户端信息 for (String outDateConnectionId : outDatedConnections) { //判断探活是否成功，如果成功了则不需要移除 if (!successConnections.contains(outDateConnectionId)) { Loggers.REMOTE_DIGEST.info("[{}]Unregister Out dated connection....", outDateConnectionId); //执行客户端注销逻辑 unregister(outDateConnectionId); } } } ... } }, 1000L, 3000L, TimeUnit.MILLISECONDS);}...}
(3)服务端检查服务实例不健康后的注销处理
进行注销处理的方法是ConnectionManager的unregister()方法。该方法主要会移除Connection连接对象 + 清除一些数据，以及发布一个ClientDisconnectEvent客户端注销事件。
