SpringCloud-Nacos注冊中心實作原理

一. Nacos介紹
二. Nacos注冊中心實作原理分析
- 2.1 Nacos架構圖
- 2.2 注冊中心的原理
三. Nacos原始碼分析
- 3.1 Nacos服務注冊
- - 案例1：用Debug來理解Nacos服務注冊流程
  - 服務注冊小總結☆：
- 3.2 Nacos服務發現
- - 案例2：用Debug來理解Nacos服務發現流程
  - 服務發現小總結☆：

一. Nacos介紹

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

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

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

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

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

服務發現和服務健康監測，

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

動態配置服務，

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

動態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中，并且發生提交執行緒任務時，還負責更新本地注冊串列中的資料，

轉載請註明出處，本文鏈接：https://www.uj5u.com/ruanti/251731.html

標籤：其他

上一篇：AWS云遷移工具方法匯總

下一篇：Tableau商業分析從新手到高手第二部分客戶主題分析