從實現原理來講，Nacos 為什么這么強？

作者：不才陳某 2024-09-04 10:44:19

Nacos支持基于DNS和RPC的服務發現，即服務消費者可以使用DNS或者HTTP的方式來查找和發現服務。Nacos提供對服務的實時的健康檢查，阻止向不健康的主機或者服務實例發送請求。Nacos支持傳輸層（Ping/TCP）、應用層（HTTP、Mysql）的健康檢查。

今天來分享一下Nacos注冊中心的底層原理，從服務注冊到服務發現，非常細致

一、 Nacos介紹

再講Nacos之前，先來講一下服務注冊和發現。我們知道，現在微服務架構是目前開發的一個趨勢。服務消費者要去調用多個服務提供者組成的集群。這里需要做到以下幾點：

服務消費者需要在本地配置文件中維護服務提供者集群的每個節點的請求地址。
服務提供者集群中如果某個節點宕機，服務消費者的本地配置中需要同步刪除這個節點的請求地址，防止請求發送到已經宕機的節點上造成請求失敗。

因此需要引入服務注冊中心，它具有以下幾個功能：

服務地址的管理。
服務注冊。
服務動態感知。

而Nacos致力于解決微服務中的統一配置，服務注冊和發現等問題。Nacos集成了注冊中心和配置中心。其相關特性包括：

服務發現和服務健康監測

動態配置服務

動態配置服務可以以中心化、外部化和動態化的方式管理所有環境的應用配置和服務配置。

動態DNS服務

支持權重路由，讓開發者更容易的實現中間層的負載均衡、更靈活的路由策略、流量控制以及DNS解析服務。

服務和元數據管理

Nacos允許開發者從微服務平臺建設的視角來管理數據中心的所有服務和元數據。如：服務的生命周期、靜態依賴分析、服務的健康狀態、服務的流量管理、路由和安全策略等。

二、Nacos注冊中心實現原理分析

2.1 Nacos架構圖

以下是Nacos的架構圖：

圖片

其中分為這么幾個模塊：

Provider APP：服務提供者。
Consumer APP：服務消費者。
Name Server：通過Virtual IP或者DNS的方式實現Nacos高可用集群的服務路由。
Nacos Server：Nacos服務提供者。

OpenAPI：功能訪問入口。

Config Service、Naming Service：Nacos提供的配置服務、名字服務模塊。

Consistency Protocol：一致性協議，用來實現Nacos集群節點的數據同步，使用Raft算法實現。

其中包含：

Nacos Console：Nacos控制臺。

小總結：

服務提供者通過VIP（Virtual IP）訪問Nacos Server高可用集群，基于OpenAPI完成服務的注冊和服務的查詢。
Nacos Server的底層則通過數據一致性算法（Raft）來完成節點的數據同步。

2.2 注冊中心的原理

這里對其原理做一個大致的介紹，在后文則從源碼角度進行分析。

首先，服務注冊的功能體現在：

服務實例啟動時注冊到服務注冊表、關閉時則注銷（服務注冊）。
服務消費者可以通過查詢服務注冊表來獲得可用的實例（服務發現）。
服務注冊中心需要調用服務實例的健康檢查API來驗證其是否可以正確的處理請求（健康檢查）。

Nacos服務注冊和發現的實現原理的圖如下：

圖片

三、Nacos源碼分析

前提（在本地或者虛機上先啟動好Nacos）這一部分從2個角度來講Nacos是如何實現的：

服務注冊。
服務發現

3.1 Nacos服務注冊

首先看下一個包：spring-cloud-commons

圖片

這個ServiceRegistry接口是SpringCloud提供的服務注冊的標準，集成到SpringCloud中實現服務注冊的組件，都需要實現這個接口。來看下它的結構：

public interface ServiceRegistry<R extends Registration> {
    void register(R registration);

    void deregister(R registration);

    void close();

    void setStatus(R registration, String status);

    <T> T getStatus(R registration);
}

那么對于Nacos而言，該接口的實現類是NacosServiceRegistry，該類在這個pom包下：

