微服務：剖析一下源碼，Nacos的健康檢查竟如此簡單

作者：二師兄 2021-07-05 06:51:41

Nacos中臨時實例基于心跳上報方式維持活性，基本的健康檢查流程基本如下：Nacos客戶端會維護一個定時任務，每隔5秒發送一次心跳請求，以確保自己處于活躍狀態。

[[409195]]

本文轉載自微信公眾號「程序新視界」，作者二師兄。轉載本文請聯系程序新視界公眾號。

前言

前面我們多次提到Nacos的健康檢查，比如《微服務之：服務掛的太干脆，Nacos還沒反應過來，怎么辦?》一文中還對健康檢查進行了自定義調優。那么，Nacos的健康檢查和心跳機制到底是如何實現的呢?在項目實踐中是否又可以參考Nacos的健康檢查機制，運用于其他地方呢?

這篇文章，就帶大家來揭開Nacos健康檢查機制的面紗。

Nacos的健康檢查

Nacos中臨時實例基于心跳上報方式維持活性，基本的健康檢查流程基本如下：Nacos客戶端會維護一個定時任務，每隔5秒發送一次心跳請求，以確保自己處于活躍狀態。Nacos服務端在15秒內如果沒收到客戶端的心跳請求，會將該實例設置為不健康，在30秒內沒收到心跳，會將這個臨時實例摘除。

原理很簡單，關于代碼層的實現，下面來就逐步來進行解析。

客戶端的心跳

實例基于心跳上報的形式來維持活性，當然就離不開心跳功能的實現了。這里以客戶端心跳實現為基準來進行分析。

Spring Cloud提供了一個標準接口ServiceRegistry，Nacos對應的實現類為NacosServiceRegistry。Spring Cloud項目啟動時會實例化NacosServiceRegistry，并調用它的register方法來進行實例的注冊。

@Override 
public void register(Registration registration) {  
   // ... 
   NamingService namingService = namingService(); 
   String serviceId = registration.getServiceId(); 
   String group = nacosDiscoveryProperties.getGroup(); 
 
   Instance instance = getNacosInstanceFromRegistration(registration); 
 
   try { 
      namingService.registerInstance(serviceId, group, instance); 
      log.info("nacos registry, {} {} {}:{} register finished", group, serviceId, 
            instance.getIp(), instance.getPort()); 
   }catch (Exception e) { 
      // ... 
   } 
}

在該方法中有兩處需要注意，第一處是構建Instance的getNacosInstanceFromRegistration方法，該方法內會設置Instance的元數據(metadata)，通過源元數據可以配置服務器端健康檢查的參數。比如，在Spring Cloud中配置的如下參數，都可以通過元數據項在服務注冊時傳遞給Nacos的服務端。

spring: 
  application: 
    name: user-service-provider 
  cloud: 
    nacos: 
      discovery: 
        server-addr: 127.0.0.1:8848 
        heart-beat-interval: 5000 
        heart-beat-timeout: 15000 
       ip-delete-timeout: 30000

其中的heart-beat-interval、heart-beat-timeout、ip-delete-timeout這些健康檢查的參數，都是基于元數據上報上去的。

register方法的第二處就是調用NamingService#registerInstance來進行實例的注冊。NamingService是由Nacos的客戶端提供，也就是說Nacos客戶端的心跳本身是由Nacos生態提供的。

在registerInstance方法中最終會調用到下面的方法：

@Override 
public void registerInstance(String serviceName, String groupName, Instance instance) throws NacosException { 
    NamingUtils.checkInstanceIsLegal(instance); 
    String groupedServiceName = NamingUtils.getGroupedName(serviceName, groupName); 
    if (instance.isEphemeral()) { 
        BeatInfo beatInfo = beatReactor.buildBeatInfo(groupedServiceName, instance); 
        beatReactor.addBeatInfo(groupedServiceName, beatInfo); 
    } 
    serverProxy.registerService(groupedServiceName, groupName, instance); 
}

