2. 程式人生 > >Nacos服務心跳和健康檢查原始碼介紹

Nacos服務心跳和健康檢查原始碼介紹

阿新 • • 發佈：2021-02-03

服務心跳

Nacos Client會維護一個定時任務通過持續呼叫服務端的介面更新心跳時間，保證自己處於存活狀態，防止服務端將服務剔除，Nacos預設5秒向服務端傳送一次，通過請求服務端介面/instance/beat傳送心跳。
客戶端服務在註冊服務的時候會增加一個心跳的任務，如下圖所示:


首先看下BeatInfo這個類，重點看標註的欄位，該欄位是給週期任務設定時間，如下圖:

該方法內部定義的一個DEFAULT_HEART_BEAT_INTERVAL的常量，設定5秒:

接下來我們看下addBeatInfo方法，該方法內部主要是將BeatTask任務加入到執行緒池ScheduledExecutorService當中，如下圖:

重點部分就是看BeatTask，BeatTask繼承Runnable，run方法就是我們的重點，該方法呼叫了NamingProxy的sendBeat方法，服務端請求地址為/instance/beat的方法


接下來我們把目光放到服務端，找到InstanceController的beat方法，如果是引數beat資訊的話，說明是第一次發起心跳，則會帶有服務例項資訊，因為發起心跳成功則服務端會返回下次不要帶beat資訊的引數，這樣客戶端第二次就不會攜帶beat資訊了。如果發現沒有該服務，又沒帶beat資訊，說明這個服務可能被移除過了，直接返回沒找到。如果沒有服務，但是發現有beat資訊，那就從beat中獲取服務例項資訊，進行註冊，整體執行流程如下圖: 
@CanDistro
@PutMapping("/beat")
@Secured(parser = NamingResourceParser.class, action = ActionTypes.WRITE)
public ObjectNode beat(HttpServletRequest request) throws Exception {

    ObjectNode result = JacksonUtils.createEmptyJsonNode();
    //設定心跳間隔
    result.put(SwitchEntry.CLIENT_BEAT_INTERVAL, switchDomain.getClientBeatInterval());

    String beat = WebUtils.optional(request, "beat", StringUtils.EMPTY);
    RsInfo clientBeat = null;
    //判斷有無心跳內容
    //如果存在心跳內容則不是輕量級心跳就轉化為RsInfo
    if (StringUtils.isNotBlank(beat)) {
        clientBeat = JacksonUtils.toObj(beat, RsInfo.class);
    }
    String clusterName = WebUtils
            .optional(request, CommonParams.CLUSTER_NAME, UtilsAndCommons.DEFAULT_CLUSTER_NAME);
    String ip = WebUtils.optional(request, "ip", StringUtils.EMPTY);
    int port = Integer.parseInt(WebUtils.optional(request, "port", "0"));
    if (clientBeat != null) {
        if (StringUtils.isNotBlank(clientBeat.getCluster())) {
            clusterName = clientBeat.getCluster();
        } else {
            // fix #2533
            clientBeat.setCluster(clusterName);
        }
        ip = clientBeat.getIp();
        port = clientBeat.getPort();
    }
    String namespaceId = WebUtils.optional(request, CommonParams.NAMESPACE_ID, Constants.DEFAULT_NAMESPACE_ID);
    String serviceName = WebUtils.required(request, CommonParams.SERVICE_NAME);
    NamingUtils.checkServiceNameFormat(serviceName);
    Loggers.SRV_LOG.debug("[CLIENT-BEAT] full arguments: beat: {}, serviceName: {}", clientBeat, serviceName);
    //獲取例項的資訊
    Instance instance = serviceManager.getInstance(namespaceId, serviceName, clusterName, ip, port);
    //如果例項不存在
    if (instance == null) {
        if (clientBeat == null) {
            result.put(CommonParams.CODE, NamingResponseCode.RESOURCE_NOT_FOUND);
            return result;
        }

        Loggers.SRV_LOG.warn("[CLIENT-BEAT] The instance has been removed for health mechanism, "
                + "perform data compensation operations, beat: {}, serviceName: {}", clientBeat, serviceName);
        //根據您心跳內容建立一個例項資訊
        instance = new Instance();
        instance.setPort(clientBeat.getPort());
        instance.setIp(clientBeat.getIp());
        instance.setWeight(clientBeat.getWeight());
        instance.setMetadata(clientBeat.getMetadata());
        instance.setClusterName(clusterName);
        instance.setServiceName(serviceName);
        instance.setInstanceId(instance.getInstanceId());
        instance.setEphemeral(clientBeat.isEphemeral());
        //註冊例項
        serviceManager.registerInstance(namespaceId, serviceName, instance);
    }
    //獲取服務的資訊
    Service service = serviceManager.getService(namespaceId, serviceName);

    if (service == null) {
        throw new NacosException(NacosException.SERVER_ERROR,
                "service not found: " + serviceName + "@" + namespaceId);
    }
    //不存在的話，要建立一個進行處理
    if (clientBeat == null) {
        clientBeat = new RsInfo();
        clientBeat.setIp(ip);
        clientBeat.setPort(port);
        clientBeat.setCluster(clusterName);
    }
    //開啟心跳檢查任務
    service.processClientBeat(clientBeat);

    result.put(CommonParams.CODE, NamingResponseCode.OK);
    //5秒間隔
    if (instance.containsMetadata(PreservedMetadataKeys.HEART_BEAT_INTERVAL)) {
        result.put(SwitchEntry.CLIENT_BEAT_INTERVAL, instance.getInstanceHeartBeatInterval());
    }
    //告訴客戶端不需要帶上心跳資訊了，變成輕量級心跳了
    result.put(SwitchEntry.LIGHT_BEAT_ENABLED, switchDomain.isLightBeatEnabled());
    return result;
}
View Code

