對于非臨時實例（ephemeral=false)，Nacos會采用主動的健康檢測，定時向實例發送請求，根據響應來判斷實例健康狀態。

入口在ServiceManager類中的registerInstance方法：

?創建空服務時：

public void createEmptyService(String namespaceId, String serviceName, boolean local) throws NacosException {// 如果服務不存在，創建新的服務createServiceIfAbsent(namespaceId, serviceName, local, null); }

創建服務流程：

public void createServiceIfAbsent(String namespaceId, String serviceName, boolean local, Cluster cluster)throws NacosException {// 嘗試獲取服務Service service = getService(namespaceId, serviceName);if (service == null) {// 發現服務不存在，開始創建新服務Loggers.SRV_LOG.info("creating empty service {}:{}", namespaceId, serviceName);service = new Service();service.setName(serviceName);service.setNamespaceId(namespaceId);service.setGroupName(NamingUtils.getGroupName(serviceName));// now validate the service. if failed, exception will be thrownservice.setLastModifiedMillis(System.currentTimeMillis());service.recalculateChecksum();if (cluster != null) {cluster.setService(service);service.getClusterMap().put(cluster.getName(), cluster);}service.validate();// ** 寫入注冊表并初始化 **putServiceAndInit(service);if (!local) {addOrReplaceService(service);}} }

關鍵在putServiceAndInit(service)方法中：

private void putServiceAndInit(Service service) throws NacosException {// 將服務寫入注冊表putService(service);service = getService(service.getNamespaceId(), service.getName());// 完成服務的初始化service.init();consistencyService.listen(KeyBuilder.buildInstanceListKey(service.getNamespaceId(), service.getName(), true), service);consistencyService.listen(KeyBuilder.buildInstanceListKey(service.getNamespaceId(), service.getName(), false), service);Loggers.SRV_LOG.info("[NEW-SERVICE] {}", service.toJson()); }

進入初始化邏輯：service.init()，這個會進入Service類中：

/*** Init service.*/ public void init() {// 開啟臨時實例的心跳監測任務HealthCheckReactor.scheduleCheck(clientBeatCheckTask);// 遍歷注冊表中的集群for (Map.Entry<String, Cluster> entry : clusterMap.entrySet()) {entry.getValue().setService(this);// 完成集群初識化entry.getValue().init();} }

這里集群的初始化entry.getValue().init();會進入Cluster類型的init()方法：

/*** Init cluster.*/ public void init() {if (inited) {return;}// 創建健康檢測的任務checkTask = new HealthCheckTask(this);// 這里會開啟對 非臨時實例的 定時健康檢測HealthCheckReactor.scheduleCheck(checkTask);inited = true; }

這里的HealthCheckReactor.scheduleCheck(checkTask);會開啟定時任務，對非臨時實例做健康檢測。檢測邏輯定義在HealthCheckTask這個類中，是一個Runnable，其中的run方法：

public void run() {try {if (distroMapper.responsible(cluster.getService().getName()) && switchDomain.isHealthCheckEnabled(cluster.getService().getName())) {// 開始健康檢測healthCheckProcessor.process(this);// 記錄日志 。。。}} catch (Throwable e) {// 記錄日志 。。。} finally {if (!cancelled) {// 結束后，再次進行任務調度，一定延遲后執行HealthCheckReactor.scheduleCheck(this);// 。。。}} }

健康檢測邏輯定義在healthCheckProcessor.process(this);方法中，在HealthCheckProcessor接口中，這個接口也有很多實現，默認是TcpSuperSenseProcessor：

進入TcpSuperSenseProcessor的process方法：

@Override public void process(HealthCheckTask task) {// 獲取所有 非臨時實例的 集合List<Instance> ips = task.getCluster().allIPs(false);if (CollectionUtils.isEmpty(ips)) {return;}for (Instance ip : ips) {// 封裝健康檢測信息到 BeatBeat beat = new Beat(ip, task);// 放入一個阻塞隊列中taskQueue.add(beat);MetricsMonitor.getTcpHealthCheckMonitor().incrementAndGet();} }

可以看到，所有的健康檢測任務都被放入一個阻塞隊列，而不是立即執行了。這里又采用了異步執行的策略，可以看到Nacos中大量這樣的設計。

而TcpSuperSenseProcessor本身就是一個Runnable，在它的構造函數中會把自己放入線程池中去執行，其run方法如下：

public void run() {while (true) {try {// 處理任務processTask();// ...} catch (Throwable e) {SRV_LOG.error("[HEALTH-CHECK] error while processing NIO task", e);}} }

通過processTask來處理健康檢測的任務：

private void processTask() throws Exception {// 將任務封裝為一個 TaskProcessor，并放入集合Collection<Callable<Void>> tasks = new LinkedList<>();do {Beat beat = taskQueue.poll(CONNECT_TIMEOUT_MS / 2, TimeUnit.MILLISECONDS);if (beat == null) {return;}tasks.add(new TaskProcessor(beat));} while (taskQueue.size() > 0 && tasks.size() < NIO_THREAD_COUNT * 64);// 批量處理集合中的任務for (Future<?> f : GlobalExecutor.invokeAllTcpSuperSenseTask(tasks)) {f.get();} }

任務被封裝到了TaskProcessor中去執行了，TaskProcessor是一個Callable，其中的call方法：

@Override public Void call() {// 獲取檢測任務已經等待的時長long waited = System.currentTimeMillis() - beat.getStartTime();if (waited > MAX_WAIT_TIME_MILLISECONDS) {Loggers.SRV_LOG.warn("beat task waited too long: " + waited + "ms");}SocketChannel channel = null;try {// 獲取實例信息Instance instance = beat.getIp();// 通過NIO建立TCP連接channel = SocketChannel.open();channel.configureBlocking(false);// only by setting this can we make the socket close event asynchronouschannel.socket().setSoLinger(false, -1);channel.socket().setReuseAddress(true);channel.socket().setKeepAlive(true);channel.socket().setTcpNoDelay(true);Cluster cluster = beat.getTask().getCluster();int port = cluster.isUseIPPort4Check() ? instance.getPort() : cluster.getDefCkport();channel.connect(new InetSocketAddress(instance.getIp(), port));// 注冊連接、讀取事件SelectionKey key = channel.register(selector, SelectionKey.OP_CONNECT | SelectionKey.OP_READ);key.attach(beat);keyMap.put(beat.toString(), new BeatKey(key));beat.setStartTime(System.currentTimeMillis());GlobalExecutor.scheduleTcpSuperSenseTask(new TimeOutTask(key), CONNECT_TIMEOUT_MS, TimeUnit.MILLISECONDS);} catch (Exception e) {beat.finishCheck(false, false, switchDomain.getTcpHealthParams().getMax(),"tcp:error:" + e.getMessage());if (channel != null) {try {channel.close();} catch (Exception ignore) {}}}return null; }

Nacos的健康檢測有兩種模式：

• 臨時實例：

  • 采用客戶端心跳檢測模式，心跳周期5秒

  • 心跳間隔超過15秒則標記為不健康

  • 心跳間隔超過30秒則從服務列表刪除

• 永久實例：

  • 采用服務端主動健康檢測方式

  • 周期為2000 + 5000毫秒內的隨機數

  • 檢測異常只會標記為不健康，不會刪除

那么為什么Nacos有臨時和永久兩種實例呢？

以淘寶為例，雙十一大促期間，流量會比平常高出很多，此時服務肯定需要增加更多實例來應對高并發，而這些實例在雙十一之后就無需繼續使用了，采用臨時實例比較合適。而對于服務的一些常備實例，則使用永久實例更合適。

與eureka相比，Nacos與Eureka在臨時實例上都是基于心跳模式實現，差別不大，主要是心跳周期不同，eureka是30秒，Nacos是5秒。

另外，Nacos支持永久實例，而Eureka不支持，Eureka只提供了心跳模式的健康監測，而沒有主動檢測功能。

總結

以上是生活随笔為你收集整理的Nacos源码主动健康检测的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。