其中BeatInfo#addBeatInfo便是進行心跳處理的入口。當然，前提條件是當前的實例需要是臨時(瞬時)實例。

對應的方法實現如下：

public void addBeatInfo(String serviceName, BeatInfo beatInfo) { 
    NAMING_LOGGER.info("[BEAT] adding beat: {} to beat map.", beatInfo); 
    String key = buildKey(serviceName, beatInfo.getIp(), beatInfo.getPort()); 
    BeatInfo existBeat = null; 
    //fix #1733 
    if ((existBeat = dom2Beat.remove(key)) != null) { 
        existBeat.setStopped(true); 
    } 
    dom2Beat.put(key, beatInfo); 
    executorService.schedule(new BeatTask(beatInfo), beatInfo.getPeriod(), TimeUnit.MILLISECONDS); 
    MetricsMonitor.getDom2BeatSizeMonitor().set(dom2Beat.size()); 
}

在倒數第二行可以看到，客戶端是通過定時任務來處理心跳的，具體的心跳請求由BeatTask完成。定時任務的執行頻次，封裝在BeatInfo，回退往上看，會發現BeatInfo的Period來源于Instance#getInstanceHeartBeatInterval()。該方法具體實現如下：

public long getInstanceHeartBeatInterval() { 
    return this.getMetaDataByKeyWithDefault("preserved.heart.beat.interval", Constants.DEFAULT_HEART_BEAT_INTERVAL); 
}

可以看出定時任務的執行間隔就是配置的metadata中的數據preserved.heart.beat.interval，與上面提到配置heart-beat-interval本質是一回事，默認是5秒。

BeatTask類具體實現如下：

class BeatTask implements Runnable { 
     
    BeatInfo beatInfo; 
     
    public BeatTask(BeatInfo beatInfo) { 
        this.beatInfo = beatInfo; 
    } 
     
    @Override 
    public void run() { 
        if (beatInfo.isStopped()) { 
            return; 
        } 
        long nextTime = beatInfo.getPeriod(); 
        try { 
            JsonNode result = serverProxy.sendBeat(beatInfo, BeatReactor.this.lightBeatEnabled); 
            long interval = result.get("clientBeatInterval").asLong(); 
            boolean lightBeatEnabled = false; 
            if (result.has(CommonParams.LIGHT_BEAT_ENABLED)) { 
                lightBeatEnabled = result.get(CommonParams.LIGHT_BEAT_ENABLED).asBoolean(); 
            } 
            BeatReactor.this.lightBeatEnabled = lightBeatEnabled; 
            if (interval > 0) { 
                nextTime = interval; 
            } 
            int code = NamingResponseCode.OK; 
            if (result.has(CommonParams.CODE)) { 
                code = result.get(CommonParams.CODE).asInt(); 
            } 
            if (code == NamingResponseCode.RESOURCE_NOT_FOUND) { 
                Instance instance = new Instance(); 
                instance.setPort(beatInfo.getPort()); 
                instance.setIp(beatInfo.getIp()); 
                instance.setWeight(beatInfo.getWeight()); 
                instance.setMetadata(beatInfo.getMetadata()); 
                instance.setClusterName(beatInfo.getCluster()); 
                instance.setServiceName(beatInfo.getServiceName()); 
                instance.setInstanceId(instance.getInstanceId()); 
                instance.setEphemeral(true); 
                try { 
                    serverProxy.registerService(beatInfo.getServiceName(), 
                            NamingUtils.getGroupName(beatInfo.getServiceName()), instance); 
                } catch (Exception ignore) { 
                } 
            } 
        } catch (NacosException ex) { 
            NAMING_LOGGER.error("[CLIENT-BEAT] failed to send beat: {}, code: {}, msg: {}", 
                    JacksonUtils.toJson(beatInfo), ex.getErrCode(), ex.getErrMsg()); 
             
        } 
        executorService.schedule(new BeatTask(beatInfo), nextTime, TimeUnit.MILLISECONDS); 
    } 
}