接下來我們看一下processClientBeat方法，該方法將ClientBeatProcessor放入到執行緒池中，接下來我們看下重點看下run方法,


該方法內部主要就是更新對應例項下心跳時間，整體上如下圖:

至此完成了從客戶端到服務端更新例項的心跳時間，下圖是整體的時序圖:

服務的健康檢查

Nacos Server會開啟一個定時任務來檢查註冊服務的健康情況，對於超過15秒沒收到客戶端的心跳例項會將它的 healthy屬性置為false，此時當客戶端不會將該例項的資訊發現，如果某個服務的例項超過30秒沒收到心跳，則剔除該例項，如果剔除的例項恢復，傳送心跳則會恢復。
當有例項註冊的時候，我們會看到有個service.init()的方法，該方法的實現主要是將ClientBeatCheckTask加入到執行緒池當中，如下圖:


ClientBeatCheckTask中的run方法主要做兩件事心跳時間超過15秒則設定該例項資訊為不健康狀況和心跳時間超過30秒則刪除該例項資訊，如下程式碼: 
public void run() {
    try {
        if (!getDistroMapper().responsible(service.getName())) {
            return;
        }

        if (!getSwitchDomain().isHealthCheckEnabled()) {
            return;
        }
        //獲取服務所有例項資訊
        List<Instance> instances = service.allIPs(true);

        // first set health status of instances:
        for (Instance instance : instances) {
            //如果心跳時間超過15秒則設定該例項資訊為不健康狀況
            if (System.currentTimeMillis() - instance.getLastBeat() > instance.getInstanceHeartBeatTimeOut()) {
                if (!instance.isMarked()) {
                    if (instance.isHealthy()) {
                        instance.setHealthy(false);
                        Loggers.EVT_LOG
                                .info("{POS} {IP-DISABLED} valid: {}:{}@{}@{}, region: {}, msg: client timeout after {}, last beat: {}",
                                        instance.getIp(), instance.getPort(), instance.getClusterName(),
                                        service.getName(), UtilsAndCommons.LOCALHOST_SITE,
                                        instance.getInstanceHeartBeatTimeOut(), instance.getLastBeat());
                        getPushService().serviceChanged(service);
                        ApplicationUtils.publishEvent(new InstanceHeartbeatTimeoutEvent(this, instance));
                    }
                }
            }
        }

        if (!getGlobalConfig().isExpireInstance()) {
            return;
        }

        // then remove obsolete instances:
        for (Instance instance : instances) {

            if (instance.isMarked()) {
                continue;
            }
            //如果心跳時間超過30秒則刪除該例項資訊
            if (System.currentTimeMillis() - instance.getLastBeat() > instance.getIpDeleteTimeout()) {
                // delete instance
                Loggers.SRV_LOG.info("[AUTO-DELETE-IP] service: {}, ip: {}", service.getName(),
                        JacksonUtils.toJson(instance));
                deleteIp(instance);
            }
        }

    } catch (Exception e) {
        Loggers.SRV_LOG.warn("Exception while processing client beat time out.", e);
    }

}
View Code

