今日熱搜：4 個(gè)維度搞懂 Nacos 注冊(cè)中心

發(fā)布時(shí)間：2023-02-28 14:00:36

來源：架構(gòu)師社區(qū)

現(xiàn)如今市面上注冊(cè)中心的輪子很多，我實(shí)際使用過的就有三款：Eureka、Gsched、Nacos，由于當(dāng)前參與 Nacos 集群的維護(hù)和開發(fā)工作，期間也參與了 Nacos 社區(qū)的一些開發(fā)和 Bug Fix 工作，過程中對(duì) Nacos 原理有了一定的積累，今天給大家分享一下 Nacos 動(dòng)態(tài)服務(wù)發(fā)現(xiàn)的原理。

【資料圖】

不 BB，上文章目錄：

01 什么是動(dòng)態(tài)服務(wù)發(fā)現(xiàn)？

服務(wù)發(fā)現(xiàn)是指使用一個(gè)注冊(cè)中心來記錄分布式系統(tǒng)中的全部服務(wù)的信息，以便其他服務(wù)能夠快速的找到這些已注冊(cè)的服務(wù)。

在單體應(yīng)用中，DNS+Nginx 可以滿足服務(wù)發(fā)現(xiàn)的要求，此時(shí)服務(wù)的IP列表配置在 nginx 上。在微服務(wù)架構(gòu)中，由于服務(wù)粒度變的更細(xì)，服務(wù)的上下線更加頻繁，我們需要一款注冊(cè)中心來動(dòng)態(tài)感知服務(wù)的上下線，并且推送IP列表變化給服務(wù)消費(fèi)者，架構(gòu)如下圖。

02 Nacos 實(shí)現(xiàn)動(dòng)態(tài)服務(wù)發(fā)現(xiàn)的原理

Nacos實(shí)現(xiàn)動(dòng)態(tài)服務(wù)發(fā)現(xiàn)的核心原理如下圖，我們接下來的內(nèi)容將圍繞這個(gè)圖來進(jìn)行。

2.1 通訊協(xié)議

整個(gè)服務(wù)注冊(cè)與發(fā)現(xiàn)過程，都離不開通訊協(xié)議，在1.x的 Nacos 版本中服務(wù)端只支持 http 協(xié)議，后來為了提升性能在2.x版本引入了谷歌的 grpc，grpc 是一款長(zhǎng)連接協(xié)議，極大的減少了 http 請(qǐng)求頻繁的連接創(chuàng)建和銷毀過程，能大幅度提升性能，節(jié)約資源。

據(jù)官方測(cè)試，Nacos服務(wù)端 grpc 版本，相比 http 版本的性能提升了9倍以上。

2.2 Nacos 服務(wù)注冊(cè)

簡(jiǎn)單來講，服務(wù)注冊(cè)的目的就是客戶端將自己的ip端口等信息上報(bào)給 Nacos 服務(wù)端，過程如下：

創(chuàng)建長(zhǎng)連接：Nacos SDK 通過Nacos服務(wù)端域名解析出服務(wù)端ip列表，選擇其中一個(gè)ip創(chuàng)建 grpc 連接，并定時(shí)檢查連接狀態(tài)，當(dāng)連接斷開，則自動(dòng)選擇服務(wù)端ip列表中的下一個(gè)ip進(jìn)行重連。

健康檢查請(qǐng)求：在正式發(fā)起注冊(cè)之前，Nacos SDK 向服務(wù)端發(fā)送一個(gè)空請(qǐng)求，服務(wù)端回應(yīng)一個(gè)空請(qǐng)求，若Nacos SDK 未收到服務(wù)端回應(yīng)，則認(rèn)為服務(wù)端不健康，并進(jìn)行一定次數(shù)重試，如果都未收到回應(yīng)，則注冊(cè)失敗。