在run方法中通過NamingProxy#sendBeat完成了心跳請求的發送，而在run方法的最后，再次開啟了一個定時任務，這樣周期性的進行心跳請求。

NamingProxy#sendBeat方法實現如下：

public JsonNode sendBeat(BeatInfo beatInfo, boolean lightBeatEnabled) throws NacosException { 
     
    if (NAMING_LOGGER.isDebugEnabled()) { 
        NAMING_LOGGER.debug("[BEAT] {} sending beat to server: {}", namespaceId, beatInfo.toString()); 
    } 
    Map<String, String> params = new HashMap<String, String>(8); 
    Map<String, String> bodyMap = new HashMap<String, String>(2); 
    if (!lightBeatEnabled) { 
        bodyMap.put("beat", JacksonUtils.toJson(beatInfo)); 
    } 
    params.put(CommonParams.NAMESPACE_ID, namespaceId); 
    params.put(CommonParams.SERVICE_NAME, beatInfo.getServiceName()); 
    params.put(CommonParams.CLUSTER_NAME, beatInfo.getCluster()); 
    params.put("ip", beatInfo.getIp()); 
    params.put("port", String.valueOf(beatInfo.getPort())); 
    String result = reqApi(UtilAndComs.nacosUrlBase + "/instance/beat", params, bodyMap, HttpMethod.PUT); 
    return JacksonUtils.toObj(result); 
}

實際上，就是調用了Nacos服務端提供的"/nacos/v1/ns/instance/beat"服務。

在客戶端的常量類Constants中定義了心跳相關的默認參數：

static { 
    DEFAULT_HEART_BEAT_TIMEOUT = TimeUnit.SECONDS.toMillis(15L); 
    DEFAULT_IP_DELETE_TIMEOUT = TimeUnit.SECONDS.toMillis(30L); 
    DEFAULT_HEART_BEAT_INTERVAL = TimeUnit.SECONDS.toMillis(5L); 
}

這樣就呼應了最開始說的Nacos健康檢查機制的幾個時間維度。

服務端接收心跳

分析客戶端的過程中已經可以看出請求的是/nacos/v1/ns/instance/beat這個服務。Nacos服務端是在Naming項目中的InstanceController中實現的。

@CanDistro 
@PutMapping("/beat") 
@Secured(parser = NamingResourceParser.class, action = ActionTypes.WRITE) 
public ObjectNode beat(HttpServletRequest request) throws Exception { 
 
    // ... 
    Instance instance = serviceManager.getInstance(namespaceId, serviceName, clusterName, ip, port); 
 
    if (instance == null) { 
        // ... 
        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); 
    // ... 
    service.processClientBeat(clientBeat); 
    // ... 
    return result; 
}

服務端在接收到請求時，主要做了兩件事：第一，如果發送心跳的實例不存在，則將其進行注冊;第二，調用其Service的processClientBeat方法進行心跳處理。

processClientBeat方法實現如下：

public void processClientBeat(final RsInfo rsInfo) { 
    ClientBeatProcessor clientBeatProcessor = new ClientBeatProcessor(); 
    clientBeatProcessor.setService(this); 
    clientBeatProcessor.setRsInfo(rsInfo); 
    HealthCheckReactor.scheduleNow(clientBeatProcessor); 
}

再來看看ClientBeatProcessor中對具體任務的實現：

@Override 
public void run() { 
    Service service = this.service; 
    // logging     
    String ip = rsInfo.getIp(); 
    String clusterName = rsInfo.getCluster(); 
    int port = rsInfo.getPort(); 
    Cluster cluster = service.getClusterMap().get(clusterName); 
    List<Instance> instances = cluster.allIPs(true); 
     
    for (Instance instance : instances) { 
        if (instance.getIp().equals(ip) && instance.getPort() == port) { 
            // logging 
            instance.setLastBeat(System.currentTimeMillis()); 
            if (!instance.isMarked()) { 
                if (!instance.isHealthy()) { 
                    instance.setHealthy(true); 
                    // logging 
                    getPushService().serviceChanged(service); 
                } 
            } 
        } 
    } 
}

