深入理解AP架構(gòu)Nacos注冊原理

2023-01-17 12:07:08來源：得物技術(shù)

1、Nacos簡介

Nacos是一款阿里巴巴開源用于管理分布式微服務(wù)的中間件，能夠幫助開發(fā)人員快速實現(xiàn)動態(tài)服務(wù)發(fā)現(xiàn)、服務(wù)配置、服務(wù)元數(shù)據(jù)及流量管理等。這篇文章主要剖析一下Nacos作為注冊中心時其服務(wù)注冊與發(fā)現(xiàn)原理。

【資料圖】

2、為什么會需要Nacos

Nacos作為注冊中心是為了更好更方便的管理應(yīng)用中的每一個服務(wù)，是各個分布式節(jié)點之間的紐帶。其作為注冊中心主要提供以下核心功能：

服務(wù)注冊與發(fā)現(xiàn)：動態(tài)的增減服務(wù)節(jié)點，服務(wù)節(jié)點增減后動態(tài)的通知服務(wù)消費者，不需要由消費者來更新配置。服務(wù)配置：動態(tài)修改服務(wù)配置，并將其推送到服務(wù)提供者和服務(wù)消費者而不需要重啟服務(wù)。健康檢查和服務(wù)摘除：主動的檢查服務(wù)健康情況，對于宕機的服務(wù)將其摘除服務(wù)列表。3、分布式架構(gòu)CAP理論

CAP定理是分布式系統(tǒng)中最基礎(chǔ)的原則，所以理解和掌握了CAP對系統(tǒng)架構(gòu)的設(shè)計至關(guān)重要。分布式架構(gòu)下所有系統(tǒng)不可能同時滿足以下三點：Consisteny（一致性）、Availability（可用性）、Partition tolerance（分區(qū)容錯性），CAP指明了任何分布式系統(tǒng)只能同時滿足這三項中的兩項。

分布式系統(tǒng)肯定都要保證其容錯性，那么可用性和一致性就只能選一個了。簡單來說分布式系統(tǒng)的CAP理論就像你想買個新手機，這個手機不可能功能強大、便宜、又好看的，它最多只能滿足兩點的，要么功能強大便宜、要么功能強大好看、要么便宜好看，不可能同時滿足三點。

4、幾種注冊中心的區(qū)別

注冊中心在分布式應(yīng)用中是經(jīng)常用到的，也是必不可少的，那注冊中心，又分為以下幾種：Eureka、Zookeeper、Nacos等。這些注冊中心最大的區(qū)別就是其基于AP架構(gòu)還是CP架構(gòu)，簡單介紹一下：

Zookeeper：用過或者了解過zk做注冊中心的同學(xué)都知道，Zookeeper集群下一旦leader節(jié)點宕機了，在短時間內(nèi)服務(wù)都不可通訊，因為它們在一定時間內(nèi)follower進行選舉來推出新的leader，因為在這段時間內(nèi)，所有的服務(wù)通信將受到影響，而且leader選取時間比較長，需要花費幾十秒甚至上百秒的時間，因此：可以理解為 Zookeeper是實現(xiàn)的CP，也就是將失去A（可用性）。Eureka：Eureka集群下每個節(jié)點之間都會定時發(fā)送心跳，定時同步數(shù)據(jù)，沒有master/slave之分，是一個完全去中心化的架構(gòu)。因此每個注冊到Eureka下的實例都會定時同步ip，服務(wù)之間的調(diào)用也是根據(jù)Eureka拿到的緩存服務(wù)數(shù)據(jù)進行調(diào)用。若一臺Eureka服務(wù)宕機，其他Eureka在一定時間內(nèi)未感知到這臺Eureka服務(wù)宕機，各個服務(wù)之間還是可以正常調(diào)用。Eureka的集群中，只要有一臺Eureka還在，就能保證注冊服務(wù)可用（保證可用性），只不過查到的信息可能不是最新的（不保證強一致性）。當(dāng)數(shù)據(jù)出現(xiàn)不一致時，雖然A, B上的注冊信息不完全相同，但每個Eureka節(jié)點依然能夠正常對外提供服務(wù)，這會出現(xiàn)查詢服務(wù)信息時如果請求A查不到，但請求B就能查到。如此保證了可用性但犧牲了一致性。Nacos：同時支持CP和AP架構(gòu)，根據(jù)根據(jù)服務(wù)注冊選擇臨時和永久來決定走AP模式還是CP模式。如果注冊Nacos的client節(jié)點注冊時ephemeral=true，那么Nacos集群對這個client節(jié)點的效果就是AP，采用distro協(xié)議實現(xiàn)；而注冊Nacos的client節(jié)點注冊時ephemeral=false，那么Nacos集群對這個節(jié)點的效果就是CP的，采用raft協(xié)議實現(xiàn)。

本篇文章主要是深入研究一下Nacos基于AP架構(gòu)微服務(wù)注冊原理，由于篇幅有限基于CP架構(gòu)的Nacos微服務(wù)注冊下次再跟你們分析。

5、Nacos服務(wù)注冊與發(fā)現(xiàn)的原理

1.微服務(wù)在啟動將自己的服務(wù)注冊到Nacos注冊中心，同時發(fā)布http接口供其他系統(tǒng)調(diào)用，一般都是基于SpringMVC。

2.服務(wù)消費者基于Feign調(diào)用服務(wù)提供者對外發(fā)布的接口，先對調(diào)用的本地接口加上注解@FeignClient，F(xiàn)eign會針對加了該注解的接口生成動態(tài)代理，服務(wù)消費者針對Feign生成的動態(tài)代理去調(diào)用方法時，會在底層生成Http協(xié)議格式的請求，類似 /stock/deduct? productId=100。

3.Feign最終會調(diào)用Ribbon從本地的Nacos注冊表的緩存里根據(jù)服務(wù)名取出服務(wù)提供在機器的列表，然后進行負載均衡并選擇一臺機器出來，對選出來的機器IP和端口拼接之前生成的url請求，生成調(diào)用的Http接口地址。

6、Nacos核心功能點

服務(wù)注冊：Nacos Client會通過發(fā)送REST請求的方式向Nacos Server注冊自己的服務(wù)，提供自身的元數(shù)據(jù)，比如ip地址、端口等信息。Nacos Server接收到注冊請求后，就會把這些元數(shù)據(jù)信息存儲在一個雙層的內(nèi)存Map中。

服務(wù)心跳：在服務(wù)注冊后，Nacos Client會維護一個定時心跳來持續(xù)通知Nacos Server，說明服務(wù)一直處于可用狀態(tài)，防止被剔除。默認5s發(fā)送一次心跳。

服務(wù)健康檢查：Nacos Server會開啟一個定時任務(wù)用來檢查注冊服務(wù)實例的健康情況，對于超過15s沒有收到客戶端心跳的實例會將它的healthy屬性置為false(客戶端服務(wù)發(fā)現(xiàn)時不會發(fā)現(xiàn))，如果某個實例超過30秒沒有收到心跳，直接剔除該實例(被剔除的實例如果恢復(fù) 發(fā)送心跳則會重新注冊)

服務(wù)發(fā)現(xiàn)：服務(wù)消費者（Nacos Client）在調(diào)用服務(wù)提供者的服務(wù)時，會發(fā)送一個REST請求給Nacos Server，獲取上面注冊的服務(wù)清單，并且緩存在Nacos Client本地，同時會在Nacos Client本地開啟一個定時任務(wù)定時拉取服務(wù)端最新的注冊表信息更新到本地緩存

服務(wù)同步：Nacos Server集群之間會互相同步服務(wù)實例，用來保證服務(wù)信息的一致性。

7、Nacos源碼分析

看Nacos源碼的不難發(fā)現(xiàn)，Nacos實際上就是一個基于Spring Boot的web應(yīng)用，不管是服務(wù)注冊還是發(fā)送心跳都是通過給Nacos服務(wù)端發(fā)送http請求實現(xiàn)的。下載并編譯Nacos源碼就不過多贅述了，首先需要搭建一個微服務(wù)作為Nacos的客戶端。

7.1 Nacos客戶端注冊

Nacos客戶端也是個Spring Boot項目，當(dāng)客戶端服務(wù)啟動時Spring Boot項目啟動時自動加載spring-cloud-starter-alibaba-nacos-discovery包的META-INF/spring.factories中包含自動裝配的配置信息，并將文件中的類加載成bean放入Spring容器中，我們可以先看一下spring.factories文件：

org.springframework.boot.autoconfigure.EnableAutoCnotallow=\  com.alibaba.cloud.nacos.discovery.NacosDiscoveryAutoConfiguration,\  com.alibaba.cloud.nacos.endpoint.NacosDiscoveryEndpointAutoConfiguration,\  com.alibaba.cloud.nacos.registry.NacosServiceRegistryAutoConfiguration,\  com.alibaba.cloud.nacos.discovery.NacosDiscoveryClientConfiguration,\  com.alibaba.cloud.nacos.discovery.reactive.NacosReactiveDiscoveryClientConfiguration,\  com.alibaba.cloud.nacos.discovery.configclient.NacosConfigServerAutoConfiguration,\  com.alibaba.cloud.nacos.NacosServiceAutoConfigurationorg.springframework.cloud.bootstrap.BootstrapCnotallow=\  com.alibaba.cloud.nacos.discovery.configclient.NacosDiscoveryClientConfigServiceBootstrapConfiguration

找到Nacos注冊中心的自動配置類：NacosServiceRegistryAutoConfiguration。

NacosServiceRegistryAutoConfiguration這個類是Nacos客戶端啟動時的一個入口類，代碼如下：

@Configuration(    proxyBeanMethods = false)@EnableConfigurationProperties@ConditionalOnNacosDiscoveryEnabled@ConditionalOnProperty(    value = {"spring.cloud.service-registry.auto-registration.enabled"},    matchIfMissing = true)@AutoConfigureAfter({AutoServiceRegistrationConfiguration.class,                     AutoServiceRegistrationAutoConfiguration.class,                      NacosDiscoveryAutoConfiguration.class})public class NacosServiceRegistryAutoConfiguration {    public NacosServiceRegistryAutoConfiguration() {    }    @Bean    public NacosServiceRegistry nacosServiceRegistry(NacosDiscoveryProperties nacosDiscoveryProperties) {        return new NacosServiceRegistry(nacosDiscoveryProperties);    }    @Bean    @ConditionalOnBean({AutoServiceRegistrationProperties.class})    public NacosRegistration nacosRegistration(ObjectProvider> registrationCustomizers, NacosDiscoveryProperties nacosDiscoveryProperties, ApplicationContext context) {        return new NacosRegistration((List)registrationCustomizers.getIfAvailable(), nacosDiscoveryProperties, context);    }    @Bean    @ConditionalOnBean({AutoServiceRegistrationProperties.class})    public NacosAutoServiceRegistration nacosAutoServiceRegistration(NacosServiceRegistry registry, AutoServiceRegistrationProperties autoServiceRegistrationProperties, NacosRegistration registration) {        return new NacosAutoServiceRegistration(registry, autoServiceRegistrationProperties, registration);    }}

看NacosServiceRegistryAutoConfiguration配置類有3個@Bean注解。

nacosServiceRegistry()方法: 定義了NacosServiceRegistry的bean，并且為其屬性nacosDiscoveryProperties賦值，即將從配置文件中讀取到的配置信息賦值進去待用；nacosRegistration()方法主要就是定義了NacosRegistration的bean，后面會用到這個bean；nacosAutoServiceRegistration：該方法比較核心它的參數(shù)中有2個就是前面定義的兩個bean，其實就是為了這個方法服務(wù)的，由NacosAutoServiceRegistration類的構(gòu)造器傳入NacosAutoServiceRegistration類中：NacosAutoServiceRegistration(registry, autoServiceRegistrationProperties, registration)，后面的流程都是以這句代碼作為入口。

