[[417043]]

本文轉(zhuǎn)載自微信公眾號(hào)「程序新視界」，作者二師兄。轉(zhuǎn)載本文請(qǐng)聯(lián)系程序新視界公眾號(hào)。

說(shuō)起Nacos的服務(wù)訂閱機(jī)制，對(duì)此不了解的朋友，可能感覺(jué)非常神秘，這篇文章就大家深入淺出的了解一下Nacos 2.0客戶端的訂閱實(shí)現(xiàn)。由于涉及到的內(nèi)容比較多，就分幾篇來(lái)講，本篇為第一篇。

Nacos訂閱概述

Nacos的訂閱機(jī)制，如果用一句話來(lái)描述就是：Nacos客戶端通過(guò)一個(gè)定時(shí)任務(wù)，每6秒從注冊(cè)中心獲取實(shí)例列表，當(dāng)發(fā)現(xiàn)實(shí)例發(fā)生變化時(shí)，發(fā)布變更事件，訂閱者進(jìn)行業(yè)務(wù)處理。該更新實(shí)例的更新實(shí)例，該更新本地緩存的更新本地緩存。

nacos

上圖畫(huà)出了訂閱方法的主線流程，涉及的內(nèi)容較多，處理細(xì)節(jié)復(fù)雜。這里只用把握住核心部分即可。下面就通過(guò)代碼和流程圖來(lái)逐步分析上述過(guò)程。

從訂閱到定時(shí)任務(wù)開(kāi)啟

我們這里聊的訂閱機(jī)制，其實(shí)本質(zhì)上就是服務(wù)發(fā)現(xiàn)的準(zhǔn)實(shí)時(shí)感知。上面已經(jīng)看到了當(dāng)執(zhí)行訂閱方法時(shí)，會(huì)觸發(fā)定時(shí)任務(wù)，定時(shí)去拉服務(wù)器端的數(shù)據(jù)。所以，本質(zhì)上，訂閱機(jī)制就是實(shí)現(xiàn)服務(wù)發(fā)現(xiàn)的一種方式，對(duì)照的方式就是直接查詢(xún)接口了。

NacosNamingService中暴露的許多重載的subscribe，重載的目的就是讓大家少寫(xiě)一些參數(shù)，這些參數(shù)呢，Nacos給默認(rèn)處理了。最終這些重載方法都會(huì)調(diào)用到下面這個(gè)方法：

// NacosNamingService 
public void subscribe(String serviceName, String groupName, List<String> clusters, EventListener listener) 
        throws NacosException { 
    if (null == listener) { 
        return; 
    } 
    String clusterString = StringUtils.join(clusters, ","); 
    changeNotifier.registerListener(groupName, serviceName, clusterString, listener); 
    clientProxy.subscribe(serviceName, groupName, clusterString); 
} 

方法中的事件監(jiān)聽(tīng)我們暫時(shí)不聊，直接看subscribe方法，這里clientProxy類(lèi)型為NamingClientProxyDelegate。實(shí)例化NacosNamingService時(shí)該類(lèi)被實(shí)例化，前面章節(jié)中已經(jīng)講到，不再贅述。

而clientProxy.subscribe方法在NamingClientProxyDelegate中實(shí)現(xiàn)：

// NamingClientProxyDelegate 
@Override 
public ServiceInfo subscribe(String serviceName, String groupName, String clusters) throws NacosException { 
    String serviceNameWithGroup = NamingUtils.getGroupedName(serviceName, groupName); 
    String serviceKey = ServiceInfo.getKey(serviceNameWithGroup, clusters); 
    // 獲取緩存中的ServiceInfo 
    ServiceInfo result = serviceInfoHolder.getServiceInfoMap().get(serviceKey); 
    if (null == result) { 
        // 如果為null，則進(jìn)行訂閱邏輯處理，基于gRPC協(xié)議 
        result = grpcClientProxy.subscribe(serviceName, groupName, clusters); 
    } 
    // 定時(shí)調(diào)度UpdateTask 
    serviceInfoUpdateService.scheduleUpdateIfAbsent(serviceName, groupName, clusters); 
    // ServiceInfo本地緩存處理 
    serviceInfoHolder.processServiceInfo(result); 
    return result; 
} 

