大綱

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客戶端注銷事件。