利用IDEA查看類結(jié)構(gòu)，如上圖所示，NacosAutoServiceRegistration繼承AbstractAutoServiceRegistration類，而AbstractAutoServiceRegistration類又實現(xiàn)了AutoServiceRegistration和ApplicationListener接口。

ApplicationListener接口是Spring提供的事件監(jiān)聽接口，Spring會在所有bean都初始化完成之后發(fā)布一個事件，ApplicationListener會監(jiān)聽所發(fā)布的事件，這里的事件是Spring Boot自定義的WebServerInitializedEvent事件，主要是項目啟動時就會發(fā)布WebServerInitializedEvent事件，然后被AbstractAutoServiceRegistration監(jiān)聽到，從而就會執(zhí)行onApplicationEvent方法，在這個方法里就會進行服務(wù)注冊。

這里AbstractAutoServiceRegistration類實現(xiàn)了Spring監(jiān)聽器接口ApplicationListener，并重寫了該接口的onApplicationEvent方法。

public void onApplicationEvent(WebServerInitializedEvent event) {     this.bind(event);}

繼續(xù)點下去看bind方法。

public void bind(WebServerInitializedEvent event) {        ApplicationContext context = event.getApplicationContext();        if (!(context instanceof ConfigurableWebServerApplicationContext) || !"management".equals(((ConfigurableWebServerApplicationContext)context).getServerNamespace())) {            this.port.compareAndSet(0, event.getWebServer().getPort());            //start方法            this.start();        }    }

看到這里發(fā)現(xiàn)了bind方法里有個非常重要的start()方法，繼續(xù)看該方法的register()就是真正的客戶端注冊方法。

public void start() {        if (!this.isEnabled()) {            if (logger.isDebugEnabled()) {                logger.debug("Discovery Lifecycle disabled. Not starting");            }        } else {            if (!this.running.get()) {                this.context.publishEvent(new InstancePreRegisteredEvent(this, this.getRegistration()));                //真正的客戶端注冊方法                this.register();                if (this.shouldRegisterManagement()) {                    this.registerManagement();                }                this.context.publishEvent(new InstanceRegisteredEvent(this, this.getConfiguration()));                this.running.compareAndSet(false, true);            }        }    }

跳過一些中間非關(guān)鍵性的代碼，可以直接看該注冊方法。

protected void register() {   this.serviceRegistry.register(getRegistration());}

這里的serviceRegistry就是NacosServiceRegistryAutoConfiguration類中第一個@Bean定義的bean，第一個@Bean就是這里的serviceRegistry對象的實現(xiàn)；其中g(shù)etRegistration()獲取的就是第二個@Bean定義的NacosRegistration的實例，這兩個bean實例都是通過第3個@Bean傳進來的，所以這里就可以把NacosServiceRegistryAutoConfiguration類中那3個@Bean給串起來了。

protected void register() {   this.serviceRegistry.register(getRegistration());}

不得不說，阿里巴巴開發(fā)的中間件，其底層源碼的命名還是很規(guī)范的，register()方法從命名上來看就可以知道這是注冊的方法，事實也確實是注冊的方法，這個方法中會通過nacos-client包來調(diào)用nacos-server的服務(wù)注冊接口來實現(xiàn)服務(wù)的注冊功能。下面我看一下調(diào)用Nacos注冊接口方法：

public void register(Registration registration) {        if (StringUtils.isEmpty(registration.getServiceId())) {            log.warn("No service to register for nacos client...");        } else {            NamingService namingService = this.namingService();            String serviceId = registration.getServiceId();            String group = this.nacosDiscoveryProperties.getGroup();            //構(gòu)建客戶端參數(shù)ip，端口號等            Instance instance = this.getNacosInstanceFromRegistration(registration);            try {                //調(diào)用注冊方法                namingService.registerInstance(serviceId, group, instance);                log.info("nacos registry, {} {} {}:{} register finished", new Object[]{group, serviceId, instance.getIp(), instance.getPort()});            } catch (Exception var7) {                log.error("nacos registry, {} register failed...{},", new Object[]{serviceId, registration.toString(), var7});                ReflectionUtils.rethrowRuntimeException(var7);            }        }    }        //構(gòu)建客戶端注冊參數(shù)    private Instance getNacosInstanceFromRegistration(Registration registration) {        Instance instance = new Instance();        instance.setIp(registration.getHost());        instance.setPort(registration.getPort());        instance.setWeight((double)this.nacosDiscoveryProperties.getWeight());        instance.setClusterName(this.nacosDiscoveryProperties.getClusterName());        instance.setEnabled(this.nacosDiscoveryProperties.isInstanceEnabled());        instance.setMetadata(registration.getMetadata());        instance.setEphemeral(this.nacosDiscoveryProperties.isEphemeral());        return instance;    }

根據(jù)源碼可以知道beatReactor.addBeatInfo()方法作用在于創(chuàng)建心跳信息實現(xiàn)健康檢測，Nacos 服務(wù)端必須要確保注冊的服務(wù)實例是健康的,而心跳檢測就是服務(wù)健康檢測的手段。而serverProxy.registerService()實現(xiàn)服務(wù)注冊，綜上可以分析出Nacos客戶端注冊流程：

到此為止還沒有真正的實現(xiàn)服務(wù)的注冊，但是至少已經(jīng)知道了Nacos客戶端的自動注冊原理是借助了Spring Boot的自動配置功能，在項目啟動時通過自動配置類。NacosServiceRegistryAutoConfiguration將NacosServiceRegistry注入進來，通過Spring的事件監(jiān)聽機制，調(diào)用該類的注冊方法register(registration)實現(xiàn)服務(wù)的自動注冊。

7.2 Nacos服務(wù)發(fā)現(xiàn)7.2.1 Nacos客戶端客戶端服務(wù)發(fā)現(xiàn)

當(dāng)Nacos服務(wù)端啟動后，會先從本地緩存的serviceInfoMap中獲取服務(wù)實例信息，獲取不到則通過NamingProxy調(diào)用Nacos服務(wù)端獲取服務(wù)實例信息，最后開啟定時任務(wù)每秒請求服務(wù)端獲取實例信息列表進而更新本地緩存serviceInfoMap，服務(wù)發(fā)現(xiàn)拉取實例信息流程圖如下：

廢話不多說，直接上服務(wù)發(fā)現(xiàn)源碼：

/**     * 客戶端服務(wù)發(fā)現(xiàn)     *     * @param serviceName name of service     * @param groupName   group of service     * @param clusters    list of cluster     * @param subscribe   if subscribe the service     * @return     * @throws NacosException     */    @Override    public List getAllInstances(String serviceName, String groupName, List clusters,            boolean subscribe) throws NacosException {                ServiceInfo serviceInfo;        if (subscribe) {            // 如果本地緩存不存在服務(wù)信息，則進行訂閱            serviceInfo = hostReactor.getServiceInfo(NamingUtils.getGroupedName(serviceName, groupName),                    StringUtils.join(clusters, ","));        } else {            // 如果非訂閱模式就直接拉取服務(wù)端的注冊表            serviceInfo = hostReactor                    .getServiceInfoDirectlyFromServer(NamingUtils.getGroupedName(serviceName, groupName),                            StringUtils.join(clusters, ","));        }        List list;        if (serviceInfo == null || CollectionUtils.isEmpty(list = serviceInfo.getHosts())) {            return new ArrayList();        }        return list;    }

/**     * 客戶端從注冊中心拉取注冊列表     *     * @param serviceName     * @param clusters     * @return     */    public ServiceInfo getServiceInfo(final String serviceName, final String clusters) {                NAMING_LOGGER.debug("failover-mode: " + failoverReactor.isFailoverSwitch());        String key = ServiceInfo.getKey(serviceName, clusters);        if (failoverReactor.isFailoverSwitch()) {            return failoverReactor.getService(key);        }        //客戶端從本地緩存中拉群注冊表信息，第一次根據(jù)服務(wù)名從注冊表map中獲取，服務(wù)表信息肯定是為null        ServiceInfo serviceObj = getServiceInfo0(serviceName, clusters);        //如果拿到緩存map中的服務(wù)列表為null，如果是第一次根據(jù)服務(wù)名拉取注冊表信息，肯定為null        if (null == serviceObj) {            serviceObj = new ServiceInfo(serviceName, clusters);                        serviceInfoMap.put(serviceObj.getKey(), serviceObj);                        updatingMap.put(serviceName, new Object());            //第一次拉取注冊表信息為null后，然后調(diào)用Nacos服務(wù)端接口更新本地注冊表            updateServiceNow(serviceName, clusters);            updatingMap.remove(serviceName);                    } else if (updatingMap.containsKey(serviceName)) {                        if (UPDATE_HOLD_INTERVAL > 0) {                // hold a moment waiting for update finish                synchronized (serviceObj) {                    try {                        serviceObj.wait(UPDATE_HOLD_INTERVAL);                    } catch (InterruptedException e) {                        NAMING_LOGGER                                .error("[getServiceInfo] serviceName:" + serviceName + ", clusters:" + clusters, e);                    }                }            }        }        /**         * 定時任務(wù)拉取，每隔幾秒鐘就去拉取一次，去拉取nacos注冊表，更新客戶端本地注冊列表的map         *         * 為啥這里要定時任務(wù)拉取呢？因為上面到注冊表map是緩存在客戶端本地的，假如有新的服務(wù)注冊到nacos         * 時，這時就要更新客戶端注冊表信息，所以這里會執(zhí)行一個訂單拉取的任務(wù)         */        scheduleUpdateIfAbsent(serviceName, clusters);                return serviceInfoMap.get(serviceObj.getKey());    }                //異步拉取任務(wù)    public void scheduleUpdateIfAbsent(String serviceName, String clusters) {        if (futureMap.get(ServiceInfo.getKey(serviceName, clusters)) != null) {            return;        }                synchronized (futureMap) {            if (futureMap.get(ServiceInfo.getKey(serviceName, clusters)) != null) {                return;            }            //執(zhí)行一個定時拉取任務(wù)            ScheduledFuture future = addTask(new UpdateTask(serviceName, clusters));            futureMap.put(ServiceInfo.getKey(serviceName, clusters), future);        }    }        //定時拉取注冊表任務(wù)        public class UpdateTask implements Runnable {                long lastRefTime = Long.MAX_VALUE;                private final String clusters;                private final String serviceName;                /**         * the fail situation. 1:can"t connect to server 2:serviceInfo"s hosts is empty         */        private int failCount = 0;                public UpdateTask(String serviceName, String clusters) {            this.serviceName = serviceName;            this.clusters = clusters;        }                private void incFailCount() {            int limit = 6;            if (failCount == limit) {                return;            }            failCount++;        }                private void resetFailCount() {            failCount = 0;        }                @Override        public void run() {            long delayTime = DEFAULT_DELAY;                        try {                ServiceInfo serviceObj = serviceInfoMap.get(ServiceInfo.getKey(serviceName, clusters));                                if (serviceObj == null) {                    //又在繼續(xù)調(diào)用拉取nacos注冊列表方法                    updateService(serviceName, clusters);                    return;                }                                if (serviceObj.getLastRefTime() <= lastRefTime) {                    //又在繼續(xù)調(diào)用拉取nacos注冊列表方法                    updateService(serviceName, clusters);                    serviceObj = serviceInfoMap.get(ServiceInfo.getKey(serviceName, clusters));                } else {                    // if serviceName already updated by push, we should not override it                    // since the push data may be different from pull through force push                    refreshOnly(serviceName, clusters);                }                                lastRefTime = serviceObj.getLastRefTime();                                if (!notifier.isSubscribed(serviceName, clusters) && !futureMap                        .containsKey(ServiceInfo.getKey(serviceName, clusters))) {                    // abort the update task                    NAMING_LOGGER.info("update task is stopped, service:" + serviceName + ", clusters:" + clusters);                    return;                }                if (CollectionUtils.isEmpty(serviceObj.getHosts())) {                    incFailCount();                    return;                }                delayTime = serviceObj.getCacheMillis();                resetFailCount();            } catch (Throwable e) {                incFailCount();                NAMING_LOGGER.warn("[NA] failed to update serviceName: " + serviceName, e);            } finally {                //最后繼續(xù)嵌套調(diào)用當(dāng)前這個任務(wù)，實現(xiàn)定時拉取                executor.schedule(this, Math.min(delayTime << failCount, DEFAULT_DELAY * 60), TimeUnit.MILLISECONDS);            }        }

這里值得注意的是，Nacos客戶端拉取注冊列表方法的最后又是一個定時任務(wù)任務(wù)，每隔10秒鐘就會拉取一次服務(wù)端Nacos的注冊表。為啥這里要定時任務(wù)拉取呢？因為上面到注冊表map是緩存在客戶端本地的，假如有新的服務(wù)注冊到Nacos時，這時就要更新客戶端注冊表信息，所以這里會執(zhí)行一個拉取的任務(wù)。

private void updateServiceNow(String serviceName, String clusters) {        try {            //拉群nacos列表，更新到本地緩存map中的注冊列表            updateService(serviceName, clusters);        } catch (NacosException e) {            NAMING_LOGGER.error("[NA] failed to update serviceName: " + serviceName, e);        }    }    /**     * Update service now.     * 拉取注冊列表     *     * @param serviceName service name     * @param clusters    clusters     */    public void updateService(String serviceName, String clusters) throws NacosException {        ServiceInfo oldService = getServiceInfo0(serviceName, clusters);        try {            //調(diào)用拉群列表接口            String result = serverProxy.queryList(serviceName, clusters, pushReceiver.getUdpPort(), false);                        if (StringUtils.isNotEmpty(result)) {                //解析返回值服務(wù)表json                processServiceJson(result);            }        } finally {            if (oldService != null) {                synchronized (oldService) {                    oldService.notifyAll();                }            }    }    /**     * Nacos客戶端查詢服務(wù)端注冊表數(shù)     *     * @param serviceName service name     * @param clusters    clusters     * @param udpPort     udp port     * @param healthyOnly healthy only     * @return instance list     * @throws NacosException nacos exception     */    public String queryList(String serviceName, String clusters, int udpPort, boolean healthyOnly)            throws NacosException {                final Map params = new HashMap(8);        params.put(CommonParams.NAMESPACE_ID, namespaceId);        params.put(CommonParams.SERVICE_NAME, serviceName);        params.put("clusters", clusters);        params.put("udpPort", String.valueOf(udpPort));        params.put("clientIP", NetUtils.localIP());        params.put("healthyOnly", String.valueOf(healthyOnly));        //調(diào)用拉取注冊列表接口        return reqApi(UtilAndComs.nacosUrlBase + "/instance/list", params, HttpMethod.GET);    }

7.2.2 服務(wù)端服務(wù)發(fā)現(xiàn)查詢注冊表api

上面分析了當(dāng)客戶端在其本地緩存中沒有找到注冊表信息，就會調(diào)用Nacos服務(wù)端api拉取注冊表信息，不難發(fā)現(xiàn)服務(wù)端查詢注冊表api為"/instance/list"。

/**     * Get all instance of input service.     * 客戶端獲取nacos所有注冊實例方法     *     * @param request http request     * @return list of instance     * @throws Exception any error during list     */    @GetMapping("/list")    @Secured(parser = NamingResourceParser.class, action = ActionTypes.READ)    public ObjectNode list(HttpServletRequest request) throws Exception {                String namespaceId = WebUtils.optional(request, CommonParams.NAMESPACE_ID, Constants.DEFAULT_NAMESPACE_ID);        String serviceName = WebUtils.required(request, CommonParams.SERVICE_NAME);        NamingUtils.checkServiceNameFormat(serviceName);                String agent = WebUtils.getUserAgent(request);        String clusters = WebUtils.optional(request, "clusters", StringUtils.EMPTY);        String clientIP = WebUtils.optional(request, "clientIP", StringUtils.EMPTY);        int udpPort = Integer.parseInt(WebUtils.optional(request, "udpPort", "0"));        String env = WebUtils.optional(request, "env", StringUtils.EMPTY);        boolean isCheck = Boolean.parseBoolean(WebUtils.optional(request, "isCheck", "false"));                String app = WebUtils.optional(request, "app", StringUtils.EMPTY);                String tenant = WebUtils.optional(request, "tid", StringUtils.EMPTY);                boolean healthyOnly = Boolean.parseBoolean(WebUtils.optional(request, "healthyOnly", "false"));                return doSrvIpxt(namespaceId, serviceName, agent, clusters, clientIP, udpPort, env, isCheck, app, tenant,                healthyOnly);    }

這里通過doSrvIpxt()方法獲取服務(wù)列表，根據(jù)namespaceId、serviceName獲取service實例，service實例中srvIPs獲取所有服務(wù)提供者的實例信息，遍歷組裝成json字符串并返回。

public ObjectNode doSrvIpxt(String namespaceId, String serviceName, String agent, String clusters, String clientIP,            int udpPort, String env, boolean isCheck, String app, String tid, boolean healthyOnly) throws Exception {                ClientInfo clientInfo = new ClientInfo(agent);        ObjectNode result = JacksonUtils.createEmptyJsonNode();        Service service = serviceManager.getService(namespaceId, serviceName);        long cacheMillis = switchDomain.getDefaultCacheMillis();                // now try to enable the push        try {            if (udpPort > 0 && pushService.canEnablePush(agent)) {                                pushService                        .addClient(namespaceId, serviceName, clusters, agent, new InetSocketAddress(clientIP, udpPort),                                pushDataSource, tid, app);                cacheMillis = switchDomain.getPushCacheMillis(serviceName);            }        } catch (Exception e) {            Loggers.SRV_LOG                    .error("[NACOS-API] failed to added push client {}, {}:{}", clientInfo, clientIP, udpPort, e);            cacheMillis = switchDomain.getDefaultCacheMillis();        }                if (service == null) {            if (Loggers.SRV_LOG.isDebugEnabled()) {                Loggers.SRV_LOG.debug("no instance to serve for service: {}", serviceName);            }            result.put("name", serviceName);            result.put("clusters", clusters);            result.put("cacheMillis", cacheMillis);            result.replace("hosts", JacksonUtils.createEmptyArrayNode());            return result;        }                checkIfDisabled(service);                List srvedIPs;        //獲取所有實例        srvedIPs = service.srvIPs(Arrays.asList(StringUtils.split(clusters, ",")));                // filter ips using selector:        if (service.getSelector() != null && StringUtils.isNotBlank(clientIP)) {            srvedIPs = service.getSelector().select(clientIP, srvedIPs);        }                if (CollectionUtils.isEmpty(srvedIPs)) {                        if (Loggers.SRV_LOG.isDebugEnabled()) {                Loggers.SRV_LOG.debug("no instance to serve for service: {}", serviceName);            }                        if (clientInfo.type == ClientInfo.ClientType.JAVA                    && clientInfo.version.compareTo(VersionUtil.parseVersion("1.0.0")) >= 0) {                result.put("dom", serviceName);            } else {                result.put("dom", NamingUtils.getServiceName(serviceName));            }                        result.put("name", serviceName);            result.put("cacheMillis", cacheMillis);            result.put("lastRefTime", System.currentTimeMillis());            result.put("checksum", service.getChecksum());            result.put("useSpecifiedURL", false);            result.put("clusters", clusters);            result.put("env", env);            result.set("hosts", JacksonUtils.createEmptyArrayNode());            result.set("metadata", JacksonUtils.transferToJsonNode(service.getMetadata()));            return result;        }                Map> ipMap = new HashMap<>(2);        ipMap.put(Boolean.TRUE, new ArrayList<>());        ipMap.put(Boolean.FALSE, new ArrayList<>());                for (Instance ip : srvedIPs) {            ipMap.get(ip.isHealthy()).add(ip);        }                if (isCheck) {            result.put("reachProtectThreshold", false);        }                double threshold = service.getProtectThreshold();                if ((float) ipMap.get(Boolean.TRUE).size() / srvedIPs.size() <= threshold) {                        Loggers.SRV_LOG.warn("protect threshold reached, return all ips, service: {}", serviceName);            if (isCheck) {                result.put("reachProtectThreshold", true);            }                        ipMap.get(Boolean.TRUE).addAll(ipMap.get(Boolean.FALSE));            ipMap.get(Boolean.FALSE).clear();        }                if (isCheck) {            result.put("protectThreshold", service.getProtectThreshold());            result.put("reachLocalSiteCallThreshold", false);                        return JacksonUtils.createEmptyJsonNode();        }                ArrayNode hosts = JacksonUtils.createEmptyArrayNode();                for (Map.Entry> entry : ipMap.entrySet()) {            List ips = entry.getValue();                        if (healthyOnly && !entry.getKey()) {                continue;            }                        for (Instance instance : ips) {                                // remove disabled instance:                if (!instance.isEnabled()) {                    continue;                }                                ObjectNode ipObj = JacksonUtils.createEmptyJsonNode();                                ipObj.put("ip", instance.getIp());                ipObj.put("port", instance.getPort());                // deprecated since nacos 1.0.0:                ipObj.put("valid", entry.getKey());                ipObj.put("healthy", entry.getKey());                ipObj.put("marked", instance.isMarked());                ipObj.put("instanceId", instance.getInstanceId());                ipObj.set("metadata", JacksonUtils.transferToJsonNode(instance.getMetadata()));                ipObj.put("enabled", instance.isEnabled());                ipObj.put("weight", instance.getWeight());                ipObj.put("clusterName", instance.getClusterName());                if (clientInfo.type == ClientInfo.ClientType.JAVA                        && clientInfo.version.compareTo(VersionUtil.parseVersion("1.0.0")) >= 0) {                    ipObj.put("serviceName", instance.getServiceName());                } else {                    ipObj.put("serviceName", NamingUtils.getServiceName(instance.getServiceName()));                }                                ipObj.put("ephemeral", instance.isEphemeral());                hosts.add(ipObj);                            }        }                result.replace("hosts", hosts);        if (clientInfo.type == ClientInfo.ClientType.JAVA                && clientInfo.version.compareTo(VersionUtil.parseVersion("1.0.0")) >= 0) {            result.put("dom", serviceName);        } else {            result.put("dom", NamingUtils.getServiceName(serviceName));        }        result.put("name", serviceName);        result.put("cacheMillis", cacheMillis);        result.put("lastRefTime", System.currentTimeMillis());        result.put("checksum", service.getChecksum());        result.put("useSpecifiedURL", false);        result.put("clusters", clusters);        result.put("env", env);        result.replace("metadata", JacksonUtils.transferToJsonNode(service.getMetadata()));        return result;    }

最后看一下獲取服務(wù)端實例方法，最后就是將臨時實例或者持久實例放在一個集合中返回給客戶端。

public List srvIPs(List clusters) {        if (CollectionUtils.isEmpty(clusters)) {            clusters = new ArrayList<>();            clusters.addAll(clusterMap.keySet());        }        return allIPs(clusters);    }    public List allIPs(List clusters) {        List result = new ArrayList<>();        for (String cluster : clusters) {            Cluster clusterObj = clusterMap.get(cluster);            if (clusterObj == null) {                continue;            }                        result.addAll(clusterObj.allIPs());        }        return result;    }    public List allIPs() {        List allInstances = new ArrayList<>();        //將nacos內(nèi)存中注冊表信息返回        allInstances.addAll(persistentInstances);        allInstances.addAll(ephemeralInstances);        return allInstances;    }

總結(jié)一下Nacos客戶端服務(wù)發(fā)現(xiàn)的核心流程：

如果沒有開啟訂閱模式，則直接通過調(diào)用/instance/list接口獲取服務(wù)實例列表信息；

如果開啟訂閱模式，則先會從本地緩存中獲取實例信息，如果不存在，則進行訂閱獲并獲取實例信息；在獲得最新的實例信息之后，也會執(zhí)行processServiceJson(result)方法來更新內(nèi)存和本地實例緩存，并發(fā)布變更時間。

開啟訂閱時，會開啟定時任務(wù)，定時執(zhí)行UpdateTask獲取服務(wù)器實例信息、更新本地緩存、發(fā)布事件等；

7.3 Nacos服務(wù)端注冊

服務(wù)端的注冊源碼邏輯相對客戶端的還是要復(fù)雜很多，所以這里我們先看一下Nacos服務(wù)端注冊的完整流程圖，避免一上來就看源碼被繞暈。

接下來我們就著重分析一下AP架構(gòu)Nacos服務(wù)注冊的源碼。

7.3.1 Nacos服務(wù)端注冊

Nacos服務(wù)端注冊當(dāng)然是本文的核心，那么首先我們來看一下Nacos服務(wù)端注冊源碼。從Nacos的客戶端注冊原理不難發(fā)現(xiàn)，客戶端通過調(diào)用Nacos服務(wù)端提供的http接口實現(xiàn)注冊，對外提供的服務(wù)接口請求地址為nacos/v1/ns/instance，實現(xiàn)代碼咋nacos-naming模塊下的InstanceController類中：

@CanDistro    @PostMapping    @Secured(parser = NamingResourceParser.class, action = ActionTypes.WRITE)    public String register(HttpServletRequest request) throws Exception {        //從請求參數(shù)匯總獲得namespaceId（命名空間Id）        final String namespaceId = WebUtils                .optional(request, CommonParams.NAMESPACE_ID, Constants.DEFAULT_NAMESPACE_ID);        //從請求參數(shù)匯總獲得serviceName（服務(wù)名）        final String serviceName = WebUtils.required(request, CommonParams.SERVICE_NAME);        NamingUtils.checkServiceNameFormat(serviceName);                final Instance instance = parseInstance(request);        //registerInstance注冊實例        serviceManager.registerInstance(namespaceId, serviceName, instance);        return "ok";    }

客戶端就是通過調(diào)用該api實現(xiàn)Nacos的注冊的，下面可以看一下Nacos的這個注冊api是怎么實現(xiàn)的。

public void registerInstance(String namespaceId, String serviceName, Instance instance) throws NacosException {                createEmptyService(namespaceId, serviceName, instance.isEphemeral());        //前面構(gòu)建過了，這里調(diào)取肯定部不為null，從serviceMap中根據(jù)namespaceId和serviceName得到一個服務(wù)對象        Service service = getService(namespaceId, serviceName);                if (service == null) {            throw new NacosException(NacosException.INVALID_PARAM,                    "service not found, namespace: " + namespaceId + ", service: " + serviceName);        }        //調(diào)用addInstance添加服務(wù)實例        //總體流程：把需要注冊的實例放到內(nèi)存阻塞隊列中，另外會起另一個線程從內(nèi)存中取出intance實例放到Service中，即注冊成功了        addInstance(namespaceId, serviceName, instance.isEphemeral(), instance);    }

registerInstance()干了兩件事兒，第一就是createEmptyService()方法從請求參數(shù)匯總獲得serviceName（服務(wù)名）和namespaceId（命名空間Id），第二就是調(diào)用registerInstance注冊實例。先看一下createEmptyService方法。

7.3.2 服務(wù)端構(gòu)建注冊表

Nacos的注冊表是多級存儲結(jié)構(gòu)，最外層是通過namespace來實現(xiàn)環(huán)境隔離，然后是group分組，分組下就是服務(wù)，一個服務(wù)有可以分為不同的集群，集群中包含多個實例。因此其注冊表結(jié)構(gòu)為一個Map，類型是：Map>外層key是namespace_id，內(nèi)層key是group + serviceName，Service內(nèi)部維護一個Map，結(jié)構(gòu)是：Map的key是clusterName，其值是集群信息；Cluster內(nèi)部維護一個Set集合Set ephemeralInstances和Set persistentInstances，元素是Instance類型，代表集群中的多個實例。

createEmptyService()方法就是服務(wù)端構(gòu)建注冊表的方法，基于AP架構(gòu)的Nacos實際就是將注冊實例信息保存在內(nèi)存中。

/**     * 1、創(chuàng)建一個Serivice對象，內(nèi)部包含了一個clusterMap。     * 2、將service對象放入到SeriviceMap中，結(jié)構(gòu)為：Map>。     * 3、開啟一個定時任務(wù)用來檢測實例的心跳是否超時，每5秒執(zhí)行一次。     *     * @param namespaceId     * @param serviceName     * @param local     * @throws NacosException     */    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);        //第一次注冊進來，從注冊表里獲取命名空間，肯定是為null，所以需要構(gòu)建一個命名空間，        //設(shè)置nameSpace等信息，如果Service實例為空，則創(chuàng)建并保存到緩存中        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();            //注冊和初始化，通過putService()方法將服務(wù)緩存到內(nèi)存            putServiceAndInit(service);            if (!local) {                addOrReplaceService(service);            }        }    }

createEmptyService()方法主要作用如下：

創(chuàng)建一個Serivice對象，內(nèi)部包含了一個clusterMap；將service對象放入到SeriviceMap中，結(jié)構(gòu)為：Map>；開啟一個定時任務(wù)用來檢測實例的心跳是否超時，每5秒執(zhí)行一次。

createServiceIfAbsent()方法主要作用在于第一次注冊進來，從注冊表里獲取命名空間，肯定是為null，所以需要構(gòu)建一個命名空間，設(shè)置nameSpace等信息并保存到緩存中。這個方法里值得注意的是putServiceAndInit()方法，可以點進來看一下這個方法：

private void putServiceAndInit(Service service) throws NacosException {        //構(gòu)建注冊表雙層map，初始化serviceMap --> Map> serviceMap        putService(service);        //初始化service，開啟心跳檢測的線程        service.init();        //實現(xiàn)數(shù)據(jù)一致性監(jiān)聽        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());    }

這里我著重putService(service)方法，這里實際是將注冊的實例緩存到內(nèi)存的注冊表中

/** * 通過putService()方法將服務(wù)緩存到內(nèi)存 * * @param service service */public void putService(Service service) {    if (!serviceMap.containsKey(service.getNamespaceId())) {        //雙檢索防止并發(fā)，為了防止同一個服務(wù)多個地方同時注冊        synchronized (putServiceLock) {            if (!serviceMap.containsKey(service.getNamespaceId())) {                //構(gòu)建NamespaceId，Serivce對象放到了ServiceMap里面了，也就是說下次我們再調(diào)用getService(namespaceId)的時候就可以獲取到一個Service對象了                serviceMap.put(service.getNamespaceId(), new ConcurrentSkipListMap<>());            }        }    }    //構(gòu)建 service name    serviceMap.get(service.getNamespaceId()).put(service.getName(), service);}

7.3.3 Nacos服務(wù)端心跳機制

接下來我們看一下 putServiceAndInit(Service service)方法中的，init()初始化方法是怎么保持心跳連接的。

/** * service.init()建立心跳機制 */public void init() {    //客戶端心跳檢查任務(wù)，每隔5s執(zhí)行一次，clientBeatCheckTask是一個線程的方法    HealthCheckReactor.scheduleCheck(clientBeatCheckTask);    for (Map.Entry entry : clusterMap.entrySet()) {        entry.getValue().setService(this);        entry.getValue().init();    }}/** * Schedule client beat check task with a delay. * * @param task client beat check task */public static void scheduleCheck(ClientBeatCheckTask task) {    //客戶端的心跳任務(wù)，這里并沒有嵌套調(diào)用，而是開啟延遲5s的任務(wù)，然后每隔5秒鐘執(zhí)行一次    futureMap.putIfAbsent(task.taskKey(), GlobalExecutor.scheduleNamingHealth(task, 5000, 5000, TimeUnit.MILLISECONDS));}public class ClientBeatCheckTask implements Runnable {        private Service service;        public ClientBeatCheckTask(Service service) {        this.service = service;    }        @JsonIgnore    public PushService getPushService() {        return ApplicationUtils.getBean(PushService.class);    }        @JsonIgnore    public DistroMapper getDistroMapper() {        return ApplicationUtils.getBean(DistroMapper.class);    }        public GlobalConfig getGlobalConfig() {        return ApplicationUtils.getBean(GlobalConfig.class);    }        public SwitchDomain getSwitchDomain() {        return ApplicationUtils.getBean(SwitchDomain.class);    }        public String taskKey() {        return KeyBuilder.buildServiceMetaKey(service.getNamespaceId(), service.getName());    }        @Override    public void run() {        try {            /**             * nacos心跳在集群架構(gòu)下只允許在一臺機器上執(zhí)行健康檢查任務(wù)             *             * 集群中有多臺機器，本方法在于對服務(wù)名稱做hash運算再對機器數(shù)量取模后，那么             * 這里每次只有定位到一臺機器，其他機器都直接return了             *             * 疑問：如果一臺機器掛了會怎么辦？這里取模會不會亂掉？那這里會不會要做一致性hash？             * 在nacos集群中每臺機器之間也是存在狀態(tài)同步的，每臺機器之間都有集群節(jié)點同步任務(wù)，詳見com.alibaba.nacos.naming.cluster.ServerListManager.ServerStatusReporter             *             */            if (!getDistroMapper().responsible(service.getName())) {                return;            }                        if (!getSwitchDomain().isHealthCheckEnabled()) {                return;            }            //獲取服務(wù)端所有實例            List instances = service.allIPs(true);                        // first set health status of instances:            /**             *  for循環(huán)對每個實例都做健康檢查             *  在這個方法里面主要是循環(huán)當(dāng)前service的每一個臨時實例 用當(dāng)前時間減去最后一次心跳時間 是否大于心跳超時時間來判斷心跳是否超時，             *  如果大于這個時間會執(zhí)行instance.setHealthy(false)將實例的健康狀態(tài)改為false；但是這個定時任務(wù)不會立即執(zhí)行，會每5秒執(zhí)行一次：             */            for (Instance instance : instances) {                //判斷心跳是否超時：當(dāng)前時間 - 實例上次心跳時間 > 心跳的超時時間【默認是15秒】？                if (System.currentTimeMillis() - instance.getLastBeat() > instance.getInstanceHeartBeatTimeOut()) {                    if (!instance.isMarked()) {                        if (instance.isHealthy()) {                            //如果大于心跳默認時間，把實例的 healthy 設(shè)置為false【服務(wù)列表一開始不會刪掉，一開始會變成false】                            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;                }                //當(dāng)前時間 - 實例上一次心跳時間 > 實例的刪除時間【默認30s】                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);        }            }        private void deleteIp(Instance instance) {                try {            NamingProxy.Request request = NamingProxy.Request.newRequest();            request.appendParam("ip", instance.getIp()).appendParam("port", String.valueOf(instance.getPort()))                    .appendParam("ephemeral", "true").appendParam("clusterName", instance.getClusterName())                    .appendParam("serviceName", service.getName()).appendParam("namespaceId", service.getNamespaceId());            //調(diào)用本地服務(wù)            String url = "http://" + IPUtil.localHostIP() + IPUtil.IP_PORT_SPLITER + EnvUtil.getPort() + EnvUtil.getContextPath()                    + UtilsAndCommons.NACOS_NAMING_CONTEXT + "/instance?" + request.toUrl();  //  /v/ns/instance                        // delete instance asynchronously:            HttpClient.asyncHttpDelete(url, null, null, new Callback() {                @Override                public void onReceive(RestResult result) {                    if (!result.ok()) {                        Loggers.SRV_LOG                                .error("[IP-DEAD] failed to delete ip automatically, ip: {}, caused {}, resp code: {}",                                        instance.toJson(), result.getMessage(), result.getCode());                    }                }                    @Override                public void onError(Throwable throwable) {                    Loggers.SRV_LOG                            .error("[IP-DEAD] failed to delete ip automatically, ip: {}, error: {}", instance.toJson(),                                    throwable);                }                    @Override                public void onCancel() {                        }            });                    } catch (Exception e) {            Loggers.SRV_LOG                    .error("[IP-DEAD] failed to delete ip automatically, ip: {}, error: {}", instance.toJson(), e);        }    }}

可以看出init方法是開啟了一個異步線程ClientBeatCheckTask去做了個周期性發(fā)送心跳的機制，方法中客戶端心跳檢查任務(wù)，開啟延遲5s的任務(wù)，然后每隔5秒鐘執(zhí)行一次。

service.init()方法主要通過定時任務(wù)不斷檢測當(dāng)前服務(wù)下所有實例最后發(fā)送心跳包的時間。在這個方法里面主要是循環(huán)當(dāng)前service的每一個臨時實例，用當(dāng)前時間減去最后一次心跳時間是否大于15s來判斷心跳是否超時，如果大于這個時間會執(zhí)行instance.setHealthy(false)將實例的健康狀態(tài)改為false，但是這個定時任務(wù)不會立即執(zhí)行，會每5秒執(zhí)行一次；當(dāng)前時間 - 實例上一次心跳時間 > 實例的刪除時間【默認30s】就會刪除實例。

那么服務(wù)實例的最后心跳包更新時間是誰來觸發(fā)的呢?實際上前面在說客戶端注冊時有說到, Nacos客戶端注冊服務(wù)的同時也建立了心跳機制。

7.3.4服務(wù)端實例注冊

上文中registerInstance注冊實例方法中還有一個最最重要的方法就是addInstance()方法，其本質(zhì)上就是把當(dāng)前注冊的服務(wù)實例保存到Service中。

public void addInstance(String namespaceId, String serviceName, boolean ephemeral, Instance... ips)            throws NacosException {                String key = KeyBuilder.buildInstanceListKey(namespaceId, serviceName, ephemeral);                Service service = getService(namespaceId, serviceName);                synchronized (service) {            //將需要注冊的實例全部放到Cluster，再將Cluster放在Service里            List instanceList = addIpAddresses(service, ephemeral, ips);                        Instances instances = new Instances();            instances.setInstanceList(instanceList);            //看一下 consistencyService 對象初始化的地方就知道走的是哪個實現(xiàn)            consistencyService.put(key, instances);        }    }    public static String buildInstanceListKey(String namespaceId, String serviceName, boolean ephemeral) {        //根據(jù) ephemeral 取值默認是 true為臨時實例，臨時實例是存放在內(nèi)存的；false即為永久實例寫到文件的，可以通過此參數(shù)區(qū)分nacos是AP還是CP架構(gòu)        return ephemeral                ? buildEphemeralInstanceListKey(namespaceId, serviceName)                : buildPersistentInstanceListKey(namespaceId, serviceName);    }

這里著重看一下這個put方法，put方法主要做了兩件事，第一對對客戶端的請求過來的實例進行注冊，第二是Nacos集群架構(gòu)下的數(shù)據(jù)同步，Nacos默認用的是臨時實例，也就是ephemeral = true，也就是本文的重點AP架構(gòu)的Nacos注冊原理。

public void put(String key, Record value) throws NacosException {        //注冊邏輯：實際就是把實例注冊任務(wù)放到內(nèi)存阻塞隊列中        onPut(key, value);        //AP 架構(gòu)下的節(jié)點數(shù)據(jù)同步        distroProtocol.sync(new DistroKey(key, KeyBuilder.INSTANCE_LIST_KEY_PREFIX), DataOperation.CHANGE,                globalConfig.getTaskDispatchPeriod() / 2);    }    /**    * 注冊邏輯    */    public void onPut(String key, Record value) {                if (KeyBuilder.matchEphemeralInstanceListKey(key)) {            Datum datum = new Datum<>();            datum.value = (Instances) value;            datum.key = key;            datum.timestamp.incrementAndGet();            //把客戶端信息注冊信息更新到注冊表            dataStore.put(key, datum);        }                if (!listeners.containsKey(key)) {            return;        }        //這里放的是DataOperation.CHANGE        notifier.addTask(key, DataOperation.CHANGE);    }

先來看一下onPut()方法，不難發(fā)現(xiàn)當(dāng)注冊實例數(shù)據(jù)有改變時，就無腦將這個實例扔到這個task內(nèi)存阻塞隊列中去，具體可以看一下addTask()方法。

public class Notifier implements Runnable {                private ConcurrentHashMap services = new ConcurrentHashMap<>(10 * 1024);        //用于存放即將要注冊實例信息的內(nèi)存阻塞隊列        private BlockingQueue> tasks = new ArrayBlockingQueue<>(1024 * 1024);                /**         * Add new notify task to queue.         *         * @param datumKey data key         * @param action   action for data         */        public void addTask(String datumKey, DataOperation action) {                        if (services.containsKey(datumKey) && action == DataOperation.CHANGE) {                return;            }            if (action == DataOperation.CHANGE) {                services.put(datumKey, StringUtils.EMPTY);            }            /**             * 把客戶端的參數(shù)封裝成pair對象后，放進了一個內(nèi)存隊列中，注冊就結(jié)束了，看這里并沒有把客戶端的注冊信息寫進雙層map中             * 憑經(jīng)驗?zāi)懿碌剑@里把客戶端對象放進內(nèi)存隊列，后續(xù)肯定是通過異步起線程的方式去注冊             */            tasks.offer(Pair.with(datumKey, action));        }                public int getTaskSize() {            return tasks.size();        }                @Override        public void run() {            Loggers.DISTRO.info("distro notifier started");            /**             * Spring啟動時就會開啟一個線程加載 Notifier任務(wù)，這里就會死循環(huán)一直從內(nèi)存隊列中拿取實例信息實現(xiàn)異步注冊             *             * 問   題1：這里的for循環(huán)會占用cpu資源嗎？             *          不會占用，因為tasks是個阻塞隊列，如果tasks中沒有實例信息，這里就會阻塞在這，不會無腦死循環(huán)             *             * 問   題2：為什么要把實例信息都無腦先放在內(nèi)存阻塞隊列中，然后另起一個線程去異步注冊呢？阿里這里為什么要這么設(shè)計？             * 個人理解：nacos是在阿里內(nèi)部使用的中間件，一般是需要滿足三高特性【高并發(fā)、高性能、高可擴展】，阿里內(nèi)部就有幾十萬臺機器，             * 如果不能實現(xiàn)高并發(fā)注冊那么肯定會有很多問題。比如訂單服務(wù)A需要注冊到nacos時，是在訂單A啟動時就需要注冊，服務(wù)注冊到nacos的邏輯             * 還是比較復(fù)雜的【詳見com.alibaba.nacos.naming.core.Service#updateIPs】，假如這里不用異步注冊而是用同步注冊的方式，那么             * 服務(wù)注冊到nacos需要花費很多時間，這才是一個注冊到nacos的行為就花費了大量時間，那么如果多幾個中間需要加載的話，那得浪費多少時間？             * 所以這里采用異步注冊。             *             * 問   題3：內(nèi)存阻塞隊列 tasks 會不會有堆積的情況呢？             * 實際上看了com.alibaba.nacos.naming.core.Cluster#updateIps注冊方法可以發(fā)現(xiàn)，注冊實際上就是把實例信息寫進一個內(nèi)存集合Set中             * 【com.alibaba.nacos.naming.core.Cluster#ephemeralInstances】這樣的操作其實是很快的，假如真有個運維寫了個批量注冊的腳本             * 把一堆機器同時注冊進來，那這樣確實有可能會造成內(nèi)存阻塞隊列tasks的堆積現(xiàn)象，但是這種情況并沒什么關(guān)系，Eureka有時候?qū)嵗远紩兄?            * 幾十秒，對當(dāng)前的nacos架構(gòu)而言，既然要實現(xiàn)高并發(fā)那么只能犧牲一點實例注冊的即使響應(yīng)時間。正常情況下，即使有幾十臺幾百臺機器同時注冊，             * 由于注冊是內(nèi)存操作，速度也很快，可以說是準實時，基本上正常情況下注冊信息1s就能感知到。             *             */            for (; ; ) {                try {                    //從內(nèi)存隊列中拿出任務(wù)                    Pair pair = tasks.take();                    //拿出pair對象中的客戶端信息去注冊                    handle(pair);                } catch (Throwable e) {                    //這個線程即使拋異常也不終止                    Loggers.DISTRO.error("[NACOS-DISTRO] Error while handling notifying task", e);                }            }        }                private void handle(Pair pair) {            try {                //前面拼接的參數(shù) "com.alibaba.nacos.naming.iplist.ephemeral.namespaceId##serviceName"                String datumKey = pair.getValue0();                //前面?zhèn)鞯腄ataOperation.CHANGE。                DataOperation action = pair.getValue1();                                services.remove(datumKey);                                int count = 0;                                if (!listeners.containsKey(datumKey)) {                    return;                }                                for (RecordListener listener : listeners.get(datumKey)) {                                        count++;                                        try {                        if (action == DataOperation.CHANGE) {                            /**                             * 拿到前面放的map中的客戶端信息dataStore.get，這里的key就是前面的                             * 拼接的參數(shù) "com.alibaba.nacos.naming.iplist.ephemeral.namespaceId##serviceName"                             * 前面放的是DataOperation.CHANGE。                             *                             */                            listener.onChange(datumKey, dataStore.get(datumKey).value);                            continue;                        }                                                if (action == DataOperation.DELETE) {                            listener.onDelete(datumKey);                            continue;                        }                    } catch (Throwable e) {                        Loggers.DISTRO.error("[NACOS-DISTRO] error while notifying listener of key: {}", datumKey, e);                    }                }                                if (Loggers.DISTRO.isDebugEnabled()) {                    Loggers.DISTRO                            .debug("[NACOS-DISTRO] datum change notified, key: {}, listener count: {}, action: {}",                                    datumKey, count, action.name());                }            } catch (Throwable e) {                Loggers.DISTRO.error("[NACOS-DISTRO] Error while handling notifying task", e);            }        }    }

當(dāng)有實例需要注冊時，直接調(diào)用addTask()方法將這個實例信息無腦扔進內(nèi)存阻塞隊列中去，注冊就結(jié)束了。這個應(yīng)該算是Nacos注冊的一個精髓吧，Nacos為了提高性能其源碼使用了大量的異步任務(wù)、異步線程等操作，用這些方式對提升Nacos性能有很大幫助。不難猜到，這里把客戶端實例對象放進內(nèi)存隊列，后續(xù)肯定是通過異步起線程的方式去注冊。

不難發(fā)現(xiàn)addTask()方法是Notifier類的方法，Notifier實現(xiàn)了Runnable接口，很明顯這就是一個異步線程，這里跟上面的猜想一致，Nacos就是通過開啟了一個異步線程實現(xiàn)注冊的，具體的注冊方法直接可以看Notifier線程的run()方法即可。那么這個Notifier線程是啥時候開啟的呢？

@DependsOn("ProtocolManager")@org.springframework.stereotype.Service("distroConsistencyService")public class DistroConsistencyServiceImpl implements EphemeralConsistencyService, DistroDataProcessor {        //本類DistroConsistencyServiceImpl注入到了Spring容器中，所以項目啟動時類加載的時候就會初始化這個方法    @PostConstruct    public void init() {        //線程池執(zhí)行器，執(zhí)行notifier線程的任務(wù)        GlobalExecutor.submitDistroNotifyTask(notifier);    }}

了解Spring的應(yīng)該都知道這個是Spring加載的一種初始化方式，Spring啟動時加載這個init方法初始化數(shù)據(jù)，就會開啟一個線程加載Notifier任務(wù)。

看了Notifier線程的run()方法，不免會有幾個疑問。第一、這里的for循環(huán)會占用cpu資源嗎？第二、為什么要把實例信息都無腦先放在內(nèi)存阻塞隊列中，然后另起一個線程去異步注冊呢？第三、阿里這里為什么要這么設(shè)計？這樣設(shè)計好處是什么呢？

這里第一個問題不會占用cpu資源，因為tasks是個阻塞隊列，如果tasks中沒有實例信息，這里就會阻塞在這，不會無腦死循環(huán)，所以是不會占用cpu資源的；

第二個問題個人理解：Nacos是在阿里內(nèi)部使用的中間件，肯定是需要滿足高并發(fā)、高性能、高可擴展，阿里內(nèi)部估計就有幾十萬臺機器，如果不能實現(xiàn)高并發(fā)注冊那么肯定會有很多問題。比如訂單服務(wù)需要注冊到nacos時，那么訂單啟動時就需要注冊，服務(wù)注冊到Nacos的邏輯還是比較復(fù)雜的【詳見com.alibaba.nacos.naming.core.Service#updateIPs】，假如這里不用異步注冊而是用同步注冊的方式，那么服務(wù)注冊到Nacos需要花費很多時間，這才是一個注冊到Nacos的行為就花費了大量時間，那么如果多幾個中間需要加載的話，那會浪費很多時間，所以這里采用異步注冊。

那么我們再看一下異步寫注冊表的方法：

public void onChange(String key, Instances value) throws Exception {                Loggers.SRV_LOG.info("[NACOS-RAFT] datum is changed, key: {}, value: {}", key, value);                for (Instance instance : value.getInstanceList()) {                        if (instance == null) {                // Reject this abnormal instance list:                throw new RuntimeException("got null instance " + key);            }            //設(shè)置權(quán)重默認值啥的            if (instance.getWeight() > 10000.0D) {                instance.setWeight(10000.0D);            }                        if (instance.getWeight() < 0.01D && instance.getWeight() > 0.0D) {                instance.setWeight(0.01D);            }        }        //真正的注冊實例的方法        updateIPs(value.getInstanceList(), KeyBuilder.matchEphemeralInstanceListKey(key));                recalculateChecksum();      }        public void updateIPs(Collection instances, boolean ephemeral) {        Map> ipMap = new HashMap<>(clusterMap.size());        for (String clusterName : clusterMap.keySet()) {            ipMap.put(clusterName, new ArrayList<>());        }                for (Instance instance : instances) {            try {                if (instance == null) {                    Loggers.SRV_LOG.error("[NACOS-DOM] received malformed ip: null");                    continue;                }                                if (StringUtils.isEmpty(instance.getClusterName())) {                    instance.setClusterName(UtilsAndCommons.DEFAULT_CLUSTER_NAME);                }                                if (!clusterMap.containsKey(instance.getClusterName())) {                    Loggers.SRV_LOG                            .warn("cluster: {} not found, ip: {}, will create new cluster with default configuration.",                                    instance.getClusterName(), instance.toJson());                    Cluster cluster = new Cluster(instance.getClusterName(), this);                    cluster.init();                    getClusterMap().put(instance.getClusterName(), cluster);                }                                List clusterIPs = ipMap.get(instance.getClusterName());                if (clusterIPs == null) {                    clusterIPs = new LinkedList<>();                    ipMap.put(instance.getClusterName(), clusterIPs);                }                                clusterIPs.add(instance);            } catch (Exception e) {                Loggers.SRV_LOG.error("[NACOS-DOM] failed to process ip: " + instance, e);            }        }                for (Map.Entry> entry : ipMap.entrySet()) {            //make every ip mine            List entryIPs = entry.getValue();            /**             * 注冊邏輯updateIps，更新注冊表信息：             *             * 這個方法里面會將已經(jīng)注冊過的實例列表復(fù)制一份，將新的實例和老的實例都更新到一個集合中，             * 最終再將這個集合更新到真正的實例列表，是一種寫時復(fù)制的思想，主要時為了解決并發(fā)沖突，             * 在寫的過程中，其他線程讀到的還是舊數(shù)據(jù)，等真正寫完之后再將數(shù)據(jù)更新回去。             *             * 思考一下，正常情況下在寫之前都要加鎖，不然可能會有讀寫并發(fā)問題，這里為什么不在加鎖？             * 假如這里加了一個鎖之后，相當(dāng)于把讀和寫操作排隊串行化執(zhí)行了，就是讀寫不能同時進行了，             * 這里并發(fā)肯定會很低，所以這里用copy on write機制，將原來的注冊表復(fù)制出一個副本，然后進             * 行修改，此時讀請求進來還是讀老的注冊表，這樣讀寫就能并發(fā)執(zhí)行。             *             * 那這里用讀寫分離和加鎖串行執(zhí)行有什么優(yōu)劣勢嗎？             * 讀寫分離：寫的時候?qū)懙氖歉北?，讀的是老得數(shù)據(jù)，這樣可能讀到讀不是最新數(shù)據(jù)，只有當(dāng)副本寫完將             * 老數(shù)據(jù)替換，此時讀的才是最新數(shù)據(jù)，讀寫分離雖然提高了讀寫并發(fā)但是對數(shù)據(jù)的一致性稍有妥協(xié)，但是             * 對于此時注冊的場景而言影響不大，即使是沒有讀到最新數(shù)據(jù)也沒關(guān)系，最多就當(dāng)此服務(wù)啟動的慢一點而             * 已，當(dāng)前這個注冊場景，還是提高并發(fā)注冊能力稍重要些，若是對讀寫數(shù)據(jù)一致場景要求很高時，就必須得             * 加鎖串行執(zhí)行             *             * 加鎖執(zhí)行：讀寫都是加鎖執(zhí)行，寫完后再去讀，讀的一定是最新的數(shù)據(jù)，讀寫數(shù)據(jù)強一致，但是這里根本不需要數(shù)據(jù)強一致             *             * 這里有個疑問注冊時每個微服務(wù)都去復(fù)制一個副本，然后將副本替換回原件時會不會有覆蓋的問題？             * 不會，這里是開的一個線程去拿內(nèi)存隊列的數(shù)據(jù)進行注冊的，所以不可能存在覆蓋并發(fā)問題。詳見             * com.alibaba.nacos.naming.consistency.ephemeral.distro.DistroConsistencyServiceImpl#init()方法             *             */            clusterMap.get(entry.getKey()).updateIps(entryIPs, ephemeral);        }                setLastModifiedMillis(System.currentTimeMillis());        //上面已經(jīng)更新過了注冊表后，這里需要發(fā)布事件，主動通知客戶端        getPushService().serviceChanged(this);        StringBuilder stringBuilder = new StringBuilder();                for (Instance instance : allIPs()) {            stringBuilder.append(instance.toIpAddr()).append("_").append(instance.isHealthy()).append(",");        }                Loggers.EVT_LOG.info("[IP-UPDATED] namespace: {}, service: {}, ips: {}", getNamespaceId(), getName(),                stringBuilder.toString());             }    /**     * Update instance list.     *     * @param ips       instance list    需要注冊的實例     * @param ephemeral whether these instances are ephemeral  是哪種實例，永久實例還是臨時實例     */    public void updateIps(List ips, boolean ephemeral) {        //根據(jù)傳進來的 ephemeral 判斷是哪種實例，直接把這種實例復(fù)制一份toUpdateInstances        Set toUpdateInstances = ephemeral                ? ephemeralInstances    //臨時實例 【這里這個實例是注冊表已有的實例，這里直接將原有的實例再次復(fù)制了一份，就是toUpdateInstances】                : persistentInstances;  //永久實例 【這里這個實例是注冊表已有的實例，這里直接將原有的實例再次復(fù)制了一份，就是toUpdateInstances】        HashMap oldIpMap = new HashMap<>(toUpdateInstances.size());        //循環(huán)將上面復(fù)制的實例toUpdateInstances放oldIpMap中        for (Instance ip : toUpdateInstances) {            oldIpMap.put(ip.getDatumKey(), ip);        }        /**         * 下面就是對比新注冊的實例與副本實例的差異，判斷新的實例是新增、刪除、還是修改，這里使用了copyOnWrite機制，上面的toUpdateInstances就是         * 復(fù)制出來的副本，新增、刪除、修改都是在副本上進行，之后再將原注冊表覆蓋         * 新增：直接新增到副本中         * 修改：直接修改副本的實例信息         * 刪除：刪除副本的實例         *         */        //更新注冊表        List updatedIPs = updatedIps(ips, oldIpMap.values());        if (updatedIPs.size() > 0) {            for (Instance ip : updatedIPs) {                Instance oldIP = oldIpMap.get(ip.getDatumKey());                                // do not update the ip validation status of updated ips                // because the checker has the most precise result                // Only when ip is not marked, don"t we update the health status of IP:                if (!ip.isMarked()) {                    ip.setHealthy(oldIP.isHealthy());                }                                if (ip.isHealthy() != oldIP.isHealthy()) {                    // ip validation status updated                    Loggers.EVT_LOG.info("{} {SYNC} IP-{} {}:{}@{}", getService().getName(),                            (ip.isHealthy() ? "ENABLED" : "DISABLED"), ip.getIp(), ip.getPort(), getName());                }                                if (ip.getWeight() != oldIP.getWeight()) {                    // ip validation status updated                    Loggers.EVT_LOG.info("{} {SYNC} {IP-UPDATED} {}->{}", getService().getName(), oldIP.toString(),                            ip.toString());                }            }        }        //刪除注冊表        List newIPs = subtract(ips, oldIpMap.values());        if (newIPs.size() > 0) {            Loggers.EVT_LOG                    .info("{} {SYNC} {IP-NEW} cluster: {}, new ips size: {}, content: {}", getService().getName(),                            getName(), newIPs.size(), newIPs.toString());                        for (Instance ip : newIPs) {                HealthCheckStatus.reset(ip);            }        }                List deadIPs = subtract(oldIpMap.values(), ips);                if (deadIPs.size() > 0) {            Loggers.EVT_LOG                    .info("{} {SYNC} {IP-DEAD} cluster: {}, dead ips size: {}, content: {}", getService().getName(),                            getName(), deadIPs.size(), deadIPs.toString());                        for (Instance ip : deadIPs) {                HealthCheckStatus.remv(ip);            }        }                toUpdateInstances = new HashSet<>(ips);

第三個問題：可以看上面這個寫注冊表的源碼，當(dāng)服務(wù)A需要注冊到Nacos時，并不是直接寫進Nacos的注冊表里，實際上是先拷貝了一個副本，訂單服務(wù)注冊寫注冊表時直接寫副本的注冊表，副本寫完后才會替換原來Nacos中的注冊表，所以當(dāng)庫存服務(wù)需要從Nacos拉取服務(wù)時，拉取的是Nacos實際注冊表中的信息，這種設(shè)計方式能夠大大提高Nacos的注冊性能。

類似于CopyOnWriteArrayList的copy on write機制，也就是寫時復(fù)制、讀寫分離設(shè)計思想。這種讀寫分離對于客戶端注冊感知實時性可能會稍差點，但是這種情況并沒什么關(guān)系，Eureka有時候?qū)嵗远紩兄獛资耄瑢Ξ?dāng)前的nacos架構(gòu)而言，既然要實現(xiàn)高并發(fā)那么只能犧牲一點實例注冊的及時響應(yīng)時間。

當(dāng)Nacos注冊成功后，就需要發(fā)布事件，主動通知客戶端，接下來可以看一下發(fā)布事件的源碼：

public void serviceChanged(Service service) {        // merge some change events to reduce the push frequency:        if (futureMap                .containsKey(UtilsAndCommons.assembleFullServiceName(service.getNamespaceId(), service.getName()))) {            return;        }        //時間發(fā)布，發(fā)布事件，通知客戶端        this.applicationContext.publishEvent(new ServiceChangeEvent(this, service));    }

這里Nacos會通過udp的方式將服務(wù)變動通知給訂閱的客戶端。Nacos的這種推送模式相對于zk那種利用tcp長連接而言還是會節(jié)約很多資源，即使有大量節(jié)點更新也不會使Nacos出現(xiàn)性能瓶頸。

當(dāng)Nacos客戶端接收到了udp消息后會給服務(wù)端返回一個ack，如果Nacos超時未收到ack，還會有重發(fā)機制，超過了這個超時時間就不再重發(fā)了。雖然udp是個不可靠協(xié)議不能保證消息一定能推送到客戶端，但是Nacos客戶端還是有定時輪訓(xùn)做兜底定時查詢Nacos注冊表。Nacas采用了這兩種機制，既保證了實時性，又保證了數(shù)據(jù)更新不會被漏掉。

7.3.5 Nacos集群新節(jié)點啟動數(shù)據(jù)同步

Nacos數(shù)據(jù)同步分為全量同步和增量同步，全量同步就是初始化數(shù)據(jù)一次性同步，而增量同步是指有數(shù)據(jù)增加的時候，只同步增加的數(shù)據(jù)。

7.3.5.1 Nacos集群全量數(shù)據(jù)同步

Nacos集群有新的節(jié)點啟動時，DistroProtocol類就會在Spring加載時調(diào)用構(gòu)造方法，同時開啟一個數(shù)據(jù)同步任務(wù)，該方法會執(zhí)行startVerifyTask()和startLoadTask()，我們重點關(guān)注startLoadTask(),具體代碼如下：

@Componentpublic class DistroProtocol {    /**     * 當(dāng)nacos集群只有兩臺機器時，此時若又新增一臺機器，此時需要將原來兩臺機器的數(shù)據(jù)同步到新的nacos機器上     *     * @param memberManager     * @param distroComponentHolder     * @param distroTaskEngineHolder     * @param distroConfig     */    public DistroProtocol(ServerMemberManager memberManager, DistroComponentHolder distroComponentHolder,            DistroTaskEngineHolder distroTaskEngineHolder, DistroConfig distroConfig) {        this.memberManager = memberManager;        this.distroComponentHolder = distroComponentHolder;        this.distroTaskEngineHolder = distroTaskEngineHolder;        this.distroConfig = distroConfig;        //本項目啟動時DistroProtocol類加載時需要加載本構(gòu)造方法，開啟一個數(shù)據(jù)同步任務(wù)        startDistroTask();    }        private void startDistroTask() {        if (EnvUtil.getStandaloneMode()) {            isInitialized = true;            return;        }        //啟動startVerifyTask，做數(shù)據(jù)同步校驗        startVerifyTask();        //加載任務(wù)        startLoadTask();    }        private void startLoadTask() {        ///處理狀態(tài)回調(diào)對象        DistroCallback loadCallback = new DistroCallback() {            //處理成功            @Override            public void onSuccess() {                isInitialized = true;            }            //處理失敗            @Override            public void onFailed(Throwable throwable) {                isInitialized = false;            }        };        GlobalExecutor.submitLoadDataTask(                new DistroLoadDataTask(memberManager, distroComponentHolder, distroConfig, loadCallback));    } }

上面方法會調(diào)用DistroLoadDataTask對象，而該對象其實是個線程，因此會執(zhí)行它的run方法，run方法會調(diào)用load()方法實現(xiàn)數(shù)據(jù)全量加載，代碼如下：

public class DistroLoadDataTask implements Runnable {        private final ServerMemberManager memberManager;        private final DistroComponentHolder distroComponentHolder;        private final DistroConfig distroConfig;        private final DistroCallback loadCallback;        private final Map loadCompletedMap;        public DistroLoadDataTask(ServerMemberManager memberManager, DistroComponentHolder distroComponentHolder,            DistroConfig distroConfig, DistroCallback loadCallback) {        this.memberManager = memberManager;        this.distroComponentHolder = distroComponentHolder;        this.distroConfig = distroConfig;        this.loadCallback = loadCallback;        loadCompletedMap = new HashMap<>(1);    }    /**     * 數(shù)據(jù)加載過程     */    @Override    public void run() {        try {            //加載數(shù)據(jù)            load();            if (!checkCompleted()) {                GlobalExecutor.submitLoadDataTask(this, distroConfig.getLoadDataRetryDelayMillis());            } else {                loadCallback.onSuccess();                Loggers.DISTRO.info("[DISTRO-INIT] load snapshot data success");            }        } catch (Exception e) {            loadCallback.onFailed(e);            Loggers.DISTRO.error("[DISTRO-INIT] load snapshot data failed. ", e);        }    }}

數(shù)據(jù)同步會通過Http請求從遠程服務(wù)器獲取數(shù)據(jù)，并同步到當(dāng)前服務(wù)的緩存中。執(zhí)行流程如下：

首先，loadAllDataSnapshotFromRemote()從遠程加載所有數(shù)據(jù)，并處理同步到本機；

第二，transportAgent.getDatumSnapshot()遠程加載數(shù)據(jù)，通過Http請求執(zhí)行遠程加載；

第三，dataProcessor.processSnapshot()處理數(shù)據(jù)同步到本地

/**     * 加載數(shù)據(jù)，并同步     *     * @throws Exception     */    private void load() throws Exception {        while (memberManager.allMembersWithoutSelf().isEmpty()) {            Loggers.DISTRO.info("[DISTRO-INIT] waiting server list init...");            TimeUnit.SECONDS.sleep(1);        }        while (distroComponentHolder.getDataStorageTypes().isEmpty()) {            Loggers.DISTRO.info("[DISTRO-INIT] waiting distro data storage register...");            TimeUnit.SECONDS.sleep(1);        }        for (String each : distroComponentHolder.getDataStorageTypes()) {            if (!loadCompletedMap.containsKey(each) || !loadCompletedMap.get(each)) {                //從遠端機器拉取數(shù)據(jù)，從遠程加載所有數(shù)據(jù)，并處理同步到本機                loadCompletedMap.put(each, loadAllDataSnapshotFromRemote(each));            }        }    }    /**     * 從遠端機器拉取數(shù)據(jù)     *     * @param resourceType     * @return     */    private boolean loadAllDataSnapshotFromRemote(String resourceType) {        DistroTransportAgent transportAgent = distroComponentHolder.findTransportAgent(resourceType);        DistroDataProcessor dataProcessor = distroComponentHolder.findDataProcessor(resourceType);        if (null == transportAgent || null == dataProcessor) {            Loggers.DISTRO.warn("[DISTRO-INIT] Can"t find component for type {}, transportAgent: {}, dataProcessor: {}",                    resourceType, transportAgent, dataProcessor);            return false;        }        //拉取不包含自己的機器        for (Member each : memberManager.allMembersWithoutSelf()) {            try {                Loggers.DISTRO.info("[DISTRO-INIT] load snapshot {} from {}", resourceType, each.getAddress());                //調(diào)取接口獲取除自己外的所有機器                DistroData distroData = transportAgent.getDatumSnapshot(each.getAddress());                //同步數(shù)據(jù)                boolean result = dataProcessor.processSnapshot(distroData);                Loggers.DISTRO                        .info("[DISTRO-INIT] load snapshot {} from {} result: {}", resourceType, each.getAddress(),                                result);                //同步成功直接return，從一臺機器上同步                if (result) {                    return true;                }            } catch (Exception e) {                Loggers.DISTRO.error("[DISTRO-INIT] load snapshot {} from {} failed.", resourceType, each.getAddress(), e);            }        }        return false;    }        private boolean checkCompleted() {        if (distroComponentHolder.getDataStorageTypes().size() != loadCompletedMap.size()) {            return false;        }        for (Boolean each : loadCompletedMap.values()) {            if (!each) {                return false;            }        }        return true;    }

到這為止實現(xiàn)數(shù)據(jù)全量同步，其實全量同步最終還是互相調(diào)用Nacos提供的api?？偨Y(jié)一下全量數(shù)據(jù)同步的過程：

啟動一個定時任務(wù)線程DistroLoadDataTask加載數(shù)據(jù)，調(diào)用load()方法加載數(shù)據(jù)調(diào)用loadAllDataSnapshotFromRemote()方法從遠程機器同步所有的數(shù)據(jù)構(gòu)造http請求，調(diào)用httpGet方法從指定的server獲取數(shù)據(jù)同步處理數(shù)據(jù)processData并執(zhí)行監(jiān)聽器listener成功后，就更新data store

7.3.5.2 Nacos集群增量數(shù)據(jù)同步

當(dāng)服務(wù)注冊完成后，Nacos需要將客戶端實例信息同步到Nacos集群其他節(jié)點，可以看一下Nacos底層是怎么實現(xiàn)的。我們再次回到put方法：

public void put(String key, Record value) throws NacosException {        //注冊邏輯：實際就是把實例注冊任務(wù)放到內(nèi)存阻塞隊列中        onPut(key, value);        //AP 架構(gòu)下的節(jié)點數(shù)據(jù)同步        distroProtocol.sync(new DistroKey(key, KeyBuilder.INSTANCE_LIST_KEY_PREFIX), DataOperation.CHANGE,                globalConfig.getTaskDispatchPeriod() / 2);    }

上文中已經(jīng)解釋過put方法中的 onPut(key, value)方法，接下來我們再了解一下AP結(jié)構(gòu)Nacos下的節(jié)點數(shù)據(jù)是同步，也就是distroProtocol.sync方法。

** * Start to sync data to all remote server. * * @param distroKey distro key of sync data * @param action    the action of data operation */public void sync(DistroKey distroKey, DataOperation action, long delay) {    //循環(huán)將新增實例同步到除自己外的所有實例，單機for循環(huán)都不會走，集群架構(gòu)就會走本方法    for (Member each : memberManager.allMembersWithoutSelf()) {        //先把每臺機器都數(shù)據(jù)封裝稱distroKeyWithTarget對象        DistroKey distroKeyWithTarget = new DistroKey(distroKey.getResourceKey(), distroKey.getResourceType(),                each.getAddress());        DistroDelayTask distroDelayTask = new DistroDelayTask(distroKeyWithTarget, action, delay);        //添加到task任務(wù)中        distroTaskEngineHolder.getDelayTaskExecuteEngine().addTask(distroKeyWithTarget, distroDelayTask);        if (Loggers.DISTRO.isDebugEnabled()) {            Loggers.DISTRO.debug("[DISTRO-SCHEDULE] {} to {}", distroKey, each.getAddress());    }}/** * 將集群中除自己外其他需要同步的機器信息添加到一個tasks任務(wù)中，由前面的知識可以猜到這里也是 * 用異步開啟一個線程去拿tasks進行同步新增實例信息到其他nacos機器中 * * @param key  key of task * @param newTask */@Overridepublic void addTask(Object key, AbstractDelayTask newTask) {    lock.lock();    try {        AbstractDelayTask existTask = tasks.get(key);        if (null != existTask) {            newTask.merge(existTask);        }        //將集群中除自己外其他需要同步的機器信息添加到一個tasks任務(wù)中,task是個保存信息的ConcurrentHashMap        tasks.put(key, newTask);    } finally {        lock.unlock();    }}

這里直接把需要同步的信息放在了內(nèi)存的ConcurrentHashMap中，我們看一下這里具體看一下怎么同步其他節(jié)點。

@Componentpublic class DistroTaskEngineHolder {        //延遲任務(wù)執(zhí)行器    private final DistroDelayTaskExecuteEngine delayTaskExecuteEngine = new DistroDelayTaskExecuteEngine();        //任務(wù)執(zhí)行引擎器    private final DistroExecuteTaskExecuteEngine executeWorkersManager = new DistroExecuteTaskExecuteEngine();        public DistroTaskEngineHolder(DistroComponentHolder distroComponentHolder) {        DistroDelayTaskProcessor defaultDelayTaskProcessor = new DistroDelayTaskProcessor(this, distroComponentHolder);        delayTaskExecuteEngine.setDefaultTaskProcessor(defaultDelayTaskProcessor);    }        public DistroDelayTaskExecuteEngine getDelayTaskExecuteEngine() {        return delayTaskExecuteEngine;    }        public DistroExecuteTaskExecuteEngine getExecuteWorkersManager() {        return executeWorkersManager;    }        public void registerNacosTaskProcessor(Object key, NacosTaskProcessor nacosTaskProcessor) {        this.delayTaskExecuteEngine.addProcessor(key, nacosTaskProcessor);    }}

這個類中會創(chuàng)建一個任務(wù)執(zhí)行引擎，代碼如下：

public class DistroDelayTaskExecuteEngine extends NacosDelayTaskExecuteEngine {        public DistroDelayTaskExecuteEngine() {        super(DistroDelayTaskExecuteEngine.class.getName(), Loggers.DISTRO);    }        @Override    public void addProcessor(Object key, NacosTaskProcessor taskProcessor) {        Object actualKey = getActualKey(key);        super.addProcessor(actualKey, taskProcessor);    }        @Override    public NacosTaskProcessor getProcessor(Object key) {        Object actualKey = getActualKey(key);        return super.getProcessor(actualKey);    }        private Object getActualKey(Object key) {        return key instanceof DistroKey ? ((DistroKey) key).getResourceType() : key;    }}//忽略一些中間省略的代碼public NacosDelayTaskExecuteEngine(String name, int initCapacity, Logger logger, long processInterval) {    super(logger);    tasks = new ConcurrentHashMap(initCapacity);    processingExecutor = ExecutorFactory.newSingleScheduledExecutorService(new NameThreadFactory(name));    //這里又執(zhí)行了一個任務(wù)    processingExecutor            .scheduleWithFixedDelay(new ProcessRunnable(), processInterval, processInterval, TimeUnit.MILLISECONDS);}/** * 執(zhí)行了一個任務(wù) */private class ProcessRunnable implements Runnable {        @Override    public void run() {        try {            processTasks();        } catch (Throwable e) {            getEngineLog().error(e.toString(), e);        }    }}

public boolean process(NacosTask task) {    if (!(task instanceof DistroDelayTask)) {        return true;    }    DistroDelayTask distroDelayTask = (DistroDelayTask) task;    DistroKey distroKey = distroDelayTask.getDistroKey();    if (DataOperation.CHANGE.equals(distroDelayTask.getAction())) {        //將延遲任務(wù)變更成異步任務(wù)，異步任務(wù)對象是一個線程        DistroSyncChangeTask syncChangeTask = new DistroSyncChangeTask(distroKey, distroComponentHolder);        //將前面封裝到任務(wù)拿出來放在一個隊列中        distroTaskEngineHolder.getExecuteWorkersManager().addTask(distroKey, syncChangeTask);        return true;    }    return false;}public void addTask(Object tag, AbstractExecuteTask task) {    //拿前面其他實例到任務(wù)    NacosTaskProcessor processor = getProcessor(tag);    if (null != processor) {        processor.process(task);        return;    }    TaskExecuteWorker worker = getWorker(tag);    //將同步數(shù)據(jù)到其他nacos實例到tasks任務(wù)放進一個queue中 在InnerWorker.run()方法中從queue隊列中拿任務(wù)執(zhí)行    worker.process(task);}public boolean process(NacosTask task) {    if (task instanceof AbstractExecuteTask) {        putTask((Runnable) task);    }    return true;}/** * 把任務(wù)同步放進內(nèi)存隊列中 * * @param task */private void putTask(Runnable task) {    try {        queue.put(task);    } catch (InterruptedException ire) {        log.error(ire.toString(), ire);    }}

將同步數(shù)據(jù)到其他Nacos實例到tasks任務(wù)放進一個queue中，然后在InnerWorker.run()方法中從queue隊列中拿任務(wù)執(zhí)行?？匆幌戮唧w是怎么執(zhí)行同步任務(wù)的：

/** * Inner execute worker. */private class InnerWorker extends Thread {        InnerWorker(String name) {        setDaemon(false);        setName(name);    }        @Override    public void run() {        while (!closed.get()) {            try {                //從queue中拿同步任務(wù)                Runnable task = queue.take();                long begin = System.currentTimeMillis();                //實際就是執(zhí)行異步同步任務(wù) DistroSyncChangeTask 的run()                task.run();                long duration = System.currentTimeMillis() - begin;                if (duration > 1000L) {                    log.warn("distro task {} takes {}ms", task, duration);                }            } catch (Throwable e) {                log.error("[DISTRO-FAILED] " + e.toString(), e);            }        }    }}

這里從隊列里面拿出來任務(wù)執(zhí)行，不難發(fā)現(xiàn)這里的任務(wù)執(zhí)行的具體方法就是DistroSyncChangeTask類的run方法：

public class DistroSyncChangeTask extends AbstractDistroExecuteTask {        private final DistroComponentHolder distroComponentHolder;        public DistroSyncChangeTask(DistroKey distroKey, DistroComponentHolder distroComponentHolder) {        super(distroKey);        this.distroComponentHolder = distroComponentHolder;    }        @Override    public void run() {        Loggers.DISTRO.info("[DISTRO-START] {}", toString());        try {            //獲取各種參數(shù)            String type = getDistroKey().getResourceType();            DistroData distroData = distroComponentHolder.findDataStorage(type).getDistroData(getDistroKey());            distroData.setType(DataOperation.CHANGE);            //調(diào)用http接口同步任務(wù)            boolean result = distroComponentHolder.findTransportAgent(type).syncData(distroData, getDistroKey().getTargetServer());            //同步失敗會繼續(xù)重試            if (!result) {                handleFailedTask();            }            Loggers.DISTRO.info("[DISTRO-END] {} result: {}", toString(), result);        } catch (Exception e) {            Loggers.DISTRO.warn("[DISTRO] Sync data change failed.", e);            handleFailedTask();        }    }    /**     * 同步失敗會繼續(xù)重試     */    private void handleFailedTask() {        String type = getDistroKey().getResourceType();        DistroFailedTaskHandler failedTaskHandler = distroComponentHolder.findFailedTaskHandler(type);        if (null == failedTaskHandler) {            Loggers.DISTRO.warn("[DISTRO] Can"t find failed task for type {}, so discarded", type);            return;        }        failedTaskHandler.retry(getDistroKey(), DataOperation.CHANGE);    }        @Override    public String toString() {        return "DistroSyncChangeTask for " + getDistroKey().toString();    }}

每個Nacos服務(wù)端實例都會提供這樣的一個api接口供其他Nacos實例調(diào)用，從而同步注冊實例數(shù)據(jù)。DistroSyncChangeTask類的run方法，就是調(diào)用http接口同步任務(wù)接口，將本節(jié)點的注冊實例數(shù)據(jù)同步到其他節(jié)點機器上。

總結(jié)一下上面增量數(shù)據(jù)同步方法：

DistroProtocol 使用 sync() 方法處理AP 架構(gòu)下的節(jié)點數(shù)據(jù)同步

向其他節(jié)點發(fā)布廣播任務(wù)調(diào)用 distroTaskEngineHolder 發(fā)布延遲任務(wù)

調(diào)用 DistroDelayTaskProcessor.process() 方法進行任務(wù)投遞：將延遲任務(wù)轉(zhuǎn)換為異步變更任務(wù)

執(zhí)行變更任務(wù) DistroSyncChangeTask.run() 方法：向指定節(jié)點發(fā)送消息

8、總結(jié)

Nacos避免并發(fā)讀寫的沖突：Nacos在更新實例列表時，會采用CopyOnWrite技術(shù)，首先將老得實例列表拷貝一份，然后更新拷貝的實例列表，再用更新后的實例列表來覆蓋舊的實例列表。

Nacos提高注冊并發(fā)：為了應(yīng)對阿里巴巴內(nèi)部數(shù)十萬服務(wù)的并發(fā)寫請求Nacos內(nèi)部會將服務(wù)注冊的任務(wù)放入阻塞隊列，采用線程池異步來完成實例更新，從而提高并發(fā)寫能力。

Nacos的服務(wù)發(fā)現(xiàn)分為兩種模式：主動拉取模式，消費者定期主動從Nacos服務(wù)端拉取服務(wù)列表并緩存起來，當(dāng)服務(wù)調(diào)用時優(yōu)先讀取本地緩存中的服務(wù)列表。訂閱模式，消費者訂閱Nacos中的服務(wù)列表，并基于UDP協(xié)議來接收服務(wù)變更通知。當(dāng)Nacos中的服務(wù)列表更新時，會發(fā)送UDP廣播給所有訂閱者。與Eureka相比，Nacos的訂閱模式服務(wù)狀態(tài)更新更及時，消費者更容易及時發(fā)現(xiàn)服務(wù)列表的變化，剔除故障服務(wù)。

關(guān)鍵詞：定時任務(wù) 服務(wù)列表不難發(fā)現(xiàn) 命名空間