vivo 互聯網服務器團隊- Li Wanghong

本文通過分析Dubbo中ZooKeeper注冊中心的實作ZooKeeperResitry的繼承體系結構，自頂向下分析了AbstractRegistry（提供了服務資料的本地快取）、FailbackRegistry（服務注冊訂閱相關的例外重試）、CacheableFailbackRegistry（Dubbo在URL推送模型做的優化）、ZooKeeperRegistry（ZooKeeper注冊中心實作原理）的原始碼，詳細介紹了Dubbo中ZooKeeper注冊中心的實作原理，

當服務提供者啟動時，服務提供者將自己提供的服務資訊注冊到注冊中心，注冊中心將這些資訊記錄下來，服務消費者啟動時，向注冊中心發起訂閱，將自己感興趣的服務提供者的資訊拉取到本地并快取起來，用于后續遠程呼叫，另外，注冊中心能夠感知到服務提供者新增或者下線，并更新自己的服務資訊，同時通知服務消費者告知服務提供者新增/下線，從而消費者也可以動態感知到服務提供者的變化，

一、ZooKeeper注冊中心

ZooKeeper 是 Apache 的頂級專案，ZooKeeper 為分布式應用提供了高效且可靠的分布式協調服務，提供了諸如統一命名服務、配置管理和分布式鎖等分布式的基礎服務，在解決分布式資料一致性方面，ZooKeeper 并沒有直接采用 Paxos 演算法，而是采用了名為 ZAB 的一致性協議，

1.1 ZooKeeper資料結構

ZooKeeper 的資料模型是一個樹形結構的檔案系統，Dubbo服務在ZooKeeper中的資料結構（舊版模型，Dubbo2.7版本后路由資料、配置資料不再寫到該目錄下）如下圖所示，

ZooKeeper中的節點分為持久節點、持久順序節點、臨時節點、臨時順序節點，Dubbo使用ZooKeeper作為注冊中心時，不關心節點的創建順序，只會創建持久節點和臨時節點，

持久節點： 服務注冊后保證節點不會丟失，注冊中心重啟也會存在，
臨時節點： 服務注冊后連接丟失或session超時，注冊的節點會自動被移除，

1.2 ZooKeeper的Watcher機制

客戶端和服務器維持資料互動通常有兩種形式：

客戶端定時向服務器輪詢
服務器主動向客戶端推送資料

ZooKeeper采用的是方式2，主動向客戶端推送資料，客戶端注冊監聽它關心的節點（注冊Watcher監聽指定節點），當節點狀態發生變化（資料變化、子節點增減變化）時，則相應的Watcher被觸發，ZooKeeper 服務會通知客戶端，這就是Watcher機制，

Watcher有以下特點：

主動推送：Watcher被觸發時，由ZooKeeper主動將更新推送給客戶端，而不需要客戶端輪詢，
一次性：在節點上注冊Watcher監聽后，當節點狀態發生變化時該Watcher只會被觸發一次，如果客戶端想再收到后續發生變化的通知，需要重新再注冊一次Watcher，
可見性：如果一個客戶端在讀請求中附帶 Watcher，Watcher 被觸發的同時再次讀取資料，客戶端在得到 Watcher 訊息之前肯定不可能看到更新后的資料，換句話說，更新通知先于更新結果，
順序性：如果多個更新觸發了多個 Watcher ，那 Watcher 被觸發的順序與更新順序一致，

1.3 ZooKeeper會話機制

ZooKeeper 客戶端通過 TCP 長連接連接到 ZooKeeper 服務集群，會話 (Session) 從第一次連接開始就已經建立，之后通過心跳檢測機制來保持有效的會話狀態，通過這個連接，客戶端可以發送請求并接收回應，同時也可以接收到 Watch 事件的通知，一旦一臺客戶端與一臺服務器建立連接，這臺服務器會為這個客戶端創建一個新的會話，

