Dubbo 中 Zookeeper 註冊中心原理分析
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推播模型上所做的優化。
參考資料：