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.
@PostConstruct
public 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客户端注销事件。
@Servicepublic class ConnectionManager extends Subscriber { private Map connectionForClientIp = new ConcurrentHashMap(16); Map connections = new ConcurrentHashMap(); @Autowired private ClientConnectionEventListenerRegistry clientConnectionEventListenerRegistry; ... //unregister a connection . public synchronized void unregister(String connectionId) { //移除客户端信息 Connection remove = this.connections.remove(connectionId); if (remove != null) { String clientIp = remove.getMetaInfo().clientIp; AtomicInteger atomicInteger = connectionForClientIp.get(clientIp); if (atomicInteger != null) { int count = atomicInteger.decrementAndGet(); if (count
页:
[1]