每個會話都會有一個超時時間，若由于服務器壓力過大、網路故障等各種原因導致客戶端連接斷開時，只要在會話超時時間之內能夠重新連接上ZooKeeper服務器，那么之前創建的會話仍然有效，若服務器在超時時間內沒有收到任何請求，則相應會話被視為過期，一旦會話過期，就無法再重新打開，且任何與該會話相關的臨時節點都會被洗掉，

通常來說，會話應該長期存在，而這需要由客戶端來保證，客戶端可以通過心跳方式來保持會話不過期，

1.4 使用ZooKeeper作為注冊中心

如下圖所示，服務提供者（集成了ZK客戶端）在服務啟動時，會通過ZK客戶端與ZK服務端建立連接，將服務提供者資訊（提供者的IP地址、埠、服務介面資訊等）注冊到ZooKeeper服務端，這時會在ZooKeeper服務端生成一個臨時節點來存盤這些服務資訊，這就是服務提供者的注冊操作，

服務消費者（集成了ZK客戶端）在服務啟動時，會通過ZK客戶端與ZK服務端建立連接，拉取自己感興趣的服務提供者資訊并快取到本地，同時也會對自己感興趣的節點（比如服務提供節點）注冊Watcher監聽，后續發起遠程呼叫時，會從本地快取的服務提供者資訊通過負載均衡等策略選擇一臺服務提供者發起服務呼叫，

如果服務提供者擴容新增了機器，服務提供者會向ZK發起新的注冊操作，在對應的目錄下創建臨時節點（代表這臺新增的服務提供者資訊），同時會觸發之前服務消費者注冊的Watcher監聽，ZooKeeper服務端會將變更資訊推送到服務消費，服務消費者在接收到變更后會更新自己本地的服務提供者資訊，這樣就完成了服務的自動擴容，同樣的，當某臺服務提供者下線，它與ZooKeeper服務端的連接會斷掉，因為服務注冊時創建的是臨時節點，因此當連接斷掉后，該臨時節點會被洗掉，同時會觸發之前服務消費者注冊的Watcher監聽，相應的服務消費者會收到通知重繪自己本地的服務提供者資訊快取，

二、原始碼分析

Node，Dubbo中用Node來表示抽象節點，Dubbo中的核心概念Registry、Invoker等都實作了介面Node，

RegistryService定義了注冊服務需要提供的基本能力，即增刪改查，Registry介面繼承了Node和Registry介面，它表示擁有注冊中心能力的節點，其中定義了4個默認方法，其中的reExportRegister和reExportUnRegiser方法都是委托給RegisterService的相應方法，

AbstractRegistry實作了Registry介面中的方法，它在記憶體中實作了注冊資料的讀寫改動，實作了本地快取的功能，

FailbackRegistry繼承了AbstractRegistry，結合時間輪，提供了失敗重試的能力，

CacheableFailbackRegistry針對URL推送模型做了優化，減少了URL的創建，

ZooKeeperRegistry提供了基于ZooKeeper的注冊中心實作，

下面重點分析AbstractRegistry、FailbackRegistry、CacheableFailbackRegistry和ZooKeeperRegistry，

2.1 AbstractRegistry

AbstractRegistry實作了Registry介面中的方法，它在記憶體中實作了注冊資料的讀寫改動，從而可以就降低注冊中心的壓力，從前文繼承體系結構可以看出，Dubbo中的注冊中心實作都繼承了該類，

AbstractRegistry的核心是通過Properties型別的properties與File型別的file欄位來記錄服務資料，properties基于記憶體存盤了服務相關的資料，file則對應磁盤檔案，兩者的資料是同步的，在創建AbstractRegistry物件時，會根據傳入的URL中的file.cache引數來決定是否開啟本地快取，默認開啟，

如果開啟，會將磁盤檔案file中的資料加載到properties當中，當properties中的資料發生變化時，會同步到磁盤檔案file中，如果傳入的URL的save.file引數為false（默認是false），會通過執行緒池來異步同步properties的資料到file，如果是true則是在當前執行緒同步，

2.2 FailbackRegistry