首先我們來看一下deleteIp方法，該方法內部主要通過構建刪除請求，傳送刪除請求，如下圖:


刪除例項的介面如下圖:

內部通過呼叫ServiceManager的removeInstance方法，如下圖:

重點看下substractIpAddresses內部通過呼叫updateIpAddresses，該方法內部主要就是移除到超過30秒的例項資訊，如下圖:

到此完成刪除例項的過程，整體的時序圖如下:

接下來我們看標記不健康時候的程式碼，這部分程式碼在客戶端註冊的時候也出現相同的程式碼，只是我們略過了，這部分也是觀察者模式的重要體現，從這裡我們可以學習到的東西在於結合Spring的事件機制，輕鬆實現觀察者模式，當然這個裡面也有部分我感覺寫的不太好，哈哈，大佬們看到勿噴。

首先我們看serviceChanged方法，該方法主要是釋出一個服務不健康的事件，如下圖:

接下來我們看下如何處理這個事件，這個時候涉及PushService這個類，整體的繼承結構如下圖:

我們看到該類的繼承ApplicationListener介面，該介面是一個支援泛型的介面，傳入了ServiceChangeEvent的類，此處就是對事件的處理，如下圖:

接下來看一下onApplicationEvent方法，這個方法主要完成了準備資料，傳送資料這幾件事情: 
public void onApplicationEvent(ServiceChangeEvent event) {
    Service service = event.getService();
    String serviceName = service.getName();
    String namespaceId = service.getNamespaceId();

    Future future = GlobalExecutor.scheduleUdpSender(() -> {
        try {
            Loggers.PUSH.info(serviceName + " is changed, add it to push queue.");
            //獲取所有需要推送的客戶端
            ConcurrentMap<String, PushClient> clients = clientMap
                    .get(UtilsAndCommons.assembleFullServiceName(namespaceId, serviceName));
            if (MapUtils.isEmpty(clients)) {
                return;
            }

            Map<String, Object> cache = new HashMap<>(16);
            long lastRefTime = System.nanoTime();
            for (PushClient client : clients.values()) {
                //超時的不刪除跳過處理
                if (client.zombie()) {
                    Loggers.PUSH.debug("client is zombie: " + client.toString());
                    clients.remove(client.toString());
                    Loggers.PUSH.debug("client is zombie: " + client.toString());
                    continue;
                }

                Receiver.AckEntry ackEntry;
                Loggers.PUSH.debug("push serviceName: {} to client: {}", serviceName, client.toString());
                String key = getPushCacheKey(serviceName, client.getIp(), client.getAgent());
                byte[] compressData = null;
                Map<String, Object> data = null;

                if (switchDomain.getDefaultPushCacheMillis() >= 20000 && cache.containsKey(key)) {
                    org.javatuples.Pair pair = (org.javatuples.Pair) cache.get(key);
                    compressData = (byte[]) (pair.getValue0());
                    data = (Map<String, Object>) pair.getValue1();
                    Loggers.PUSH.debug("[PUSH-CACHE] cache hit: {}:{}", serviceName, client.getAddrStr());
                }
                //準備UDP資料
                if (compressData != null) {
                    ackEntry = prepareAckEntry(client, compressData, data, lastRefTime);
                } else {
                    ackEntry = prepareAckEntry(client, prepareHostsData(client), lastRefTime);
                    if (ackEntry != null) {
                        cache.put(key, new org.javatuples.Pair<>(ackEntry.origin.getData(), ackEntry.data));
                    }
                }

                Loggers.PUSH.info("serviceName: {} changed, schedule push for: {}, agent: {}, key: {}",
                        client.getServiceName(), client.getAddrStr(), client.getAgent(),
                        (ackEntry == null ? null : ackEntry.key));
                //傳送資料
                udpPush(ackEntry);
            }
        } catch (Exception e) {
            Loggers.PUSH.error("[NACOS-PUSH] failed to push serviceName: {} to client, error: {}", serviceName, e);

        } finally {
            //傳送完成刪除
            futureMap.remove(UtilsAndCommons.assembleFullServiceName(namespaceId, serviceName));
        }

    }, 1000, TimeUnit.MILLISECONDS);
    //增加待推送的任務
    futureMap.put(UtilsAndCommons.assembleFullServiceName(namespaceId, serviceName), future);

}
View Code

