Nacos源碼—8.Nacos升級gRPC分析三
大綱
7.服務(wù)端對服務(wù)實(shí)例進(jìn)行健康檢查
8.服務(wù)下線如何注銷注冊表和客戶端等信息
9.事件驅(qū)動架構(gòu)源碼分析
7.服務(wù)端對服務(wù)實(shí)例進(jìn)行健康檢查
(1)服務(wù)端對服務(wù)實(shí)例進(jìn)行健康檢查的設(shè)計(jì)邏輯
(2)服務(wù)端對服務(wù)實(shí)例進(jìn)行健康檢查的源碼
(3)服務(wù)端檢查服務(wù)實(shí)例不健康后的注銷處理
(1)服務(wù)端對服務(wù)實(shí)例進(jìn)行健康檢查的設(shè)計(jì)邏輯
一.首先會獲取所有客戶端的Connection連接對象
Connection連接對象里有個屬性叫l(wèi)astActiveTime，表示的是最后存活時間。
二.然后判斷當(dāng)前時間-最后存活時間是否大于20s
如果大于，則把該Connection連接對象的connectionId放入到一個集合里。這個集合是一個名為outDatedConnections的待移除集合Set，此時該Connection連接對象并不會馬上刪除。
三.當(dāng)判斷完全部的Connection連接對象后會遍歷outDatedConnections集合
向遍歷到的Connection連接對象發(fā)起一次請求，確認(rèn)是否真的下線。如果響應(yīng)成功，則往successConnections集合中添加connectionId，并且刷新Connection連接對象的lastActiveTime屬性。這個機(jī)制有一個專業(yè)的名稱叫做：探活機(jī)制。
四.遍歷待移除集合進(jìn)行注銷并且在注銷之前先判斷一下是否探活成功
也就是connectionId存在于待移除集合outDatedConnections中，但是不存在于探活成功集合successConnections中，那么這個connectionId對應(yīng)的客戶端就會被注銷掉。
(2)服務(wù)端對服務(wù)實(shí)例進(jìn)行健康檢查的源碼
對服務(wù)實(shí)例進(jìn)行健康檢查的源碼入口是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沒有發(fā)來心跳信息了，如果是則會將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連接對象后，并不會立馬進(jìn)行刪除，而是進(jìn)行探活，服務(wù)端主動請求客戶端，來確認(rèn)是否真的下線
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();
//調(diào)用GrpcConnection.asyncRequest()方法異步發(fā)送請求
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()) {
//響應(yīng)成功刷新心跳時間
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());
//經(jīng)過探活還是不成功的Connection連接對象，就準(zhǔn)備進(jìn)行移除了
//遍歷20s沒有心跳的客戶端，準(zhǔn)備移除客戶端信息
for (String outDateConnectionId : outDatedConnections) {
//判斷探活是否成功，如果成功了則不需要移除
if (!successConnections.contains(outDateConnectionId)) {
Loggers.REMOTE_DIGEST.info("[{}]Unregister Out dated connection....", outDateConnectionId);
//執(zhí)行客戶端注銷邏輯
unregister(outDateConnectionId);
}
}
}
...
}
}, 1000L, 3000L, TimeUnit.MILLISECONDS);
}
...
}
(3)服務(wù)端檢查服務(wù)實(shí)例不健康后的注銷處理
進(jìn)行注銷處理的方法是ConnectionManager的unregister()方法。該方法主要會移除Connection連接對象 + 清除一些數(shù)據(jù)，以及發(fā)布一個ClientDisconnectEvent客戶端注銷事件。