FailbackRegistry處理的是注冊中心相關的邏輯處理例外時如何處理，它會使用時間輪來處理例外重試，像ZooKeeperRegistry、NacosRegistry等注冊中心實作，都繼承了FailbackRegistry，因此也都有了例外重試的能力，

FailbackRegistry繼承了AbstractRegistry，復寫了register/unregister、subscribe/unsubscribe、notify這5個核心方法，在這些方法中首先會呼叫父類AbstractRegistry對應的方法，然后真正的注冊訂閱等邏輯還是通過模板方法模式委托給了子類實作，重試邏輯則交由時間輪來處理，

2.2.1 核心欄位

FailbackRegistry的核心欄位如下：

1）ConcurrentMap < URL,FailedRegisteredTask > failedRegistered，注冊失敗的URL集合，key是注冊失敗的URL，value是FailedRegisteredTask，也就是重試注冊時要執行的邏輯，該注冊重試邏輯會交給時間輪執行，當注冊失敗時，會呼叫方法addFailedRegistered添加注冊失敗的URL，

2）ConcurrentMap < URL,FailedUnregisteredTask > failedUnregistered，取消注冊時發生失敗的URL集合，key是取消注冊失敗的URL，value是FailedUnregisteredTask，也就是重試取消注冊時要執行的邏輯，

3）ConcurrentMap < Holder,FailedSubscribedTask > failedSubscribed，訂閱失敗的URL集合，key是Holder（封裝了訂閱失敗的URL以及對應的監聽器NotifyListener）,value是FailedSubscribedTask，也就是重試訂閱時要執行的邏輯，

4）ConcurrentMap < Holder,FailedUnsubscribedTask >failedUnsubscribed，取消訂閱發生失敗的URL集合，key是Holder（封裝了取消訂閱失敗的URL以及對應的監聽器NotifyListener），value是FailedUnsubscribedTask，也就是重試取消訂閱時要執行的邏輯，

5）int retryPeriod，重試操作時間間隔，

6）HashedWheelTimer retryTimer，時間輪，用于執行重試操作，

2.2.2 核心方法

以subscribe為例來分析FailbackRegistry中是如何處理重試的（其余方法類似），如下圖所示，首先FailbackRegistry的subscribe方法會呼叫父類AbstractRegistry的subcribe方法，將訂閱資料添加到記憶體中進行維護，接著會從訂閱失敗/取消訂閱失敗的集合中移除該URL相關的訂閱資料，

然后呼叫子類的doSubscribe方法將真正的訂閱邏輯交給子類實作，這是典型的模板方法設計模式，如果發生例外，會呼叫getCacheUrls方法獲取快取的服務提供者資料，如果快取的服務資料非空，因為這邊訂閱失敗了，所以需要手動觸發下notify方法，回呼相關的NotifyListener方法，重繪消費者本地的服務提供者串列、路由、配置的資料，如果快取資料為空，則需要判斷下，是否檢測訂閱失敗以及訂閱失敗的例外是否可以跳過，根據這個來判斷是否需要拋出例外還是忽略僅列印日志，

最后會呼叫addFailedSubscribed方法將訂閱失敗的資訊添加到failedSubscribed集合，以及將任務添加到時間輪中，這樣當時間到了，時間輪就可以處理該重試任務了，

這邊有一點需要注意，如果任務已經存在failedSubscribed集合中，就不需要重復添加了，failedSubscribed是Map結構，通過key來判斷資料是否存在，因此這邊的Holder作為key時，需要復寫hashCode和equals方法，