這段方法是不是眼熟啊?對(duì)的，在前面分析《Nacos Client服務(wù)發(fā)現(xiàn)》時(shí)我們已經(jīng)講過(guò)了?？磥?lái)殊途同歸，查詢(xún)服務(wù)列表和訂閱最終都調(diào)用了同一個(gè)方法。

上篇講了其他流程，我們這里重點(diǎn)看任務(wù)調(diào)度：

// ServiceInfoUpdateService 
public void scheduleUpdateIfAbsent(String serviceName, String groupName, String clusters) { 
    String serviceKey = ServiceInfo.getKey(NamingUtils.getGroupedName(serviceName, groupName), clusters); 
    if (futureMap.get(serviceKey) != null) { 
        return; 
    } 
    synchronized (futureMap) { 
        if (futureMap.get(serviceKey) != null) { 
            return; 
        } 
        // 構(gòu)建UpdateTask 
        ScheduledFuture<?> future = addTask(new UpdateTask(serviceName, groupName, clusters)); 
        futureMap.put(serviceKey, future); 
    } 
} 

該方法包含了構(gòu)建serviceKey、通過(guò)serviceKey判重，最后添加UpdateTask。

而其中的addTask的實(shí)現(xiàn)就是發(fā)起了一個(gè)定時(shí)任務(wù)：

// ServiceInfoUpdateService 
private synchronized ScheduledFuture<?> addTask(UpdateTask task) { 
    return executor.schedule(task, DEFAULT_DELAY, TimeUnit.MILLISECONDS); 
} 

定時(shí)任務(wù)延時(shí)1秒執(zhí)行。

跟蹤到這里就告一階段了。核心功能只有兩個(gè)：調(diào)用訂閱方法和發(fā)起定時(shí)任務(wù)。

定時(shí)任務(wù)都干了啥

UpdateTask封裝了訂閱機(jī)制的核心業(yè)務(wù)邏輯，先來(lái)通過(guò)一張流程圖看一下都做了啥。

nacos

有了上述流程圖，基本就很清晰的了解UpdateTask所做的事情了。直接貼出run方法的所有代碼：

public void run() { 
    long delayTime = DEFAULT_DELAY; 
 
    try { 
        // 判斷該注冊(cè)的Service是否被訂閱，如果沒(méi)有訂閱則不再執(zhí)行 
        if (!changeNotifier.isSubscribed(groupName, serviceName, clusters) && !futureMap.containsKey(serviceKey)) { 
            NAMING_LOGGER 
                    .info("update task is stopped, service:" + groupedServiceName + ", clusters:" + clusters); 
            return; 
        } 
 
        // 獲取緩存的service信息 
        ServiceInfo serviceObj = serviceInfoHolder.getServiceInfoMap().get(serviceKey); 
        if (serviceObj == null) { 
            // 根據(jù)serviceName從注冊(cè)中心服務(wù)端獲取Service信息 
            serviceObj = namingClientProxy.queryInstancesOfService(serviceName, groupName, clusters, 0, false); 
            serviceInfoHolder.processServiceInfo(serviceObj); 
            lastRefTime = serviceObj.getLastRefTime(); 
            return; 
        } 
 
        // 過(guò)期服務(wù)（服務(wù)的最新更新時(shí)間小于等于緩存刷新時(shí)間），從注冊(cè)中心重新查詢(xún) 
        if (serviceObj.getLastRefTime() <= lastRefTime) { 
            serviceObj = namingClientProxy.queryInstancesOfService(serviceName, groupName, clusters, 0, false); 
            // 處理Service消息 
            serviceInfoHolder.processServiceInfo(serviceObj); 
        } 
        // 刷新更新時(shí)間 
        lastRefTime = serviceObj.getLastRefTime(); 
        if (CollectionUtils.isEmpty(serviceObj.getHosts())) { 
            incFailCount(); 
            return; 
        } 
        // 下次更新緩存時(shí)間設(shè)置，默認(rèn)為6秒 
        // TODO multiple time can be configured. 
        delayTime = serviceObj.getCacheMillis() * DEFAULT_UPDATE_CACHE_TIME_MULTIPLE; 
        // 重置失敗數(shù)量為0 
        resetFailCount(); 
    } catch (Throwable e) { 
        incFailCount(); 
        NAMING_LOGGER.warn("[NA] failed to update serviceName: " + groupedServiceName, e); 
    } finally { 
        // 下次調(diào)度刷新時(shí)間，下次執(zhí)行的時(shí)間與failCount有關(guān) 
        // failCount=0，則下次調(diào)度時(shí)間為6秒，最長(zhǎng)為1分鐘 
        // 即當(dāng)無(wú)異常情況下緩存實(shí)例的刷新時(shí)間是6秒 
        executor.schedule(this, Math.min(delayTime << failCount, DEFAULT_DELAY * 60), TimeUnit.MILLISECONDS); 
    } 
} 

