Dubbo 源码分析 - 集群容错之Directory
1. 簡(jiǎn)介
前面文章分析了服務(wù)的導(dǎo)出與引用過程，從本篇文章開始，我將開始分析 Dubbo 集群容錯(cuò)方面的源碼。這部分源碼包含四個(gè)部分，分別是服務(wù)目錄 Directory、服務(wù)路由 Router、集群 Cluster 和負(fù)載均衡 LoadBalance。這幾個(gè)部分的源碼邏輯比較獨(dú)立，我會(huì)分四篇文章進(jìn)行分析。本篇文章作為集群容錯(cuò)的開篇文章，將和大家一起分析服務(wù)目錄相關(guān)的源碼。在進(jìn)行深入分析之前，我們先來了解一下服務(wù)目錄是什么。服務(wù)目錄中存儲(chǔ)了一些和服務(wù)提供者有關(guān)的信息，通過服務(wù)目錄，服務(wù)消費(fèi)者可獲取到服務(wù)提供者的信息，比如 ip、端口、服務(wù)協(xié)議等。通過這些信息，服務(wù)消費(fèi)者就可通過 Netty 等客戶端進(jìn)行遠(yuǎn)程調(diào)用。在一個(gè)服務(wù)集群中，服務(wù)提供者數(shù)量并不是一成不變的，如果集群中新增了一臺(tái)機(jī)器，相應(yīng)地在服務(wù)目錄中就要新增一條服務(wù)提供者記錄。或者，如果服務(wù)提供者的配置修改了，服務(wù)目錄中的記錄也要做相應(yīng)的更新。如果這樣說，服務(wù)目錄和注冊(cè)中心的功能不就雷同了嗎。確實(shí)如此，這里這么說是為了方便大家理解。實(shí)際上服務(wù)目錄在獲取注冊(cè)中心的服務(wù)配置信息后，會(huì)為每條配置信息生成一個(gè) Invoker 對(duì)象，并把這個(gè) Invoker 對(duì)象存儲(chǔ)起來，這個(gè) Invoker 才是服務(wù)目錄最終持有的對(duì)象。Invoker 有什么用呢？看名字就知道了，這是一個(gè)具有遠(yuǎn)程調(diào)用功能的對(duì)象。講到這大家應(yīng)該知道了什么是服務(wù)目錄了，它可以看做是 Invoker 集合，且這個(gè)集合中的元素會(huì)隨注冊(cè)中心的變化而進(jìn)行動(dòng)態(tài)調(diào)整。
好了，關(guān)于服務(wù)目錄這里就先介紹這些，大家先有個(gè)大致印象即可。接下來我們通過繼承體系圖來了解一下服務(wù)目錄的家族成員都有哪些。
?2. 繼承體系
服務(wù)目錄目前內(nèi)置的實(shí)現(xiàn)有兩個(gè)，分別為 StaticDirectory 和 RegistryDirectory，它們均是 AbstractDirectory 的子類。AbstractDirectory 實(shí)現(xiàn)了 Directory 接口，這個(gè)接口包含了一個(gè)重要的方法定義，即 list(Invocation)，用于列舉 Invoker。下面我們來看一下他們的繼承體系圖。
如上，Directory 繼承自 Node 接口，Node 這個(gè)接口繼承者比較多，像 Registry、Monitor、Invoker 等繼承了這個(gè)接口。這個(gè)接口包含了一個(gè)獲取配置信息的方法 getUrl，實(shí)現(xiàn)該接口的類可以向外提供配置信息。另外，大家注意看 RegistryDirectory 實(shí)現(xiàn)了 NotifyListener 接口，當(dāng)注冊(cè)中心節(jié)點(diǎn)信息發(fā)生變化后，RegistryDirectory 可以通過此接口方法得到變更信息，并根據(jù)變更信息動(dòng)態(tài)調(diào)整內(nèi)部 Invoker 列表。
現(xiàn)在大家對(duì)服務(wù)目錄的繼承體系應(yīng)該比較清楚了，下面我們深入到源碼中，探索服務(wù)目錄是如何實(shí)現(xiàn)的。
?3. 源碼分析
本章我將分析 AbstractDirectory 和它兩個(gè)子類的源碼。這里之所以要分析 AbstractDirectory，而不是直接分析子類是有一定原因的。AbstractDirectory 封裝了 Invoker 列舉流程，具體的列舉邏輯則由子類實(shí)現(xiàn)，這是典型的模板模式。所以，接下來我們先來看一下 AbstractDirectory 的源碼。
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 | public List<Invoker<T>> list(Invocation invocation) throws RpcException {if (destroyed) {throw new RpcException("Directory already destroyed...");}// 調(diào)用 doList 方法列舉 Invoker，這里的 doList 是模板方法，由子類實(shí)現(xiàn)List<Invoker<T>> invokers = doList(invocation);// 獲取路由器List<Router> localRouters = this.routers;if (localRouters != null && !localRouters.isEmpty()) {for (Router router : localRouters) {try {// 獲取 runtime 參數(shù)，并根據(jù)參數(shù)決定是否進(jìn)行路由if (router.getUrl() == null || router.getUrl().getParameter(Constants.RUNTIME_KEY, false)) {// 進(jìn)行服務(wù)路由invokers = router.route(invokers, getConsumerUrl(), invocation);}} catch (Throwable t) {logger.error("Failed to execute router: ...");}}}return invokers; }// 模板方法，由子類實(shí)現(xiàn) protected abstract List<Invoker<T>> doList(Invocation invocation) throws RpcException; |
上面就是 AbstractDirectory 的 list 方法源碼，這個(gè)方法封裝了 Invoker 的列舉過程。如下：
以上步驟中，doList 是模板方法，需由子類實(shí)現(xiàn)。Router 的 runtime 參數(shù)這里簡(jiǎn)單說明一下，這個(gè)參數(shù)決定了是否在每次調(diào)用服務(wù)時(shí)都執(zhí)行路由規(guī)則。如果 runtime 為 true，那么每次調(diào)用服務(wù)前，都需要進(jìn)行服務(wù)路由。這個(gè)對(duì)性能造成影響，慎重配置。關(guān)于該參數(shù)更詳細(xì)的說明，請(qǐng)參考官方文檔。
介紹完 AbstractDirectory，接下來我們開始分析子類的源碼。
?3.1 StaticDirectory
StaticDirectory 即靜態(tài)服務(wù)目錄，顧名思義，它內(nèi)部存放的 Invoker 是不會(huì)變動(dòng)的。所以，理論上它和不可變 List 的功能很相似。下面我們來看一下這個(gè)類的實(shí)現(xiàn)。
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 | public class StaticDirectory<T> extends AbstractDirectory<T> {// Invoker 列表private final List<Invoker<T>> invokers;// 省略構(gòu)造方法@Overridepublic Class<T> getInterface() {// 獲取接口類return invokers.get(0).getInterface();}// 檢測(cè)服務(wù)目錄是否可用@Overridepublic boolean isAvailable() {if (isDestroyed()) {return false;}for (Invoker<T> invoker : invokers) {if (invoker.isAvailable()) {// 只要有一個(gè) Invoker 是可用的，就任務(wù)當(dāng)前目錄是可用的return true;}}return false;}@Overridepublic void destroy() {if (isDestroyed()) {return;}// 調(diào)用父類銷毀邏輯super.destroy();// 遍歷 Invoker 列表，并執(zhí)行相應(yīng)的銷毀邏輯for (Invoker<T> invoker : invokers) {invoker.destroy();}invokers.clear();}@Overrideprotected List<Invoker<T>> doList(Invocation invocation) throws RpcException {// 列舉 Inovker，也就是直接返回 invokers 成員變量return invokers;} } |
以上就是 StaticDirectory 的代碼邏輯，很簡(jiǎn)單，大家都能看懂，我就不多說了。下面來看看 RegistryDirectory，這個(gè)類的邏輯比較復(fù)雜。
?3.2 RegistryDirectory
RegistryDirectory 是一種動(dòng)態(tài)服務(wù)目錄，它實(shí)現(xiàn)了 NotifyListener 接口。當(dāng)注冊(cè)中心服務(wù)配置發(fā)生變化后，RegistryDirectory 可收到與當(dāng)前服務(wù)相關(guān)的變化。收到變更通知后，RegistryDirectory 可根據(jù)配置變更信息刷新 Invoker 列表。RegistryDirectory 中有幾個(gè)比較重要的邏輯，第一是 Invoker 的列舉邏輯，第二是接受服務(wù)配置變更的邏輯，第三是 Invoker 的刷新邏輯。接下來，我將按順序?qū)@三塊邏輯。
?3.2.1 列舉 Invoker
Invoker 列舉邏輯封裝在 doList 方法中，這是個(gè)模板方法，前面已經(jīng)介紹過了。那這里就不過多啰嗦了，我們直入主題吧。
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 | public List<Invoker<T>> doList(Invocation invocation) {if (forbidden) {// 服務(wù)提供者關(guān)閉或禁用了服務(wù)，此時(shí)拋出 No provider 異常throw new RpcException(RpcException.FORBIDDEN_EXCEPTION,"No provider available from registry ...");}List<Invoker<T>> invokers = null;// 獲取 Invoker 本地緩存Map<String, List<Invoker<T>>> localMethodInvokerMap = this.methodInvokerMap;if (localMethodInvokerMap != null && localMethodInvokerMap.size() > 0) {// 獲取方法名和參數(shù)列表String methodName = RpcUtils.getMethodName(invocation);Object[] args = RpcUtils.getArguments(invocation);// 檢測(cè)參數(shù)列表的第一個(gè)參數(shù)是否為 String 或 enum 類型if (args != null && args.length > 0 && args[0] != null&& (args[0] instanceof String || args[0].getClass().isEnum())) {// 通過 方法名 + 第一個(gè)參數(shù)名稱 查詢 Invoker 列表，具體的使用場(chǎng)景暫時(shí)沒想到invokers = localMethodInvokerMap.get(methodName + "." + args[0]);}if (invokers == null) {// 通過方法名獲取 Invoker 列表invokers = localMethodInvokerMap.get(methodName);}if (invokers == null) {// 通過星號(hào) * 獲取 Invoker 列表invokers = localMethodInvokerMap.get(Constants.ANY_VALUE);}if (invokers == null) {Iterator<List<Invoker<T>>> iterator = localMethodInvokerMap.values().iterator();if (iterator.hasNext()) {// 通過迭代器獲取 Invoker 列表invokers = iterator.next();}}}// 返回 Invoker 列表return invokers == null ? new ArrayList<Invoker<T>>(0) : invokers; } |
以上代碼進(jìn)行多次嘗試，以期從 localMethodInvokerMap 中獲取到 Invoker 列表。一般情況下，普通的調(diào)用可通過方法名獲取到對(duì)應(yīng)的 Invoker 列表，泛化調(diào)用可通過 ***** 獲取到 Invoker 列表。按現(xiàn)有的邏輯，不管什么情況下，***** 到 Invoker 列表的映射關(guān)系 <*****, invokers> 總是存在的，也就意味著 localMethodInvokerMap.get(Constants.ANY_VALUE) 總是有值返回。除非這個(gè)值是 null，才會(huì)通過通過迭代器獲取 Invoker 列表。至于什么情況下為空，我暫時(shí)未完全搞清楚，我猜測(cè)是被路由規(guī)則（用戶可基于 Router 接口實(shí)現(xiàn)自定義路由器）處理后，可能會(huì)得到一個(gè) null。目前僅是猜測(cè)，未做驗(yàn)證。
本節(jié)的邏輯主要是從 localMethodInvokerMap 中獲取 Invoker，localMethodInvokerMap 源自 RegistryDirectory 類的成員變量 methodInvokerMap。doList 方法可以看做是對(duì) methodInvokerMap 變量的讀操作，至于對(duì) methodInvokerMap 變量的寫操作，這個(gè)將在后續(xù)進(jìn)行分析。
?3.2.2 接收服務(wù)變更通知
RegistryDirectory 是一個(gè)動(dòng)態(tài)服務(wù)目錄，它需要接受注冊(cè)中心配置進(jìn)行動(dòng)態(tài)調(diào)整。因此 RegistryDirectory 實(shí)現(xiàn)了 NotifyListener 接口，通過這個(gè)接口獲取注冊(cè)中心變更通知。下面我們來看一下具體的邏輯。
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 | public synchronized void notify(List<URL> urls) {// 定義三個(gè)集合，分別用于存放服務(wù)提供者 url，路由 url，配置器 urlList<URL> invokerUrls = new ArrayList<URL>();List<URL> routerUrls = new ArrayList<URL>();List<URL> configuratorUrls = new ArrayList<URL>();for (URL url : urls) {String protocol = url.getProtocol();// 獲取 category 參數(shù)String category = url.getParameter(Constants.CATEGORY_KEY, Constants.DEFAULT_CATEGORY);// 根據(jù) category 參數(shù)將 url 分別放到不同的列表中if (Constants.ROUTERS_CATEGORY.equals(category)|| Constants.ROUTE_PROTOCOL.equals(protocol)) {// 添加路由器 urlrouterUrls.add(url);} else if (Constants.CONFIGURATORS_CATEGORY.equals(category)|| Constants.OVERRIDE_PROTOCOL.equals(protocol)) {// 添加配置器 urlconfiguratorUrls.add(url);} else if (Constants.PROVIDERS_CATEGORY.equals(category)) {// 添加服務(wù)提供者 urlinvokerUrls.add(url);} else {// 忽略不支持的 categorylogger.warn("Unsupported category ...");}}if (configuratorUrls != null && !configuratorUrls.isEmpty()) {// 將 url 轉(zhuǎn)成 Configuratorthis.configurators = toConfigurators(configuratorUrls);}if (routerUrls != null && !routerUrls.isEmpty()) {// 將 url 轉(zhuǎn)成 RouterList<Router> routers = toRouters(routerUrls);if (routers != null) {setRouters(routers);}}List<Configurator> localConfigurators = this.configurators;this.overrideDirectoryUrl = directoryUrl;if (localConfigurators != null && !localConfigurators.isEmpty()) {for (Configurator configurator : localConfigurators) {// 配置 overrideDirectoryUrlthis.overrideDirectoryUrl = configurator.configure(overrideDirectoryUrl);}}// 刷新 Invoker 列表refreshInvoker(invokerUrls); } |
如上，notify 方法首先是根據(jù) url 的 category 參數(shù)對(duì) url 進(jìn)行分門別類存儲(chǔ)，然后通過 toRouters 和 toConfigurators 將 url 列表轉(zhuǎn)成 Router 和 Configurator 列表。最后調(diào)用 refreshInvoker 方法刷新 Invoker 列表。這里的 toRouters 和 toConfigurators 方法邏輯不復(fù)雜，大家自行分析。接下來，我們把重點(diǎn)放在 refreshInvoker 方法上。
?3.2.3 刷新 Invoker 列表
接著上一節(jié)繼續(xù)分析，refreshInvoker 方法是保證 RegistryDirectory 隨注冊(cè)中心變化而變化的關(guān)鍵所在。這一塊邏輯比較多，接下來一一進(jìn)行分析。
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 | private void refreshInvoker(List<URL> invokerUrls) {// invokerUrls 僅有一個(gè)元素，且 url 協(xié)議頭為 empty，此時(shí)表示禁用所有服務(wù)if (invokerUrls != null && invokerUrls.size() == 1 && invokerUrls.get(0) != null&& Constants.EMPTY_PROTOCOL.equals(invokerUrls.get(0).getProtocol())) {// 設(shè)置 forbidden 為 truethis.forbidden = true;this.methodInvokerMap = null;// 銷毀所有 InvokerdestroyAllInvokers();} else {this.forbidden = false;Map<String, Invoker<T>> oldUrlInvokerMap = this.urlInvokerMap;if (invokerUrls.isEmpty() && this.cachedInvokerUrls != null) {// 添加緩存 url 到 invokerUrls 中invokerUrls.addAll(this.cachedInvokerUrls);} else {this.cachedInvokerUrls = new HashSet<URL>();// 緩存 invokerUrlsthis.cachedInvokerUrls.addAll(invokerUrls);}if (invokerUrls.isEmpty()) {return;}// 將 url 轉(zhuǎn)成 InvokerMap<String, Invoker<T>> newUrlInvokerMap = toInvokers(invokerUrls);// 將 newUrlInvokerMap 轉(zhuǎn)成方法名到 Invoker 列表的映射Map<String, List<Invoker<T>>> newMethodInvokerMap = toMethodInvokers(newUrlInvokerMap);// 轉(zhuǎn)換出錯(cuò)，直接打印異常，并返回if (newUrlInvokerMap == null || newUrlInvokerMap.size() == 0) {logger.error(new IllegalStateException("urls to invokers error ..."));return;}// 合并多個(gè)組的 Invokerthis.methodInvokerMap = multiGroup ? toMergeMethodInvokerMap(newMethodInvokerMap) : newMethodInvokerMap;// 保存為本地緩存this.urlInvokerMap = newUrlInvokerMap;try {// 銷毀無用 InvokerdestroyUnusedInvokers(oldUrlInvokerMap, newUrlInvokerMap);} catch (Exception e) {logger.warn("destroyUnusedInvokers error. ", e);}} } |
上面方法的代碼不是很多，但是邏輯卻不少。首先時(shí)根據(jù)入?yún)?invokerUrls 的數(shù)量和協(xié)議頭判斷是否禁用所有的服務(wù)，如果禁用，則將 forbidden 設(shè)為 true，并銷毀所有的 Invoker。若不禁用，則將 url 轉(zhuǎn)成 Invoker，得到 <url, Invoker> 的映射關(guān)系。然后進(jìn)一步進(jìn)行轉(zhuǎn)換，得到 <methodName, Invoker 列表>。之后進(jìn)行多組 Invoker 合并操作，并將合并結(jié)果賦值給 methodInvokerMap。methodInvokerMap 變量在 doList 方法中會(huì)被用到，doList 會(huì)對(duì)該變量進(jìn)行讀操作，在這里是寫操作。當(dāng)新的 Invoker 列表生成后，還要一個(gè)重要的工作要做，就是銷毀無用的 Invoker，避免服務(wù)消費(fèi)者調(diào)用已下線的服務(wù)的服務(wù)。
接下里，我將對(duì)上面涉及到的調(diào)用進(jìn)行分析。按照順序，這里先來分析 url 到 Invoker 的轉(zhuǎn)換過程。
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 | private Map<String, Invoker<T>> toInvokers(List<URL> urls) {Map<String, Invoker<T>> newUrlInvokerMap = new HashMap<String, Invoker<T>>();if (urls == null || urls.isEmpty()) {return newUrlInvokerMap;}Set<String> keys = new HashSet<String>();// 獲取服務(wù)消費(fèi)端配置的協(xié)議String queryProtocols = this.queryMap.get(Constants.PROTOCOL_KEY);for (URL providerUrl : urls) {if (queryProtocols != null && queryProtocols.length() > 0) {boolean accept = false;String[] acceptProtocols = queryProtocols.split(",");// 檢測(cè)服務(wù)提供者協(xié)議是否被服務(wù)消費(fèi)者所支持for (String acceptProtocol : acceptProtocols) {if (providerUrl.getProtocol().equals(acceptProtocol)) {accept = true;break;}}if (!accept) {// 若服務(wù)消費(fèi)者協(xié)議頭不被消費(fèi)者所支持，則忽略當(dāng)前 providerUrlcontinue;}}// 忽略 empty 協(xié)議if (Constants.EMPTY_PROTOCOL.equals(providerUrl.getProtocol())) {continue;}// 通過 SPI 檢測(cè)服務(wù)端協(xié)議是否被消費(fèi)端支持if (!ExtensionLoader.getExtensionLoader(Protocol.class).hasExtension(providerUrl.getProtocol())) {logger.error(new IllegalStateException("Unsupported protocol..."));continue;}// 合并 urlURL url = mergeUrl(providerUrl);String key = url.toFullString();if (keys.contains(key)) {// 忽略重復(fù) urlcontinue;}keys.add(key);// 本地 Invoker 緩存列表Map<String, Invoker<T>> localUrlInvokerMap = this.urlInvokerMap;Invoker<T> invoker = localUrlInvokerMap == null ? null : localUrlInvokerMap.get(key);// 緩存未命中if (invoker == null) {try {boolean enabled = true;if (url.hasParameter(Constants.DISABLED_KEY)) {// 獲取 disable 配置，并修改 enable 變量enabled = !url.getParameter(Constants.DISABLED_KEY, false);} else {enabled = url.getParameter(Constants.ENABLED_KEY, true);}if (enabled) {// 調(diào)用 refer 獲取 Invokerinvoker = new InvokerDelegate<T>(protocol.refer(serviceType, url), url, providerUrl);}} catch (Throwable t) {logger.error("Failed to refer invoker for interface...");}if (invoker != null) {// 緩存 Invoker 實(shí)例newUrlInvokerMap.put(key, invoker);}} else {// 緩存命中，將 invoker 存儲(chǔ)到 newUrlInvokerMap 中newUrlInvokerMap.put(key, invoker);}}keys.clear();return newUrlInvokerMap; } |
toInvokers 方法一開始會(huì)對(duì)服務(wù)提供者 url 進(jìn)行檢測(cè)，若服務(wù)消費(fèi)端的配置不支持服務(wù)端的協(xié)議，或服務(wù)端 url 協(xié)議頭為 empty 時(shí)，toInvokers 均會(huì)忽略服務(wù)提供方 url。必要的檢測(cè)做完后，緊接著是合并 url，然后訪問緩存，嘗試獲取與 url 對(duì)應(yīng)的 invoker。如果緩存命中，直接將 Invoker 存入 newUrlInvokerMap 中即可。如果未命中，則需要新建 Invoker。Invoker 是通過 Protocol 的 refer 方法創(chuàng)建的，這個(gè)我在上一篇文章中已經(jīng)分析過了，這里就不贅述了。
toInvokers 方法返回的是 <url, Invoker> 映射關(guān)系表，接下來還要對(duì)這個(gè)結(jié)果進(jìn)行進(jìn)一步處理，得到方法名到 Invoker 列表的映射關(guān)系。這個(gè)過程由 toMethodInvokers 方法完成，如下：
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 | private Map<String, List<Invoker<T>>> toMethodInvokers(Map<String, Invoker<T>> invokersMap) {// 方法名 -> Invoker 列表Map<String, List<Invoker<T>>> newMethodInvokerMap = new HashMap<String, List<Invoker<T>>>();List<Invoker<T>> invokersList = new ArrayList<Invoker<T>>();if (invokersMap != null && invokersMap.size() > 0) {for (Invoker<T> invoker : invokersMap.values()) {// 獲取 methods 參數(shù)String parameter = invoker.getUrl().getParameter(Constants.METHODS_KEY);if (parameter != null && parameter.length() > 0) {// 切分 methods 參數(shù)值，得到方法名數(shù)組String[] methods = Constants.COMMA_SPLIT_PATTERN.split(parameter);if (methods != null && methods.length > 0) {for (String method : methods) {// 方法名不為 *if (method != null && method.length() > 0&& !Constants.ANY_VALUE.equals(method)) {// 根據(jù)方法名獲取 Invoker 列表List<Invoker<T>> methodInvokers = newMethodInvokerMap.get(method);if (methodInvokers == null) {methodInvokers = new ArrayList<Invoker<T>>();newMethodInvokerMap.put(method, methodInvokers);}// 存儲(chǔ) Invoker 到列表中methodInvokers.add(invoker);}}}}invokersList.add(invoker);}}// 進(jìn)行服務(wù)級(jí)別路由，參考：https://github.com/apache/incubator-dubbo/pull/749List<Invoker<T>> newInvokersList = route(invokersList, null);// 存儲(chǔ) <*, newInvokersList> 映射關(guān)系newMethodInvokerMap.put(Constants.ANY_VALUE, newInvokersList);if (serviceMethods != null && serviceMethods.length > 0) {for (String method : serviceMethods) {List<Invoker<T>> methodInvokers = newMethodInvokerMap.get(method);if (methodInvokers == null || methodInvokers.isEmpty()) {methodInvokers = newInvokersList;}// 進(jìn)行方法級(jí)別路由newMethodInvokerMap.put(method, route(methodInvokers, method));}}// 排序，轉(zhuǎn)成不可變列表for (String method : new HashSet<String>(newMethodInvokerMap.keySet())) {List<Invoker<T>> methodInvokers = newMethodInvokerMap.get(method);Collections.sort(methodInvokers, InvokerComparator.getComparator());newMethodInvokerMap.put(method, Collections.unmodifiableList(methodInvokers));}return Collections.unmodifiableMap(newMethodInvokerMap); } |
上面方法主要做了三件事情， 第一是對(duì)入?yún)⑦M(jìn)行遍歷，然后獲取 methods 參數(shù)，并切分成數(shù)組。隨后以方法名為鍵，Invoker 列表為值，將映射關(guān)系存儲(chǔ)到 newMethodInvokerMap 中。第二是分別基于類和方法對(duì) Invoker 列表進(jìn)行路由操作。第三是對(duì) Invoker 列表進(jìn)行排序，并轉(zhuǎn)成不可變列表。關(guān)于 toMethodInvokers 方法就先分析到這，我們繼續(xù)向下分析，這次要分析的多組服務(wù)的合并邏輯。
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 | private Map<String, List<Invoker<T>>> toMergeMethodInvokerMap(Map<String, List<Invoker<T>>> methodMap) {Map<String, List<Invoker<T>>> result = new HashMap<String, List<Invoker<T>>>();// 遍歷入?yún)or (Map.Entry<String, List<Invoker<T>>> entry : methodMap.entrySet()) {String method = entry.getKey();List<Invoker<T>> invokers = entry.getValue();// group -> Invoker 列表Map<String, List<Invoker<T>>> groupMap = new HashMap<String, List<Invoker<T>>>();// 遍歷 Invoker 列表for (Invoker<T> invoker : invokers) {// 獲取分組配置String group = invoker.getUrl().getParameter(Constants.GROUP_KEY, "");List<Invoker<T>> groupInvokers = groupMap.get(group);if (groupInvokers == null) {groupInvokers = new ArrayList<Invoker<T>>();// 緩存 <group, List<Invoker>> 到 groupMap 中g(shù)roupMap.put(group, groupInvokers);}// 存儲(chǔ) invoker 到 groupInvokersgroupInvokers.add(invoker);}if (groupMap.size() == 1) {// 如果 groupMap 中僅包含一組鍵值對(duì)，此時(shí)直接取出該鍵值對(duì)的值即可result.put(method, groupMap.values().iterator().next());// groupMap 中包含多組鍵值對(duì)，比如：// {// "dubbo": [invoker1, invoker2, invoker3, ...],// "hello": [invoker4, invoker5, invoker6, ...]// }} else if (groupMap.size() > 1) {List<Invoker<T>> groupInvokers = new ArrayList<Invoker<T>>();for (List<Invoker<T>> groupList : groupMap.values()) {// 通過集群類合并每個(gè)分組對(duì)應(yīng)的 Invoker 列表groupInvokers.add(cluster.join(new StaticDirectory<T>(groupList)));}// 緩存結(jié)果result.put(method, groupInvokers);} else {result.put(method, invokers);}}return result; } |
上面方法首先是生成 group 到 Invoker 類比的映射關(guān)系表，若關(guān)系表中的映射關(guān)系數(shù)量大于1，表示有多組服務(wù)。此時(shí)通過集群類合并每組 Invoker，并將合并結(jié)果存儲(chǔ)到 groupInvokers 中。之后將方法名與 groupInvokers 存到到 result 中，并返回，整個(gè)邏輯結(jié)束。
接下來我們?cè)賮砜匆幌?Invoker 列表刷新邏輯的最后一個(gè)動(dòng)作 – 刪除無用 Invoker。如下：
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 | private void destroyUnusedInvokers(Map<String, Invoker<T>> oldUrlInvokerMap, Map<String, Invoker<T>> newUrlInvokerMap) {if (newUrlInvokerMap == null || newUrlInvokerMap.size() == 0) {destroyAllInvokers();return;}List<String> deleted = null;if (oldUrlInvokerMap != null) {// 獲取新生成的 Invoker 列表Collection<Invoker<T>> newInvokers = newUrlInvokerMap.values();// 遍歷老的 <url, Invoker> 映射表for (Map.Entry<String, Invoker<T>> entry : oldUrlInvokerMap.entrySet()) {// 檢測(cè) newInvokers 中是否包含老的 Invokerif (!newInvokers.contains(entry.getValue())) {if (deleted == null) {deleted = new ArrayList<String>();}// 若不包含，則將老的 Invoker 對(duì)應(yīng)的 url 存入 deleted 列表中deleted.add(entry.getKey());}}}if (deleted != null) {// 遍歷 deleted 集合，并到老的 <url, Invoker> 映射關(guān)系表查出 Invoker，銷毀之for (String url : deleted) {if (url != null) {// 從 oldUrlInvokerMap 中移除 url 對(duì)應(yīng)的 InvokerInvoker<T> invoker = oldUrlInvokerMap.remove(url);if (invoker != null) {try {// 銷毀 Invokerinvoker.destroy();} catch (Exception e) {logger.warn("destroy invoker...");}}}}} } |
destroyUnusedInvokers 方法的主要邏輯是通過 newUrlInvokerMap 找出待刪除 Invoker 對(duì)應(yīng)的 url，并將 url 存入到 deleted 列表中。然后再遍歷 deleted 列表，并從 oldUrlInvokerMap 中移除相應(yīng)的 Invoker，銷毀之。整個(gè)邏輯大致如此，不是很難理解。
到此關(guān)于 Invoker 列表的刷新邏輯就分析了，這里對(duì)整個(gè)過程進(jìn)行簡(jiǎn)單總結(jié)。如下：
Invoker 的刷新邏輯還是比較復(fù)雜的，大家在看的過程中多寫點(diǎn) demo 進(jìn)行調(diào)試。好了，本節(jié)就到這。
?4. 總結(jié)
本篇文章對(duì) Dubbo 服務(wù)目錄進(jìn)行了較為詳細(xì)的分析，篇幅主要集中在 RegistryDirectory 的源碼分析上。分析下來，不由得感嘆，想讓本地服務(wù)目錄和注冊(cè)中心保持一致還是需要做很多事情的，并不簡(jiǎn)單。服務(wù)目錄是 Dubbo 集群容錯(cuò)的一部分，也是比較基礎(chǔ)的部分，所以大家務(wù)必搞懂。
好了，本篇文章就先到這了。感謝大家閱讀。
- 本文鏈接：?https://www.tianxiaobo.com/2018/11/17/Dubbo-源碼分析-集群容錯(cuò)之Directory/
http://www.tianxiaobo.com/2018/11/17/Dubbo-%E6%BA%90%E7%A0%81%E5%88%86%E6%9E%90-%E9%9B%86%E7%BE%A4%E5%AE%B9%E9%94%99%E4%B9%8BDirectory/?
總結(jié)
以上是生活随笔為你收集整理的Dubbo 源码分析 - 集群容错之Directory的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： Dubbo 源码分析 - 服务引用
- 下一篇： Dubbo 源码分析 - 集群容错之 R