@Override
public void subscribe(URL url, NotifyListener listener) {
    // 呼叫父類方法, 在記憶體中記錄下訂閱相關的資料
    super.subscribe(url, listener);
    // 從訂閱失敗集合、取消訂閱失敗集合中移除該URL相關的資料, 如果存在需要取消相關任務的執行
    removeFailedSubscribed(url, listener);
    try {
        // 模板方法設計模式, 委托給子類實作真正的訂閱邏輯
        doSubscribe(url, listener);
    } catch (Exception e) {
        Throwable t = e;
        // 訂閱發生了例外, 則使用本地快取的服務提供者資料
        List<URL> urls = getCacheUrls(url);
        if (CollectionUtils.isNotEmpty(urls)) {
            // 呼叫notify方法回呼相關的listener方法, 通知消費者重繪自己本地
            // 的服務提供者、路由、配置等資料
            notify(url, listener, urls);
        } else {
            // 根據check以及是否是可忽略的例外來判斷是否需要拋出例外
            boolean check = getUrl().getParameter(Constants.CHECK_KEY, true)
                && url.getParameter(Constants.CHECK_KEY, true);
            boolean skipFailback = t instanceof SkipFailbackWrapperException;
            if (check || skipFailback) {
                if (skipFailback) {
                    t = t.getCause();
                }
                // 拋例外
            } else {
                // 僅列印錯誤日志
            }
        }
 
        // 添加訂閱失敗的URL資訊
        addFailedSubscribed(url, listener);
    }
}

protected void addFailedSubscribed(URL url, NotifyListener listener) {
    // 封裝成Holder
    Holder h = new Holder(url, listener);
    // 判斷該Holder是否已經存在, 如果已經存在則直接回傳, 不需要多次添加
    FailedSubscribedTask oldOne = failedSubscribed.get(h);
    if (oldOne != null) {
        return;
    }
    // 創建訂閱失敗的重試任務
    FailedSubscribedTask newTask = new FailedSubscribedTask(url, this, listener);
    // 向訂閱失敗的集合里面添加該任務, 如果回傳null說明之前不存在該任務,
    // 這次需要往時間輪中注冊該任務
    oldOne = failedSubscribed.putIfAbsent(h, newTask);
    if (oldOne == null) {
        // 將該任務添加到時間輪中
        retryTimer.newTimeout(newTask, retryPeriod, TimeUnit.MILLISECONDS);
    }
}

2.3 CacheableFailbackRegistry

在2.7.8版本中并沒有CacheableFailbackRegistry這個類，在Dubbo3.0中，針對服務資料的推送做了一系列的優化，CacheableFailbackRegistry正是其中一個改動，下面來進行具體的講解，

2.3.1 URL推送模型

下圖所示是Dubbo2.7.8中的URL推送模型，消費者啟動后會向ZooKeeper服務端訂閱感興趣的服務，當有新的消費者感興趣的服務提供者節點（提供者3）加入服務時，該服務（提供者3）會把自己的服務資訊注冊到ZooKeeper服務端，接著ZooKeeper服務端會把providers節點下的所有服務實體資訊（提供者1、2、3）全量推送給消費者，消費者收到后根據推送的資料全量生成URL串列，該URL串列后續會被消費者用來構建自己本地的可選的服務提供者串列資料，后續發起遠程呼叫時，就可以從該服務提供者串列中選擇一個節點，

可以看到，當服務提供者實體數量較小時，推送的資料量較小，消費者端構建URL的壓力就小一些，但是當某個介面有大量的服務提供者時，當發生服務擴容/縮容，就會有大量的URL被創建，Dubbo3.0中的對該URL推送模型做了一系列的優化，主要是對URL的類結構做了一定調整，引入了多級快取，下面具體分析，

2.3.2 URL結構變更及多級快取

如下兩圖分別是Dubbo2.7.8和Dubbo3.0.7中的URL結構，可以看到在3.0.7中新增了URLAddress和URLParam類，原本URL中的host、port等資訊移到了URLAddress中，原本URL的引數資訊則移到了URLParam中，為什么要這么做？很大程度上是為了利用快取，避免重復創建URL物件，比如將URL中的host、port等資訊抽取到URLAddress物件中，當發生服務資料變更推送時，根據host、port等資訊很大程度上能夠從快取中找到已有的URLAddress物件，這樣就避免了一些不必要的URL創建，

2.3.3 核心方法

如下圖所示，當消費者接收到ZooKeeper注冊中心推送過來的全量服務提供者串列資訊（providers）時，會呼叫到該方法進行ServiceAddressURL串列的創建，首先，會根據消費者URL嘗試從快取stringUrls中獲取服務提供者資訊對應的Map（key是服務提供者資訊的字串表示，value是對應的ServiceAddressURL），如果該Map為空，則providers對應的ServiceAddressURL都需要創建，如果Map不為空，則先從該Map中根據provider的字串key嘗試獲取快取，如果存在則不需要呼叫createURL創建，