圖片

再回過頭來看spring-cloud-commons包:

圖片

spring.factories主要是包含了自動裝配的配置信息，如圖：

圖片

在我之前的文章里我有提到過，在spring.factories中配置EnableAutoConfiguration的內容后，項目在啟動的時候，會導入相應的自動配置類，那么也就允許對該類的相關屬性進行一個自動裝配。那么顯然，在這里導入了AutoServiceRegistrationAutoConfiguration這個類，而這個類顧名思義是服務注冊相關的配置類。

該類的完整代碼如下：

@Configuration(
    proxyBeanMethods = false
)
@Import({AutoServiceRegistrationConfiguration.class})
@ConditionalOnProperty(
    value = {"spring.cloud.service-registry.auto-registration.enabled"},
    matchIfMissing = true
)
public class AutoServiceRegistrationAutoConfiguration {
    @Autowired(
        required = false
    )
    private AutoServiceRegistration autoServiceRegistration;
    @Autowired
    private AutoServiceRegistrationProperties properties;

    public AutoServiceRegistrationAutoConfiguration() {
    }

    @PostConstruct
    protected void init() {
        if (this.autoServiceRegistration == null && this.properties.isFailFast()) {
            throw new IllegalStateException("Auto Service Registration has been requested, but there is no AutoServiceRegistration bean");
        }
    }
}

這里做一個分析，AutoServiceRegistrationAutoConfiguration中注入了AutoServiceRegistration實例，該類的關系圖如下：

圖片

我們先來看一下這個抽象類AbstractAutoServiceRegistration：

public abstract class AbstractAutoServiceRegistration<R extends Registration> implements AutoServiceRegistration, 
ApplicationContextAware, 
ApplicationListener<WebServerInitializedEvent> {
 public void onApplicationEvent(WebServerInitializedEvent event) {
     this.bind(event);
 }
}

這里實現了ApplicationListener接口，并且傳入了WebServerInitializedEvent作為泛型，啥意思嘞，意思是：

NacosAutoServiceRegistration監聽WebServerInitializedEvent事件。
也就是WebServer初始化完成后，會調用對應的事件綁定方法，調用onApplicationEvent（），該方法最終調用NacosServiceRegistry的register（）方法（NacosServiceRegistry實現了Spring的一個服務注冊標準接口）。

對于register（）方法，主要調用的是Nacos Client SDK中的NamingService下的registerInstance（）方法完成服務的注冊。

public void register(Registration registration) {
    if (StringUtils.isEmpty(registration.getServiceId())) {
        log.warn("No service to register for nacos client...");
    } else {
        String serviceId = registration.getServiceId();
        String group = this.nacosDiscoveryProperties.getGroup();
        Instance instance = this.getNacosInstanceFromRegistration(registration);

        try {
            this.namingService.registerInstance(serviceId, group, instance);
            log.info("nacos registry, {} {} {}:{} register finished", new Object[]{group, serviceId, instance.getIp(), instance.getPort()});
        } catch (Exception var6) {
            log.error("nacos registry, {} register failed...{},", new Object[]{serviceId, registration.toString(), var6});
            ReflectionUtils.rethrowRuntimeException(var6);
        }

    }
}

public void registerInstance(String serviceName, String groupName, Instance instance) throws NacosException {
    if (instance.isEphemeral()) {
        BeatInfo beatInfo = new BeatInfo();
        beatInfo.setServiceName(NamingUtils.getGroupedName(serviceName, groupName));
        beatInfo.setIp(instance.getIp());
        beatInfo.setPort(instance.getPort());
        beatInfo.setCluster(instance.getClusterName());
        beatInfo.setWeight(instance.getWeight());
        beatInfo.setMetadata(instance.getMetadata());
        beatInfo.setScheduled(false);
        long instanceInterval = instance.getInstanceHeartBeatInterval();
        beatInfo.setPeriod(instanceInterval == 0L ? DEFAULT_HEART_BEAT_INTERVAL : instanceInterval);
        // 1.addBeatInfo（）負責創建心跳信息實現健康監測。因為Nacos Server必須要確保注冊的服務實例是健康的。
        // 而心跳監測就是服務健康監測的一種手段。
        this.beatReactor.addBeatInfo(NamingUtils.getGroupedName(serviceName, groupName), beatInfo);
    }
 // 2.registerService（）實現服務的注冊
    this.serverProxy.registerService(NamingUtils.getGroupedName(serviceName, groupName), groupName, instance);
}