接下里我們重點看下udpPush的方法，整個方法主要是通過一個Map物件來記錄UDP請求，如果沒收到就重試傳送請求，整體如下:

private static Receiver.AckEntry udpPush(Receiver.AckEntry ackEntry) {
    if (ackEntry == null) {
        Loggers.PUSH.error("[NACOS-PUSH] ackEntry is null.");
        return null;
    }

    //如果大於最大的嘗試次數
    //移除傳送的資料和待確認的key
    //失敗推送的次數+1
    if (ackEntry.getRetryTimes() > MAX_RETRY_TIMES) {
        Loggers.PUSH.warn("max re-push times reached, retry times {}, key: {}", ackEntry.retryTimes, ackEntry.key);
        ackMap.remove(ackEntry.key);
        udpSendTimeMap.remove(ackEntry.key);
        failedPush += 1;
        return ackEntry;
    }

    try {
        if (!ackMap.containsKey(ackEntry.key)) {
            totalPush++;
        }
        //記錄UDP請求的返回資訊
        ackMap.put(ackEntry.key, ackEntry);
        udpSendTimeMap.put(ackEntry.key, System.currentTimeMillis());

        Loggers.PUSH.info("send udp packet: " + ackEntry.key);
        //傳送UDP請求
        udpSocket.send(ackEntry.origin);

        ackEntry.increaseRetryTime();
        //如果UDP沒收到返回資訊 每10秒嘗試一下
        GlobalExecutor.scheduleRetransmitter(new Retransmitter(ackEntry),
                TimeUnit.NANOSECONDS.toMillis(ACK_TIMEOUT_NANOS), TimeUnit.MILLISECONDS);

        return ackEntry;
    } catch (Exception e) {
        Loggers.PUSH.error("[NACOS-PUSH] failed to push data: {} to client: {}, error: {}", ackEntry.data,
                ackEntry.origin.getAddress().getHostAddress(), e);
        ackMap.remove(ackEntry.key);
        udpSendTimeMap.remove(ackEntry.key);
        failedPush += 1;

        return null;
    }
}
View Code

服務端有傳送，那麼客戶端就有接收的，接收部分我理解上是服務發現部分，這裡我們就不做過多介紹，待下一篇再來聊聊。

結束

歡迎大家點點關注，點點贊，感謝!

相關推薦

Nacos服務心跳和健康檢查原始碼介紹