在run方法中先檢查了發送心跳的實例和IP是否一致，如果一致則更新最后一次心跳時間。同時，如果該實例之前未被標記且處于不健康狀態，則將其改為健康狀態，并將變動通過PushService提供事件機制進行發布。事件是由Spring的ApplicationContext進行發布，事件為ServiceChangeEvent。

通過上述心跳操作，Nacos服務端的實例的健康狀態和最后心跳時間已經被刷新。那么，如果沒有收到心跳時，服務器端又是如何判斷呢?

服務端心跳檢查

客戶端發起心跳，服務器端來檢查客戶端的心跳是否正常，或者說對應的實例中的心跳更新時間是否正常。

服務器端心跳的觸發是在服務實例注冊時觸發的，同樣在InstanceController中，register注冊實現如下：

@CanDistro 
@PostMapping 
@Secured(parser = NamingResourceParser.class, action = ActionTypes.WRITE) 
public String register(HttpServletRequest request) throws Exception { 
    // ... 
    final Instance instance = parseInstance(request); 
 
    serviceManager.registerInstance(namespaceId, serviceName, instance); 
    return "ok"; 
}

ServiceManager#registerInstance實現代碼如下：

public void registerInstance(String namespaceId, String serviceName, Instance instance) throws NacosException { 
     
    createEmptyService(namespaceId, serviceName, instance.isEphemeral()); 
    // ... 
}

心跳相關實現在第一次創建空的Service中實現，最終會調到如下方法：

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 thrown 
        service.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()方法實現：

public void init() { 
    HealthCheckReactor.scheduleCheck(clientBeatCheckTask); 
    for (Map.Entry<String, Cluster> entry : clusterMap.entrySet()) { 
        entry.getValue().setService(this); 
        entry.getValue().init(); 
    } 
}

HealthCheckReactor#scheduleCheck方法實現：

public static void scheduleCheck(ClientBeatCheckTask task) { 
    futureMap.computeIfAbsent(task.taskKey(), 
            k -> GlobalExecutor.scheduleNamingHealth(task, 5000, 5000, TimeUnit.MILLISECONDS)); 
}

延遲5秒執行，每5秒檢查一次。

在init方法的第一行便可以看到執行健康檢查的Task，具體Task是由ClientBeatCheckTask來實現，對應的run方法核心代碼如下：

@Override 
public void run() { 
    // ...         
    List<Instance> instances = service.allIPs(true); 
     
    // first set health status of instances: 
    for (Instance instance : instances) { 
        if (System.currentTimeMillis() - instance.getLastBeat() > instance.getInstanceHeartBeatTimeOut()) { 
            if (!instance.isMarked()) { 
                if (instance.isHealthy()) { 
                    instance.setHealthy(false); 
                    // logging... 
                    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; 
        } 
         
        if (System.currentTimeMillis() - instance.getLastBeat() > instance.getIpDeleteTimeout()) { 
            // delete instance 
            deleteIp(instance); 
        } 
    } 
}

在第一個for循環中，先判斷當前時間與上次心跳時間的間隔是否大于超時時間。如果實例已經超時，且為被標記，且健康狀態為健康，則將健康狀態設置為不健康，同時發布狀態變化的事件。

在第二個for循環中，如果實例已經被標記則跳出循環。如果未標記，同時當前時間與上次心跳時間的間隔大于刪除IP時間，則將對應的實例刪除。

小結

通過本文的源碼分析，我們從Spring Cloud開始，追蹤到Nacos Client中的心跳時間，再追蹤到Nacos服務端接收心跳的實現和檢查實例是否健康的實現。想必通過整個源碼的梳理，你已經對整個Nacos心跳的實現有所了解。關注我，持續更新Nacos的最新干貨。

責任編輯：武曉燕來源：程序新視界

Nacos 微服務源碼