再來看一下心跳監測的方法addBeatInfo（）：

public void addBeatInfo(String serviceName, BeatInfo beatInfo) {
    LogUtils.NAMING_LOGGER.info("[BEAT] adding beat: {} to beat map.", beatInfo);
    String key = this.buildKey(serviceName, beatInfo.getIp(), beatInfo.getPort());
    BeatInfo existBeat = null;
    if ((existBeat = (BeatInfo)this.dom2Beat.remove(key)) != null) {
        existBeat.setStopped(true);
    }

    this.dom2Beat.put(key, beatInfo);
    // 通過schedule（）方法，定時的向服務端發送一個數據包，然后啟動一個線程不斷地檢測服務端的回應。
    // 如果在指定的時間內沒有收到服務端的回應，那么認為服務器出現了故障。
    // 參數1：可以說是這個實例的相關信息。
    // 參數2：一個long類型的時間，代表從現在開始推遲執行的時間，默認是5000
    // 參數3：時間的單位，默認是毫秒，結合5000即代表每5秒發送一次心跳數據包
    this.executorService.schedule(new BeatReactor.BeatTask(beatInfo), beatInfo.getPeriod(), TimeUnit.MILLISECONDS);
    MetricsMonitor.getDom2BeatSizeMonitor().set((double)this.dom2Beat.size());
}

心跳檢查如果正常，即代表這個需要注冊的服務是健康的，那么執行下面的注冊方法registerInstance（）：

public void registerService(String serviceName, String groupName, Instance instance) throws NacosException {
    LogUtils.NAMING_LOGGER.info("[REGISTER-SERVICE] {} registering service {} with instance: {}", new Object[]{this.namespaceId, serviceName, instance});
    Map<String, String> params = new HashMap(9);
    params.put("namespaceId", this.namespaceId);
    params.put("serviceName", serviceName);
    params.put("groupName", groupName);
    params.put("clusterName", instance.getClusterName());
    params.put("ip", instance.getIp());
    params.put("port", String.valueOf(instance.getPort()));
    params.put("weight", String.valueOf(instance.getWeight()));
    params.put("enable", String.valueOf(instance.isEnabled()));
    params.put("healthy", String.valueOf(instance.isHealthy()));
    params.put("ephemeral", String.valueOf(instance.isEphemeral()));
    params.put("metadata", JSON.toJSONString(instance.getMetadata()));
    // 這里可以看出來，把上述服務實例的一些必要參數保存到一個Map中，通過OpenAPI的方式發送注冊請求
    this.reqAPI(UtilAndComs.NACOS_URL_INSTANCE, params, (String)"POST");
}

下面直接Debug走一遍。兩個前提（這里不再展開）：

啟動一個Nacos服務。
搞一個Maven項目，集成Nacos。

案例1：用Debug來理解Nacos服務注冊流程

1.項目初始化后，根據上文說法，會執行抽象類AbstractAutoServiceRegistration下面的onApplicationEvent（）方法，即事件被監聽到。

圖片

2.作為抽象類的子類實現NacosAutoServiceRegistration，監聽到Web服務啟動后，開始執行super.register（）方法。

圖片

3.執行NacosServiceRegistry下的register（）方法（super），前面說過，集成到SpringCloud中實現服務注冊的組件，都需要實現ServiceRegistry這個接口，而對于Nacos而言，NacosServiceRegistry就是具體的實現子類。執行注冊方法需要傳入的三個參數：

實例名稱serviceId。
實例歸屬的組。
具體實例

圖片

而registerInstance（）主要做兩件事：