服務心跳 Nacos Client會維護一個定時任務通過持續呼叫服務端的介面更新心跳時間，保證自己處於存活狀態，防止服務端將服務剔除，Nacos預設5秒向服務端傳送一次，通過請求服務端介面/instance/beat傳送心跳。客戶端服務在註冊服務的時候會增加一個心跳的任務，如下圖所示: 首先看下BeatI

Nacos使用和註冊部分原始碼介紹

Nacos簡單介紹 Nacos致力於幫助您發現、配置和管理微服務。Nacos提供了一組簡單易用的特性集，幫助您快速實現動態服務發現、服務配置、服務元資料及流量管理。Nacos幫助您更敏捷和容易地構建、交付和管理微服務平臺。Nacos是構建以“服務”為中心的現代應用架構 (例如微服務正規化、雲原生正規化) 的

Java靜態檢測工具/Java代碼規範和質量檢查簡單介紹（轉）

pub 詳細 職責 web 循環 問題： 集成 死鎖 參考 靜態檢查： 靜態測試包括代碼檢查、靜態結構分析、代碼質量度量等。它可以由人工進行，充分發揮人的邏輯思維優勢，也可以借助軟件工具自動進行。代碼檢查代碼檢查包括代碼走查、桌面檢查、代碼審查等，主要檢查代碼和設計的一致

Eureka的自我保護模式、多網絡卡環境下的IP選擇和健康檢查

Eureka的自我保護模式 禁用自我保護模式： eureka.server.enable-self-preservation = false 多網絡卡環境下的IP選擇 對於多網絡卡的伺服器，各個微服務註冊到Eureka Server上的IP要如何指定呢？ 指定IP在某些場合下

K8s中Pod健康檢查原始碼分析

摘要： 本文基於k8s 1.11.0版本的從原始碼的角度分析了Pod的健康檢查實現邏輯。建議通過k8s部署生產環境應用時

Java靜態檢測工具/Java程式碼規範和質量檢查簡單介紹

靜態檢查： 靜態測試包括程式碼檢查、靜態結構分析、程式碼質量度量等。它可以由人工進行，充分發揮人的邏輯思維優勢，也可以藉助軟體工

Spring Cloud Alibaba（一） 如何使用nacos服務註冊和發現