發(fā)起注冊(cè)：當(dāng)你查看Nacos java SDK的注冊(cè)方法時(shí)，你會(huì)發(fā)現(xiàn)沒有返回值，這是因?yàn)镹acos SDK做了補(bǔ)償機(jī)制，在真實(shí)給服務(wù)端上報(bào)數(shù)據(jù)之前，會(huì)先往緩存中插入一條記錄表示開始注冊(cè)，注冊(cè)成功之后再從緩存中標(biāo)記這條記錄為注冊(cè)成功，當(dāng)注冊(cè)失敗時(shí)，緩存中這條記錄是未注冊(cè)成功的狀態(tài)，Nacos SDK開啟了一個(gè)定時(shí)任務(wù)，定時(shí)查詢異常的緩存數(shù)據(jù)，重新發(fā)起注冊(cè)。

Nacos SDK注冊(cè)失敗時(shí)的自動(dòng)補(bǔ)償機(jī)制時(shí)序圖。

相關(guān)源碼如下：

@OverridepublicvoidregisterService(StringserviceName,StringgroupName,Instanceinstance)throwsNacosException{NAMING_LOGGER.info(\"[REGISTER-SERVICE]{}registeringservice{}withinstance{}\",namespaceId,serviceName,instance);//添加redo日志redoService.cacheInstanceForRedo(serviceName,groupName,instance);doRegisterService(serviceName,groupName,instance);}publicvoiddoRegisterService(StringserviceName,StringgroupName,Instanceinstance)throwsNacosException{//向服務(wù)端發(fā)起注冊(cè)InstanceRequestrequest=newInstanceRequest(namespaceId,serviceName,groupName,NamingRemoteConstants.REGISTER_INSTANCE,instance);requestToServer(request,Response.class);//標(biāo)記注冊(cè)成功redoService.instanceRegistered(serviceName,groupName);}

執(zhí)行補(bǔ)償定時(shí)任務(wù)RedoScheduledTask。

@Overridepublicvoidrun(){if(!redoService.isConnected()){LogUtils.NAMING_LOGGER.warn(\"GrpcConnectionisdisconnect,skipcurrentredotask\");return;}try{redoForInstances();redoForSubscribes();}catch(Exceptione){LogUtils.NAMING_LOGGER.warn(\"Redotaskrunwithunexpectedexception:\",e);}}privatevoidredoForInstances(){for(InstanceRedoDataeach:redoService.findInstanceRedoData()){try{redoForInstance(each);}catch(NacosExceptione){LogUtils.NAMING_LOGGER.error(\"Redoinstanceoperation{}for{}@@{}failed.\",each.getRedoType(),each.getGroupName(),each.getServiceName(),e);}}}

服務(wù)端數(shù)據(jù)同步(Distro協(xié)議)：Nacos SDK只會(huì)與服務(wù)端某個(gè)節(jié)點(diǎn)建立長(zhǎng)連接，當(dāng)服務(wù)端接受到客戶端注冊(cè)的實(shí)例數(shù)據(jù)后，還需要將實(shí)例數(shù)據(jù)同步給其他節(jié)點(diǎn)。Nacos自己實(shí)現(xiàn)了一個(gè)一致性協(xié)議名為Distro，服務(wù)注冊(cè)的時(shí)候會(huì)觸發(fā)Distro一次同步，每個(gè)Nacos節(jié)點(diǎn)之間會(huì)定時(shí)互相發(fā)送Distro數(shù)據(jù)，以此保證數(shù)據(jù)最終一致。

服務(wù)實(shí)例上線推送：Nacos服務(wù)端收到服務(wù)實(shí)例數(shù)據(jù)后會(huì)將服務(wù)的最新實(shí)例列表通過grpc推送給該服務(wù)的所有訂閱者。

服務(wù)注冊(cè)過程源碼時(shí)序圖：整理了一下服務(wù)注冊(cè)過程整體時(shí)序圖，對(duì)源碼實(shí)現(xiàn)感興趣的可以按照根據(jù)這個(gè)時(shí)序圖view一下源碼。

2.3 Nacos 心跳機(jī)制

目前主流的注冊(cè)中心，比如Consul、Eureka、Zk包括我們公司自研的Gsched，都是通過心跳機(jī)制來感知服務(wù)的下線。Nacos也是通過心跳機(jī)制來實(shí)現(xiàn)的。

Nacos目前SDK維護(hù)了兩個(gè)分支的版本（1.x、2.x），這兩個(gè)版本心跳機(jī)制的實(shí)現(xiàn)不一樣。其中1.x版本的SDK通過http協(xié)議來定時(shí)向服務(wù)端發(fā)送心跳維持自己的健康狀態(tài)，2.x版本的SDK則通過grpc自身的心跳機(jī)制來?；睿?dāng)Nacos服務(wù)端接受不到服務(wù)實(shí)例的心跳，會(huì)認(rèn)為實(shí)例下線。如下圖：

grpc監(jiān)測(cè)到連接斷開事件，發(fā)送ClientDisconnectEvent。

publicclassConnectionBasedClientManagerextendsClientConnectionEventListenerimplementsClientManager{//連接斷開，發(fā)送連接斷開事件publicbooleanclientDisconnected(StringclientId){Loggers.SRV_LOG.info(\"Clientconnection{}disconnect,removeinstancesandsubscribers\",clientId);ConnectionBasedClientclient=clients.remove(clientId);if(null==client){returntrue;}client.release();NotifyCenter.publishEvent(newClientEvent.ClientDisconnectEvent(client));returntrue;}}

移除客戶端注冊(cè)的服務(wù)實(shí)例

publicclassClientServiceIndexesManagerextendsSmartSubscriber{@OverridepublicvoidonEvent(Eventevent){//接收失去連接事件if(eventinstanceofClientEvent.ClientDisconnectEvent){handleClientDisconnect((ClientEvent.ClientDisconnectEvent)event);}elseif(eventinstanceofClientOperationEvent){handleClientOperation((ClientOperationEvent)event);}}privatevoidhandleClientDisconnect(ClientEvent.ClientDisconnectEventevent){Clientclient=event.getClient();for(Serviceeach:client.getAllSubscribeService()){removeSubscriberIndexes(each,client.getClientId());}//移除客戶端注冊(cè)的服務(wù)實(shí)例for(Serviceeach:client.getAllPublishedService()){removePublisherIndexes(each,client.getClientId());}}//移除客戶端注冊(cè)的服務(wù)實(shí)例privatevoidremovePublisherIndexes(Serviceservice,StringclientId){if(!publisherIndexes.containsKey(service)){return;}publisherIndexes.get(service).remove(clientId);NotifyCenter.publishEvent(newServiceEvent.ServiceChangedEvent(service,true));}}

2.4 Nacos 服務(wù)訂閱

當(dāng)一個(gè)服務(wù)發(fā)生上下線，Nacos如何知道要推送給哪些客戶端？

Nacos SDK 提供了訂閱和取消訂閱方法，當(dāng)客戶端向服務(wù)端發(fā)起訂閱請(qǐng)求，服務(wù)端會(huì)記錄發(fā)起調(diào)用的客戶端為該服務(wù)的訂閱者，同時(shí)將服務(wù)的最新實(shí)例列表返回。當(dāng)客戶端發(fā)起了取消訂閱，服務(wù)端就會(huì)從該服務(wù)的訂閱者列表中把當(dāng)前客戶端移除。

當(dāng)客戶端發(fā)起訂閱時(shí)，服務(wù)端除了會(huì)同步返回最新的服務(wù)實(shí)例列表，還會(huì)異步的通過grpc推送給該訂閱者最新的服務(wù)實(shí)例列表，這樣做的目的是為了異步更新客戶端本地緩存的服務(wù)數(shù)據(jù)。

當(dāng)客戶端訂閱的服務(wù)上下線，該服務(wù)所有的訂閱者會(huì)立刻收到最新的服務(wù)列表并且將服務(wù)最新的實(shí)例數(shù)據(jù)更新到內(nèi)存。

我們也看一下相關(guān)源碼，服務(wù)端接收到訂閱數(shù)據(jù)，首先保存到內(nèi)存中。

@OverridepublicvoidsubscribeService(Serviceservice,Subscribersubscriber,StringclientId){Servicesingleton=ServiceManager.getInstance().getSingletonIfExist(service).orElse(service);Clientclient=clientManager.getClient(clientId);//校驗(yàn)長(zhǎng)連接是否正常if(!clientIsLegal(client,clientId)){return;}//保存訂閱數(shù)據(jù)client.addServiceSubscriber(singleton,subscriber);client.setLastUpdatedTime();//發(fā)送訂閱事件NotifyCenter.publishEvent(newClientOperationEvent.ClientSubscribeServiceEvent(singleton,clientId));}privatevoidhandleClientOperation(ClientOperationEventevent){Serviceservice=event.getService();StringclientId=event.getClientId();if(eventinstanceofClientOperationEvent.ClientRegisterServiceEvent){addPublisherIndexes(service,clientId);}elseif(eventinstanceofClientOperationEvent.ClientDeregisterServiceEvent){removePublisherIndexes(service,clientId);}elseif(eventinstanceofClientOperationEvent.ClientSubscribeServiceEvent){//處理訂閱操作addSubscriberIndexes(service,clientId);}elseif(eventinstanceofClientOperationEvent.ClientUnsubscribeServiceEvent){removeSubscriberIndexes(service,clientId);}}

然后發(fā)布訂閱事件。

privatevoidaddSubscriberIndexes(Serviceservice,StringclientId){//保存訂閱數(shù)據(jù)subscriberIndexes.computeIfAbsent(service,(key)->newConcurrentHashSet<>());//Fix#5404,Onlyfirsttimeaddneednotifyevent.if(subscriberIndexes.get(service).add(clientId)){//發(fā)布訂閱事件NotifyCenter.publishEvent(newServiceEvent.ServiceSubscribedEvent(service,clientId));}}

服務(wù)端自己消費(fèi)訂閱事件，并且推送給訂閱的客戶端最新的服務(wù)實(shí)例數(shù)據(jù)。

@OverridepublicvoidonEvent(Eventevent){if(!upgradeJudgement.isUseGrpcFeatures()){return;}if(eventinstanceofServiceEvent.ServiceChangedEvent){//Ifservicechanged,pushtoallsubscribers.ServiceEvent.ServiceChangedEventserviceChangedEvent=(ServiceEvent.ServiceChangedEvent)event;Serviceservice=serviceChangedEvent.getService();delayTaskEngine.addTask(service,newPushDelayTask(service,PushConfig.getInstance().getPushTaskDelay()));}elseif(eventinstanceofServiceEvent.ServiceSubscribedEvent){//Ifserviceissubscribedbyoneclient,onlypushthisclient.ServiceEvent.ServiceSubscribedEventsubscribedEvent=(ServiceEvent.ServiceSubscribedEvent)event;Serviceservice=subscribedEvent.getService();delayTaskEngine.addTask(service,newPushDelayTask(service,PushConfig.getInstance().getPushTaskDelay(),subscribedEvent.getClientId()));}}

2.5 Nacos 推送

推送方式

前面說了服務(wù)的注冊(cè)和訂閱都會(huì)發(fā)生推送（服務(wù)端->客戶端），那推送到底是如何實(shí)現(xiàn)的呢？

在早期的Nacos版本，當(dāng)服務(wù)實(shí)例變化，服務(wù)端會(huì)通過udp協(xié)議將最新的數(shù)據(jù)發(fā)送給客戶端，后來發(fā)現(xiàn)udp推送有一定的丟包率，于是新版本的Nacos支持了grpc推送。Nacos服務(wù)端會(huì)自動(dòng)判斷客戶端的版本來選擇哪種方式來進(jìn)行推送，如果你使用1.4.2以前的SDK進(jìn)行注冊(cè)，那Nacos服務(wù)端會(huì)使用udp協(xié)議來進(jìn)行推送，反之則使用grpc。

推送失敗重試

當(dāng)發(fā)送推送時(shí)，客戶端可能正在重啟，或者連接不穩(wěn)定導(dǎo)致推送失敗，這個(gè)時(shí)候Nacos會(huì)進(jìn)行重試。Nacos將每個(gè)推送都封裝成一個(gè)任務(wù)對(duì)象，放入到隊(duì)列中，再開啟一個(gè)線程不停的從隊(duì)列取出任務(wù)執(zhí)行，執(zhí)行之前會(huì)先刪除該任務(wù)，如果執(zhí)行失敗則將任務(wù)重新添加到隊(duì)列，該線程會(huì)記錄任務(wù)執(zhí)行的時(shí)間，如果超過1秒，則會(huì)記錄到日志。

推送部分源碼

添加推送任務(wù)到執(zhí)行隊(duì)列中。

privatestaticclassPushDelayTaskProcessorimplementsNacosTaskProcessor{privatefinalPushDelayTaskExecuteEngineexecuteEngine;publicPushDelayTaskProcessor(PushDelayTaskExecuteEngineexecuteEngine){this.executeEngine=executeEngine;}@Overridepublicbooleanprocess(NacosTasktask){PushDelayTaskpushDelayTask=(PushDelayTask)task;Serviceservice=pushDelayTask.getService();NamingExecuteTaskDispatcher.getInstance().dispatchAndExecuteTask(service,newPushExecuteTask(service,executeEngine,pushDelayTask));returntrue;}}

推送任務(wù)PushExecuteTask 的執(zhí)行。

publicclassPushExecuteTaskextendsAbstractExecuteTask{//..省略@Overridepublicvoidrun(){try{//封裝要推送的服務(wù)實(shí)例數(shù)據(jù)PushDataWrapperwrapper=generatePushData();ClientManagerclientManager=delayTaskEngine.getClientManager();//如果是服務(wù)上下線導(dǎo)致的推送，獲取所有訂閱者//如果是訂閱導(dǎo)致的推送，獲取訂閱者for(Stringeach:getTargetClientIds()){Clientclient=clientManager.getClient(each);if(null==client){//meansthisclienthasdisconnectcontinue;}Subscribersubscriber=clientManager.getClient(each).getSubscriber(service);//推送給訂閱者delayTaskEngine.getPushExecutor().doPushWithCallback(each,subscriber,wrapper,newNamingPushCallback(each,subscriber,wrapper.getOriginalData(),delayTask.isPushToAll()));}}catch(Exceptione){Loggers.PUSH.error(\"Pushtaskforservice\"+service.getGroupedServiceName()+\"executefailed\",e);//當(dāng)推送發(fā)生異常，重新將推送任務(wù)放入執(zhí)行隊(duì)列delayTaskEngine.addTask(service,newPushDelayTask(service,1000L));}}//如果是服務(wù)上下線導(dǎo)致的推送，獲取所有訂閱者//如果是訂閱導(dǎo)致的推送，獲取訂閱者privateCollectiongetTargetClientIds(){returndelayTask.isPushToAll()?delayTaskEngine.getIndexesManager().getAllClientsSubscribeService(service):delayTask.getTargetClients();}

執(zhí)行推送任務(wù)線程InnerWorker 的執(zhí)行。

/***Innerexecuteworker.*/privateclassInnerWorkerextendsThread{InnerWorker(Stringname){setDaemon(false);setName(name);}@Overridepublicvoidrun(){while(!closed.get()){try{//從隊(duì)列中取出任務(wù)PushExecuteTaskRunnabletask=queue.take();longbegin=System.currentTimeMillis();//執(zhí)行PushExecuteTasktask.run();longduration=System.currentTimeMillis()-begin;if(duration>1000L){log.warn(\"task{}takes{}ms\",task,duration);}}catch(Throwablee){log.error(\"[TASK-FAILED]\"+e.toString(),e);}}}}

2.6 Nacos SDK 查詢服務(wù)實(shí)例

服務(wù)消費(fèi)者首先需要調(diào)用Nacos SDK的接口來獲取最新的服務(wù)實(shí)例，然后才能從獲取到的實(shí)例列表中以加權(quán)輪詢的方式選擇出一個(gè)實(shí)例（包含ip，port等信息），最后再發(fā)起調(diào)用。

前面已經(jīng)提到Nacos服務(wù)發(fā)生上下線、訂閱的時(shí)候都會(huì)推送最新的服務(wù)實(shí)例列表到當(dāng)客戶端，客戶端再更新本地內(nèi)存中的緩沖數(shù)據(jù)，所以調(diào)用Nacos SDK提供的查詢實(shí)例列表的接口時(shí)，不會(huì)直接請(qǐng)求服務(wù)端獲取數(shù)據(jù)，而是會(huì)優(yōu)先使用內(nèi)存中的服務(wù)數(shù)據(jù)，只有內(nèi)存中查不到的情況下才會(huì)發(fā)起訂閱請(qǐng)求服務(wù)端數(shù)據(jù)。

Nacos SDK內(nèi)存中的數(shù)據(jù)除了接受來自服務(wù)端的推送更新之外，自己本地也會(huì)有一個(gè)定時(shí)任務(wù)定時(shí)去獲取服務(wù)端數(shù)據(jù)來進(jìn)行兜底。Nacos SDK在查詢的時(shí)候也了容災(zāi)機(jī)制，即從磁盤獲取服務(wù)數(shù)據(jù)，而這個(gè)磁盤的數(shù)據(jù)其實(shí)也是來自于內(nèi)存，有一個(gè)定時(shí)任務(wù)定時(shí)從內(nèi)存緩存中獲取然后加載到磁盤。Nacos SDK的容災(zāi)機(jī)制默認(rèn)關(guān)閉，可通過設(shè)置環(huán)境變量failover-mode=true來開啟。

架構(gòu)圖

用戶查詢流程

查詢服務(wù)實(shí)例部分源碼

privatefinalConcurrentMapserviceInfoMap;@OverridepublicListgetAllInstances(StringserviceName,StringgroupName,Listclusters,booleansubscribe)throwsNacosException{ServiceInfoserviceInfo;StringclusterString=StringUtils.join(clusters,\",\");//這里默認(rèn)傳過來是trueif(subscribe){//從本地內(nèi)存獲取服務(wù)數(shù)據(jù)，如果獲取不到則從磁盤獲取serviceInfo=serviceInfoHolder.getServiceInfo(serviceName,groupName,clusterString);if(null==serviceInfo||!clientProxy.isSubscribed(serviceName,groupName,clusterString)){//如果從本地獲取不到數(shù)據(jù)，則調(diào)用訂閱方法serviceInfo=clientProxy.subscribe(serviceName,groupName,clusterString);}}else{//適用于不走訂閱，直接從服務(wù)端獲取數(shù)據(jù)的情況serviceInfo=clientProxy.queryInstancesOfService(serviceName,groupName,clusterString,0,false);}Listlist;if(serviceInfo==null||CollectionUtils.isEmpty(list=serviceInfo.getHosts())){returnnewArrayList();}returnlist;}}//從本地內(nèi)存獲取服務(wù)數(shù)據(jù)，如果開啟了故障轉(zhuǎn)移則直接從磁盤獲取，因?yàn)楫?dāng)服務(wù)端掛了，本地啟動(dòng)時(shí)內(nèi)存中也沒有數(shù)據(jù)publicServiceInfogetServiceInfo(finalStringserviceName,finalStringgroupName,finalStringclusters){NAMING_LOGGER.debug(\"failover-mode:{}\",failoverReactor.isFailoverSwitch());StringgroupedServiceName=NamingUtils.getGroupedName(serviceName,groupName);Stringkey=ServiceInfo.getKey(groupedServiceName,clusters);//故障轉(zhuǎn)移則直接從磁盤獲取if(failoverReactor.isFailoverSwitch()){returnfailoverReactor.getService(key);}//返回內(nèi)存中數(shù)據(jù)returnserviceInfoMap.get(key);}