檢查服務的健康（this.beatReactor.addBeatInfo（））。
執行服務的注冊（this.serverProxy.registerService（））。

圖片

服務健康的檢查：

圖片

檢查通過后，發送OpenAPI進行服務的注冊：

圖片

服務注冊小總結☆：

這里來做一個大框架式的梳理（也許前面寫的有點亂，這里通過幾個問答的形式來進行總結）

問題1：Nacos的服務注冊為什么和spring-cloud-commons這個包扯上關系？

回答：

1.首先，Nacos的服務注冊肯定少不了pom包：spring-cloud-starter-alibaba-nacos-discovery吧。

2.這個包下面包括了spring-cloud-commons包，那么這個包有什么用？

3.spring-cloud-commons中有一個接口叫做ServiceRegistry，而集成到SpringCloud中實現服務注冊的組件，都需要實現這個接口。

4.因此對于需要注冊到Nacos上的服務，也需要實現這個接口，那么具體的實現子類為NacosServiceRegistry。

問題2：為什么我的項目加了這幾個依賴，服務啟動時依舊沒有注冊到Nacos中？

回答：

1.本文提到過，進行Nacos服務注冊的時候，會有一個事件的監聽過程，而監聽的對象是WebServer，因此，這個項目需要是一個Web項目！

2.因此查看你的pom文件中是否有依賴：spring-boot-starter-web。

問題3：除此之外，spring-cloud-commons這個包還有什么作用？

回答：

1.這個包下的spring.factories文件中，配置了相關的服務注冊的置類，即支持其自動裝配。

2.這個配置類叫做AutoServiceRegistrationAutoConfiguration。其注入了類AutoServiceRegistration，而NacosAutoServiceRegistration是該類的一個具體實現。

3.當WebServer初始化的時候，通過綁定的事件監聽器，會實現監聽，執行服務的注冊邏輯。

說白了：

第一件事情：引入一個Spring監聽器，當容器初始化后，執行Nacos服務的注冊。
第二件事情：而Nacos服務注冊的方法的實現，其需要實現的接口來自于該包下的ServiceRegistry接口。

接下來就對Nacos注冊的流程進行一個總結：

服務（項目）啟動時，根據spring-cloud-commons中spring.factories的配置，自動裝配了類AutoServiceRegistrationAutoConfiguration。
AutoServiceRegistrationAutoConfiguration類中注入了類AutoServiceRegistration，其最終實現子類實現了Spring的監聽器。
根據監聽器，執行了服務注冊方法。而這個服務注冊方法則是調用了NacosServiceRegistry的register（）方法。
該方法主要調用的是Nacos Client SDK中的NamingService下的registerInstance（）方法完成服務的注冊。
registerInstance（）方法主要做兩件事：服務實例的健康監測和實例的注冊。
通過schedule（）方法定時的發送數據包，檢測實例的健康。
若健康監測通過，調用registerService（）方法，通過OpenAPI方式執行服務注冊，其中將實例Instance的相關信息存儲到HashMap中。

3.2 Nacos服務發現

有一點我們需要清楚：Nacos服務的發現發生在什么時候。例如：微服務發生遠程接口調用的時候。一般我們在使用OpenFeign進行遠程接口調用時，都需要用到對應的微服務名稱，而這個名稱就是用來進行服務發現的。

舉個例子：

@FeignClient("test-application")
public interface MyFeignService {
    @RequestMapping("getInfoById")
    R info(@PathVariable("id") Long id);
}

接下來直接開始講重點，Nacos在進行服務發現的時候，會調用NacosServerList類下的getServers()方法：

public class NacosServerList extends AbstractServerList<NacosServer> {
 private List<NacosServer> getServers() {
        try {
            String group = this.discoveryProperties.getGroup();
            // 1.通過唯一的serviceId（一般是服務名稱）和組來獲得對應的所有實例。
            List<Instance> instances = this.discoveryProperties.namingServiceInstance().selectInstances(this.serviceId, group, true);
            // 2.將List<Instance>轉換成List<NacosServer>數據，然后返回。
            return this.instancesToServerList(instances);
        } catch (Exception var3) {
            throw new IllegalStateException("Can not get service instances from nacos, serviceId=" + this.serviceId, var3);
        }
    }
}