Nacos介紹 Nacos 致力於幫助您發現、配置和管理微服務。Nacos 提供了一組簡單易用的特性集，幫助您快速實現動態服務發現、服務配置、服務元資料及流量管理。 Nacos 幫助您更敏捷和容易地構建、交付和管理微服務平臺。 Nacos 是構建以“服務”為中心的現代應用架構 (例如微服務正規化、雲原生正規化

架構師之路--視頻業務介紹，離線服務架構和各種集群原理

目的 -- 自己的 超過 覆蓋 基本上 添加節點 電視 是我 　　先聊聊業務。我們媒資這邊目前的核心數據是樂視視頻的樂視meta和專門存儲電視劇，綜藝節目，體育賽事這種長視頻的作品庫。樂視視頻的數據都是多方審核的，需要很多運營。但是作品庫部分卻是弱運營的，運營都不超過10個

SaltStack介紹——SaltStack是一種新的基礎設施管理方法開發軟件,簡單易部署,可伸縮的足以管理成千上萬的服務器,和足夠快的速度控制,與他們交流

con mar stack 通信 class 交流 ast 集中 速度 SaltStack介紹和架構解析 簡介 SaltStack是一種新的基礎設施管理方法開發軟件,簡單易部署,可伸縮的足以管理成千上萬的服務器,和足夠快的速度控制,與他們交流,以毫秒為單位。S

用腳本自主初始優化和全面檢查服務器

shell腳本 系統優化 系統檢查 前段時間看到一篇用腳本一鍵檢查服務器的優秀博文，寫的真棒，美中不足的是在實際應用中發現一個小問題：他的腳本好是好，但是由於環境和服務不一樣，會檢查沒有的服務項，要去掉這些檢查，還要一行行註釋掉，很是麻煩，於是就優化腳本內容，把所有的檢查項都定義為一個

[Kubernetes]容器健康檢查和恢復機制

file ima star 同時 bob bsp ber pan 出現 　　在Kubernetes中，可以為Pod裏的容器定義一個健康檢查探針（Probe），這樣Kubernetes會根據這個Probe的返回值決定這個容器的狀態，而不是直接以容器是否允許（來自Docker返

Spring Cloud微服務系統架構的一些簡單介紹和使用

Spring Cloud 目錄 特徵 雲原生應用程式 Spring Cloud上下文：應用程式上下文服務 引導應用程式上下文 應用程式上下文層次結構

【小家java】SortedMap和NavigableMap的使用介紹---TreeMap的原始碼簡單分析

相關閱讀 【小家java】java5新特性（簡述十大新特性） 重要一躍 【小家java】java6新特性（簡述十大新特性） 雞肋升級 【小家java】java7新特性（簡述八大新特性） 不溫不火 【小家java】java8新特性（簡述十大新特性） 飽受讚譽 【小家java】java9

kong閘道器健康檢查和斷路器配置

使用kong閘道器實現負載均衡， 根據配置規則將後端tatget節點路由到客戶端請求，當後端某個target服務不能訪問時，我們希望kong閘道器在負載中跳過該節點，當該節點可用時，我們希望kong閘道器能夠將該節點加入負載，這些都可以通過upstream配置來實現 kong閘道器健康檢查有

nginx健康檢查和主備切換

1、參考部落格 https://blog.csdn.net/moqiang02/article/details/42846221 2、使用第一種方式注意點 在主備切換的方式下需要指定備用節點 upstream mysvr { server 192.168.0.104:8080;

架構師之路--視訊業務介紹，離線服務架構和各種叢集原理

java架構師、叢集、高可用、高可擴充套件、高效能、高併發、效能優化、Spring boot、Redis、ActiveMQ、Nginx、Netty、Jvm大型分散式專案實戰 視訊課程內容包含： 高階Java架構師包含：Spring boot、Spring  cloud、

第二篇：SpringCloud 構建微服務系統之服務註冊和發現（nacos）

上一篇我們學習了一下consul在soringcloud中的使用。今天要給大家介紹的阿里巴巴中介軟體團隊出品的Nacos來作為新一代的服務管理中介軟體。 首先學習Nacos之前，我們應該看看Nacos的官網，對它有一個初步的認識。 1. Nacos 官網 （https://nacos

Kubernetes之健康檢查與服務依賴處理_Kubernetes中文社群

【編者的話】對線上業務來說，保證服務的正常穩定是重中之重，對故障服務的及時處理避免影響業務以及快速恢復一直是開發運維的難點。Kubernetes提供了健康檢查服務，對於檢測到故障服務會被及時自動下線，以及通過重啟服務的方式使服務自動恢復。而對於服務依賴，無論資源描述檔案是pod, rc或dep

基於Spring Boot和Spring Cloud實現微服務架構學習(一)-Spring框架介紹

總結 看了幾周Spring相關框架的書籍和官方demo，是時候開始總結下這中間的學習感悟。 首先，最想說的是，當你要學習一套最新的技術時，官網的英文文件是學習的最佳渠道。因為網上流傳的多數資料是官網翻譯而來，很多描述的重點也都偏向於作者自身碰到的問題，這樣就很容易讓你理解

WebSocket在服務端和客戶端通訊demo,支援心跳檢測+斷線重連

一、為什麼需要 WebSocket？ 初次接觸 WebSocket 的人，都會問同樣的問題：我們已經有了 HTTP 協議，為什麼還需要另一個協議？它能帶來什麼好處？ 答案很簡單，因為 HTTP 協議有一個缺陷：通訊只能由客戶端發起。 舉例來說，我們想了解今天的天氣，只能是客戶端向伺服器發出