createURL也是依次從快取stringAddress和stringParam獲取對應的URLAddress和URLParam，如果快取中不存在則創建，接著，利用剛付訓取到的URLAddress和URLParam以及consumerURL創建ServiceAddressURL，

創建完成后會判斷該ServiceAddressURL是否和當前消費者匹配（version、group這些資訊是否匹配，前文已經分析過），如果不匹配則回傳空，后續也不會被更新到stringUrls快取中，

創建完成后會呼叫evictURLCache，將待清理的URL資訊放入waitForRemove中，cacheRemovalScheduler會定時清理快取資料，

protected List<URL> toUrlsWithoutEmpty(URL consumer, Collection<String> providers) {
    // 根據消費者URL資訊從stringUrls中獲取對應的服務提供者資訊
    Map<String, ServiceAddressURL> oldURLs = stringUrls.get(consumer);
    // create new urls
    Map<String, ServiceAddressURL> newURLs;
    // 去除consumerURL中的release、dubbo、methods、時間戳、dubbo.tag這些引數
    URL copyOfConsumer = removeParamsFromConsumer(consumer);
    // 如果快取中沒有, 肯定都是需要新建的
    if (oldURLs == null) {
        newURLs = new HashMap<>((int) (providers.size() / 0.75f + 1));
        for (String rawProvider : providers) {
            // 去除掉rawProvider中的時間戳、PID引數
            rawProvider = stripOffVariableKeys(rawProvider);
            // 呼叫createURL創建ServiceAddressURL
            ServiceAddressURL cachedURL =
                         createURL(rawProvider, copyOfConsumer, getExtraParameters());
            // 如果創建的為空則忽略不放入newURLs
            if (cachedURL == null) {
                continue;
            }
            newURLs.put(rawProvider, cachedURL);
        }
    } else {
        newURLs = new HashMap<>((int) (providers.size() / 0.75f + 1));
        // maybe only default , or "env" + default
        for (String rawProvider : providers) {
            rawProvider = stripOffVariableKeys(rawProvider);
            // 從快取中獲取, 如果快取中存在就不需要再創建了
            ServiceAddressURL cachedURL = oldURLs.remove(rawProvider);
            if (cachedURL == null) {
                cachedURL = createURL(rawProvider, copyOfConsumer, getExtraParameters());
                if (cachedURL == null) {
                    continue;
                }
            }
            newURLs.put(rawProvider, cachedURL);
        }
    }
 
    // 清除老的快取資料
    evictURLCache(consumer);
    // 更新stringUrls快取
    stringUrls.put(consumer, newURLs);
 
    return new ArrayList<>(newURLs.values());
}

protected ServiceAddressURL createURL(String rawProvider, URL consumerURL,
                                      Map<String, String> extraParameters) {
    boolean encoded = true;
    
    int paramStartIdx = rawProvider.indexOf(ENCODED_QUESTION_MARK);
     
    if (paramStartIdx == -1) {
        encoded = false;
    }
    // 決議rawProvider, 一分為二, 一個用來創建URLAddress、一個用來創建URLParam
    String[] parts = URLStrParser.parseRawURLToArrays(rawProvider, paramStartIdx);
    if (parts.length <= 1) {
        return DubboServiceAddressURL.valueOf(rawProvider, consumerURL);
    }
 
    String rawAddress = parts[0];
    String rawParams = parts[1];
    boolean isEncoded = encoded;
     
    // 從快取stringAddress快取中獲取URLAddress, 不存在則創建
    URLAddress address =
        stringAddress.computeIfAbsent(rawAddress,
                   k -> URLAddress.parse(k, getDefaultURLProtocol(), isEncoded));
    address.setTimestamp(System.currentTimeMillis());
 
    // 從快取stringParam快取中獲取URLParam, 不存在則創建
    URLParam param = stringParam.computeIfAbsent(rawParams,
                             k -> URLParam.parse(k, isEncoded, extraParameters));
    param.setTimestamp(System.currentTimeMillis());
 
    // 用獲取到的URLAddress、URLParam直接new創建ServiceAddressURL
    ServiceAddressURL cachedURL = createServiceURL(address, param, consumerURL);
     
    // 判斷創建出來的ServiceAddressURL是否和當前消費者匹配, 不匹配回傳null
    if (isMatch(consumerURL, cachedURL)) {
        return cachedURL;
    }
    return null;
}