接下來來看一下NacosNamingService.selectInstances（）方法：

public List<Instance> selectInstances(String serviceName, String groupName, boolean healthy) throws NacosException {
   return this.selectInstances(serviceName, groupName, healthy, true);
}

該方法最終會調用到其重載方法：

public List<Instance> selectInstances(String serviceName, String groupName, List<String> clusters, 
  boolean healthy, boolean subscribe) throws NacosException {
 // 保存服務實例信息的對象
    ServiceInfo serviceInfo;
    // 如果該消費者訂閱了這個服務，那么會在本地維護一個服務列表，服務從本地獲取
    if (subscribe) {
        serviceInfo = this.hostReactor.getServiceInfo(NamingUtils.getGroupedName(serviceName, groupName), StringUtils.join(clusters, ","));
    } else {
    // 否則實例會從服務中心進行獲取。
        serviceInfo = this.hostReactor.getServiceInfoDirectlyFromServer(NamingUtils.getGroupedName(serviceName, groupName), StringUtils.join(clusters, ","));
    }

    return this.selectInstances(serviceInfo, healthy);
}

這里應該重點關注this.hostReactor這個對象，它里面比較重要的是幾個Map類型的存儲結構：

public class HostReactor {
    private static final long DEFAULT_DELAY = 1000L;
    private static final long UPDATE_HOLD_INTERVAL = 5000L;
    // 存放線程異步調用的一個回調結果
    private final Map<String, ScheduledFuture<?>> futureMap;
    // 本地已存在的服務列表，key是服務名稱，value是ServiceInfo
    private Map<String, ServiceInfo> serviceInfoMap;
    // 待更新的實例列表
    private Map<String, Object> updatingMap;
    // 定時任務（負責服務列表的實時更新）
    private ScheduledExecutorService executor;
    ....
}

再看一看它的getServiceInfo（）方法：

public ServiceInfo getServiceInfo(String serviceName, String clusters) {
    LogUtils.NAMING_LOGGER.debug("failover-mode: " + this.failoverReactor.isFailoverSwitch());
    String key = ServiceInfo.getKey(serviceName, clusters);
    if (this.failoverReactor.isFailoverSwitch()) {
        return this.failoverReactor.getService(key);
    } else {
     // 1.先通過serverName即服務名獲得一個serviceInfo
        ServiceInfo serviceObj = this.getServiceInfo0(serviceName, clusters);
        // 如果沒有serviceInfo，則通過傳進來的參數new出一個新的serviceInfo對象，并且同時維護到本地Map和更新Map
        // 這里是serviceInfoMap和updatingMap
        if (null == serviceObj) {
            serviceObj = new ServiceInfo(serviceName, clusters);
            this.serviceInfoMap.put(serviceObj.getKey(), serviceObj);
            this.updatingMap.put(serviceName, new Object());
            // 2.updateServiceNow（），立刻去Nacos服務端拉去數據。
            this.updateServiceNow(serviceName, clusters);
            this.updatingMap.remove(serviceName);
        } else if (this.updatingMap.containsKey(serviceName)) {
            synchronized(serviceObj) {
                try {
                    serviceObj.wait(5000L);
                } catch (InterruptedException var8) {
                    LogUtils.NAMING_LOGGER.error("[getServiceInfo] serviceName:" + serviceName + ", clusters:" + clusters, var8);
                }
            }
        }
  // 3.定時更新實例信息
        this.scheduleUpdateIfAbsent(serviceName, clusters);
        // 最后返回服務實例數據（前面已經進行了更新）
        return (ServiceInfo)this.serviceInfoMap.get(serviceObj.getKey());
    }
}

來看下scheduleUpdateIfAbsent（）方法：