首先在判斷服務(wù)是否是被訂閱過(guò)，實(shí)現(xiàn)方法是ChangeNotifier#isSubscribed：

public boolean isSubscribed(String groupName, String serviceName, String clusters) { 
    String key = ServiceInfo.getKey(NamingUtils.getGroupedName(serviceName, groupName), clusters); 
    ConcurrentHashSet<EventListener> eventListeners = listenerMap.get(key); 
    return CollectionUtils.isNotEmpty(eventListeners); 
} 

查看該方法的源碼會(huì)發(fā)現(xiàn)，這里的listenerMap正是最開(kāi)始的subscribe方法中registerListener注冊(cè)的EventListener。

run方法后面的業(yè)務(wù)處理基本上都雷同了，先判斷緩存是否有ServiceInfo信息，如果沒(méi)有則查詢(xún)注冊(cè)中心、處理ServiceInfo、更新上次處理時(shí)間。

而下面判斷ServiceInfo是否失效，正是通過(guò)“上次更新時(shí)間”與當(dāng)前ServiceInfo中的“上次更新時(shí)間”做比較來(lái)判斷。如果失效，也會(huì)查詢(xún)注冊(cè)中心、處理ServiceInfo、更新上次處理時(shí)間等一系列操作。

業(yè)務(wù)邏輯最后會(huì)計(jì)算下一次定時(shí)任務(wù)的執(zhí)行時(shí)間，通過(guò)delayTime來(lái)延遲執(zhí)行。delayTime默認(rèn)為 1000L * 6，也就是6秒。而在finally里面真的發(fā)起下一次定時(shí)任務(wù)。當(dāng)出現(xiàn)異常時(shí)，下次執(zhí)行的時(shí)間與失敗次數(shù)有關(guān)，但最長(zhǎng)不超過(guò)1分鐘。

小結(jié)

這一篇我們講了Nacos客戶端服務(wù)訂閱機(jī)制的源碼，主要有以下步驟：

第一步：訂閱方法的調(diào)用，并進(jìn)行EventListener的注冊(cè)，后面UpdateTask要用來(lái)進(jìn)行判斷;

第二步：通過(guò)委托代理類(lèi)來(lái)處理訂閱邏輯，此處與獲取實(shí)例列表方法使用了同一個(gè)方法;

第三步：通過(guò)定時(shí)任務(wù)執(zhí)行UpdateTask方法，默認(rèn)執(zhí)行間隔為6秒，當(dāng)發(fā)生異常時(shí)會(huì)延長(zhǎng)，但不超過(guò)1分鐘;

第四步：UpdateTask方法中會(huì)比較本地是否存在緩存，緩存是否過(guò)期。當(dāng)不存在或過(guò)期時(shí)，查詢(xún)注冊(cè)中心，獲取最新實(shí)例，更新最后獲取時(shí)間，處理ServiceInfo。

第五步：重新計(jì)算定時(shí)任務(wù)時(shí)間，循環(huán)執(zhí)行上述流程。