2.4 ZooKeeperRegistry

2.4.1 注冊

根據傳入的URL生成該節點要在ZooKeeper服務端上注冊的節點路徑，值為如下形式：/dubbo/org.apache.dubbo.springboot.demo.DemoService/providers/服務資訊，/dubbo是根路徑，接下來是服務的介面名，然后/providers是型別資訊（如果是消費者則是/consumers節點，路由資訊則是/routers），最后是URL的字串表示，得到節點路徑后，會根據URL的dynamic引數（默認是true）決定在ZooKeeper服務端上創建的是臨時節點，還是持久節點，默認是臨時節點，

注冊后資料結構如下圖所示，

2.4.2 訂閱

訂閱的核心是通過ZooKeeperClient在指定的節點的添加ChildListener，當該節點的子節點資料發生變化時，ZooKeeper服務端會通知到該ChildListener的notify方法，然后呼叫到對應的NotifyListener方法，重繪消費者本地的服務提供者串列等資訊，

doSubscribe方法主要分為兩個分支：URL的interface引數明確指定了為*（訂閱所有，通常實際使用中不會這么使用，Dubbo的控制后臺會訂閱所有）或者就訂閱某個服務介面，接下來分析訂閱某個指定服務介面這個分支代碼，這塊涉及到三個監聽器（它們是一對一的）：

1）NotifyListener，Dubbo中定義的通用的訂閱相關的監聽器，它是定義在dubbo-registry-api模塊中的，不僅僅在ZooKeeper注冊中心模塊中使用，

2）ChildListener，Dubbo中定義的針對ZooKeeper注冊中心的監聽器，用來監聽指定節點子節點的資料變化，

3）CuratorWatcher，Curator框架中的監聽器，Curator是常用的ZooKeeper客戶端，如果Dubbo采用其它ZooKeeper客戶端工具，這塊就是其它相關的監聽器邏輯了，

當訂閱的服務資料發生變化時，最先會觸發到CuratorWatcher的process方法，process方法中會呼叫ChildListener的childChanged方法，在childChanged方法會繼續觸發呼叫到ZooKeeperRegistry的notify方法，這里有兩點需要注意：

1）因為doSubscribe方法中通過ZooKeeperClient添加Watcher監聽器時，使用的是usingWatcher，這是一次性的，所以在CuratorWatcher的實作CuratorWatcherImpl的process方法中，當收到ZooKeeper的變更資料推送時，會再次在path上注冊Watcher，

2）在ChildListener的實作RegistryChildListenerImpl的doNotify方法中，會呼叫ZooKeeperRegistry的toUrlsWithEmpty將傳入的字串形式的服務提供者串列等資料轉換成對應的ServiceAddressURL串列資料，以供后面使用，

明確了三個監聽器的含義之后，接下來分析doSubscribe的邏輯就簡單了，首先會呼叫toCategoriesPath方法獲取三個path路徑，分別是

/dubbo/org.apache.dubbo.demo.DemoService/providers
/dubbo/org.apache.dubbo.demo.DemoService/configurators
/dubbo/org.apache.dubbo.demo.DemoService/routers

表示當前消費者（url引數代表）需要訂閱這三個節點，當任意一個節點的資料發生變化時，ZooKeeper服務端都會通知到當前注冊中心客戶端，更新本地的服務資料，依次遍歷這三個path路徑，分別在這三個path上注冊監聽器，