// 通過心跳的方式，每10秒去更新一次數據，并不是只有在調用服務的時候才會進行更新，而是通過定時任務來異步進行。
public void scheduleUpdateIfAbsent(String serviceName, String clusters) {
    if (this.futureMap.get(ServiceInfo.getKey(serviceName, clusters)) == null) {
        synchronized(this.futureMap) {
            if (this.futureMap.get(ServiceInfo.getKey(serviceName, clusters)) == null) {
             // 創建一個UpdateTask的更新線程任務，每10秒去異步更新集合數據
                ScheduledFuture<?> future = this.addTask(new HostReactor.UpdateTask(serviceName, clusters));
                this.futureMap.put(ServiceInfo.getKey(serviceName, clusters), future);
            }
        }
    }
}

案例2：用Debug來理解Nacos服務發現流程

1.進行遠程接口調用，觸發服務的發現，調用NacosServerList的getServers（）方法。傳入的serviceId和對應Feign接口上的接口@FeignClient中的名稱一致。

圖片

例如，我這里調用的Feign接口是：

@FeignClient("gulimall-member")
public interface MemberFeignService {
    @RequestMapping("/member/member/info/{id}")
    R info(@PathVariable("id") Long id);
}

這里可以看出來，返回的是一個Instance類型的List，對應的服務也發現并返回了。

圖片

2.這里則調用了NacosNamingService的selectInstances（）方法，我這里的subscribe值是true，即代表我這個消費者直接訂閱了這個服務，因此最終的信息是從本地Map中獲取，即Nacos維護了一個注冊列表。

圖片

3.再看下HostReactor的getServiceInfo（）方法：最終所需要的結果是從serviceInfoMap中獲取，并且通過多個Map進行維護服務實例，若存在數據的變化，還會通過強制睡眠5秒鐘的方式來等待數據的更新。

圖片

4.無論怎樣都會調用this.scheduleUpdateIfAbsent(serviceName, clusters)方法：

圖片

5.通過scheduleUpdateIfAbsent（）方法定時的獲取實時的實例數據，并且負責維護本地的服務注冊列表，若服務發生更新，則更新本地的服務數據。

圖片

服務發現小總結☆：

經常有人說過，Nacos有個好處，就是當一個服務掛了之后，短時間內不會造成影響，因為有個本地注冊列表，在服務不更新的情況下，服務還能夠正常的運轉，其原因如下：

Nacos的服務發現，一般是通過訂閱的形式來獲取服務數據。
而通過訂閱的方式，則是從本地的服務注冊列表中獲取（可以理解為緩存）。相反，如果不訂閱，那么服務的信息將會從Nacos服務端獲取，這時候就需要對應的服務是健康的。（宕機就不能使用了）
在代碼設計上，通過Map來存放實例數據，key為實例名稱，value為實例的相關信息數據（ServiceInfo對象）。

最后，服務發現的流程就是：

以調用遠程接口（OpenFeign）為例，當執行遠程調用時，需要經過服務發現的過程。
服務發現先執行NacosServerList類中的getServers()方法，根據遠程調用接口上@FeignClient中的屬性作為serviceId傳入NacosNamingService.selectInstances（）方法中進行調用。
根據subscribe的值來決定服務是從本地注冊列表中獲取還是從Nacos服務端中獲取。
以本地注冊列表為例，通過調用HostReactor.getServiceInfo（）來獲取服務的信息（serviceInfo），Nacos本地注冊列表由3個Map來共同維護。

本地Map–>serviceInfoMap，

更新Map–>updatingMap

異步更新結果Map–>futureMap,

最終的結果從serviceInfoMap當中獲取。

HostReactor類中的getServiceInfo（）方法通過this.scheduleUpdateIfAbsent() 方法和updateServiceNow（）方法實現服務的定時更新和立刻更新。
而對于scheduleUpdateIfAbsent（）方法，則通過線程池來進行異步的更新，將回調的結果（Future）保存到futureMap中，并且發生提交線程任務時，還負責更新本地注冊列表中的數據。

責任編輯：武曉燕來源：碼猿技術專欄