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

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

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

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

基于 Spring Boot + MyBatis Plus + Vue & Element 實(shí)現(xiàn)的后臺(tái)管理系統(tǒng) + 用戶小程序，支持 RBAC 動(dòng)態(tài)權(quán)限、多租戶、數(shù)據(jù)權(quán)限、工作流、三方登錄、支付、短信、商城等功能

項(xiàng)目地址：https://github.com/YunaiV/ruoyi-vue-pro

視頻教程：https://doc.iocoder.cn/video/

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

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

2.2 Nacos 服務(wù)注冊

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

創(chuà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)行重連。

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

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

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

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

@Override
publicvoidregisterService(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ā)起注冊
InstanceRequestrequest=newInstanceRequest(namespaceId,serviceName,groupName,
NamingRemoteConstants.REGISTER_INSTANCE,instance);
requestToServer(request,Response.class);
//標(biāo)記注冊成功
redoService.instanceRegistered(serviceName,groupName);
}

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

@Override
publicvoidrun(){
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)建立長連接，當(dāng)服務(wù)端接受到客戶端注冊的實(shí)例數(shù)據(jù)后，還需要將實(shí)例數(shù)據(jù)同步給其他節(jié)點(diǎn)。Nacos自己實(shí)現(xiàn)了一個(gè)一致性協(xié)議名為Distro，服務(wù)注冊的時(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ù)注冊過程源碼時(shí)序圖：整理了一下服務(wù)注冊過程整體時(shí)序圖，對源碼實(shí)現(xiàn)感興趣的可以按照根據(jù)這個(gè)時(shí)序圖view一下源碼。

2.3 Nacos 心跳機(jī)制

目前主流的注冊中心，比如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)測到連接斷開事件，發(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;
}
}

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

publicclassClientServiceIndexesManagerextendsSmartSubscriber{

@Override
publicvoidonEvent(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());
}
//移除客戶端注冊的服務(wù)實(shí)例
for(Serviceeach:client.getAllPublishedService()){
removePublisherIndexes(each,client.getClientId());
}
}

//移除客戶端注冊的服務(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ā)起訂閱請求，服務(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)存中。

@Override
publicvoidsubscribeService(Serviceservice,Subscribersubscriber,StringclientId){
Servicesingleton=ServiceManager.getInstance().getSingletonIfExist(service).orElse(service);
Clientclient=clientManager.getClient(clientId);
//校驗(yàn)長連接是否正常
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ù)。

@Override
publicvoidonEvent(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ù)的注冊和訂閱都會(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)行注冊，那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ì)列中，再開啟一個(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;
}

@Override
publicbooleanprocess(NacosTasktask){
PushDelayTaskpushDelayTask=(PushDelayTask)task;
Serviceservice=pushDelayTask.getService();
NamingExecuteTaskDispatcher.getInstance()
.dispatchAndExecuteTask(service,newPushExecuteTask(service,executeEngine,pushDelayTask));
returntrue;
}
}

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

publicclassPushExecuteTaskextendsAbstractExecuteTask{

//..省略

@Override
publicvoidrun(){
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){
//meansthisclienthasdisconnect
continue;
}
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);
}

@Override
publicvoidrun(){
while(!closed.get()){
try{
//從隊(duì)列中取出任務(wù)PushExecuteTask
Runnabletask=queue.take();
longbegin=System.currentTimeMillis();
//執(zhí)行PushExecuteTask
task.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ì)直接請求服務(wù)端獲取數(shù)據(jù)，而是會(huì)優(yōu)先使用內(nèi)存中的服務(wù)數(shù)據(jù)，只有內(nèi)存中查不到的情況下才會(huì)發(fā)起訂閱請求服務(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;
@Override
publicListgetAllInstances(StringserviceName,StringgroupName,Listclusters,
booleansubscribe)throwsNacosException{
ServiceInfoserviceInfo;
StringclusterString=StringUtils.join(clusters,",");
//這里默認(rèn)傳過來是true
if(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);
}

基于 Spring Cloud Alibaba + Gateway + Nacos + RocketMQ + Vue & Element 實(shí)現(xiàn)的后臺(tái)管理系統(tǒng) + 用戶小程序，支持 RBAC 動(dòng)態(tài)權(quán)限、多租戶、數(shù)據(jù)權(quán)限、工作流、三方登錄、支付、短信、商城等功能

項(xiàng)目地址：https://github.com/YunaiV/yudao-cloud

視頻教程：https://doc.iocoder.cn/video/

03 結(jié)語

本篇文章向大家介紹 Nacos 服務(wù)發(fā)現(xiàn)的基本概念和核心能力以及實(shí)現(xiàn)的原理，旨在讓大家對 Nacos 的服務(wù)注冊與發(fā)現(xiàn)功能有更多的了解，做到心中有數(shù)。