ZooKeeperRegistry通過zkListeners這個Map維護了所有消費者（URL）的所有監聽器資料，首先根據url引數獲取當前消費者對應的監聽器listeners，這是一個Map，key是NotifyListener，value是對應的ChildListener，如果不存在則需要創建ChildListener的實作RegistryChildListenerImpl，當創建完成后，會在ZooKeeper服務端創建持久化的path路徑，并且在該path路徑上注冊監聽器，首次訂閱注冊監聽器時，會獲取到該路徑下的所有服務資料的字串形式，并呼叫toUrlsWithEmpty轉成URL形式，接著呼叫notify方法重繪消費者本地的服務相關資料，

@Override
public void doSubscribe(final URL url, final NotifyListener listener) {
    
    CountDownLatch latch = new CountDownLatch(1);
    try {
        List<URL> urls = new ArrayList<>();
        // 獲取三個路徑, 分別是/providers、/routers、/configurators
        for (String path : toCategoriesPath(url)) {
             
            // 獲取該消費者URL對應的監聽器Map
            ConcurrentMap<NotifyListener, ChildListener> listeners =
                      zkListeners.computeIfAbsent(url, k -> new ConcurrentHashMap<>());
             
            // 查找NotifyListener對應的ChildListener是否存在, 不存在則創建
            ChildListener zkListener = listeners.computeIfAbsent(listener,
                              k -> new RegistryChildListenerImpl(url, path, k, latch));
            if (zkListener instanceof RegistryChildListenerImpl) {
                ((RegistryChildListenerImpl) zkListener).setLatch(latch);
            }
             
            // 在ZooKeeper服務端創建持久節點
            zkClient.create(path, false);
            // 添加ChildListener監聽器, 并全量拉取下相關服務資料
            List<String> children = zkClient.addChildListener(path, zkListener);
            if (children != null) {
                // 將字串形式的服務資料轉換成URL串列資料
                urls.addAll(toUrlsWithEmpty(url, path, children));
            }
        }
        // 首次訂閱后觸發下notify邏輯, 重繪消費者本地的服務提供者串列等資料
        notify(url, listener, urls);
    } finally {
        latch.countDown();
    }
}

接下來看下AbstractZooKeeperClient的addChildListener方法，邏輯也比較簡單，在指定path路徑上添加一次性的Watcher，取消訂閱的邏輯則是會將傳入的UR和NotifyListener對應的ChildListener移除，不再監聽，

public List<String> addChildListener(String path, final ChildListener listener) {
    ConcurrentMap<ChildListener, TargetChildListener> listeners =
                childListeners.computeIfAbsent(path, k -> new ConcurrentHashMap<>());
     
    TargetChildListener targetListener = listeners.computeIfAbsent(listener,
                                            k -> createTargetChildListener(path, k));
    return addTargetChildListener(path, targetListener);
}
 
@Override
public List<String> addTargetChildListener(String path, CuratorWatcherImpl listener) {
    try {
        return client.getChildren().usingWatcher(listener).forPath(path);
    } catch (NoNodeException e) {
        return null;
    } catch (Exception e) {
        throw new IllegalStateException(e.getMessage(), e);
    }
}

三、總結

本文通過分析Dubbo3.0中AbstractRegistry、FailbackRegistry、ZooKeeperRegistry，詳細介紹了Dubbo中ZooKeeper注冊中心的實作原理，包括服務資料本地快取、服務注冊訂閱例外重試等，另外，網上大部分文章是基于2.x版本的，本文基于Dubbo3.0，重點分析了Dubbo3.0中新引入的CacheableFailbackRegistry，詳細介紹了Dubbo在URL推送模型上所做的優化，

參考資料：

Dubbo3.0 阿里大規模實踐決議-URL重構
深入淺出ZooKeeper

分享 vivo 互聯網技術干貨與沙龍活動，推薦最新行業動態與熱門會議，

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

標籤：Java

上一篇：讀Java8函式式編程筆記08_測驗、除錯和重構

下一篇：dapr入門與本地托管模式嘗試

Dubbo 中 Zookeeper 注冊中心原理分析