摘要：在一個(gè)服務(wù)集群中，服務(wù)提供者數(shù)量并不是一成不變的，如果集群中新增了一臺(tái)機(jī)器，相應(yīng)地在服務(wù)目錄中就要新增一條服務(wù)提供者記錄。
1. 簡介
前面文章分析了服務(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ù)目錄和注冊中心的功能不就雷同了嗎。確實(shí)如此，這里這么說是為了方便大家理解。實(shí)際上服務(wù)目錄在獲取注冊中心的服務(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ì)隨注冊中心的變化而進(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)注冊中心節(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 的源碼。
public List> list(Invocation invocation) throws RpcException {
if (destroyed) {
throw new RpcException("Directory already destroyed...");
}
// 調(diào)用 doList 方法列舉 Invoker，這里的 doList 是模板方法，由子類實(shí)現(xiàn)
List> invokers = doList(invocation);
// 獲取路由器
List 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> doList(Invocation invocation) throws RpcException;
上面就是 AbstractDirectory 的 list 方法源碼，這個(gè)方法封裝了 Invoker 的列舉過程。如下：
調(diào)用 doList 獲取 Invoker 列表
根據(jù) Router 的 getUrl 返回值為空與否，以及 runtime 參數(shù)決定是否進(jìn)行服務(wù)路由
以上步驟中，doList 是模板方法，需由子類實(shí)現(xiàn)。Router 的 runtime 參數(shù)這里簡單說明一下，這個(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)。
public class StaticDirectory extends AbstractDirectory {
// Invoker 列表
private final List> invokers;
// 省略構(gòu)造方法
@Override
public Class getInterface() {
// 獲取接口類
return invokers.get(0).getInterface();
}
// 檢測服務(wù)目錄是否可用
@Override
public boolean isAvailable() {
if (isDestroyed()) {
return false;
}
for (Invoker invoker : invokers) {
if (invoker.isAvailable()) {
// 只要有一個(gè) Invoker 是可用的，就任務(wù)當(dāng)前目錄是可用的
return true;
}
}
return false;
}
@Override
public void destroy() {
if (isDestroyed()) {
return;
}
// 調(diào)用父類銷毀邏輯
super.destroy();
// 遍歷 Invoker 列表，并執(zhí)行相應(yīng)的銷毀邏輯
for (Invoker invoker : invokers) {
invoker.destroy();
}
invokers.clear();
}
@Override
protected List> doList(Invocation invocation) throws RpcException {
// 列舉 Inovker，也就是直接返回 invokers 成員變量
return invokers;
}
}
以上就是 StaticDirectory 的代碼邏輯，很簡單，大家都能看懂，我就不多說了。下面來看看 RegistryDirectory，這個(gè)類的邏輯比較復(fù)雜。
3.2 RegistryDirectory
RegistryDirectory 是一種動(dòng)態(tài)服務(wù)目錄，它實(shí)現(xiàn)了 NotifyListener 接口。當(dāng)注冊中心服務(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)介紹過了。那這里就不過多啰嗦了，我們直入主題吧。
public List> doList(Invocation invocation) {
if (forbidden) {
// 服務(wù)提供者關(guān)閉或禁用了服務(wù)，此時(shí)拋出 No provider 異常
throw new RpcException(RpcException.FORBIDDEN_EXCEPTION,
"No provider available from registry ...");
}
List> invokers = null;
// 獲取 Invoker 本地緩存
Map>> localMethodInvokerMap = this.methodInvokerMap;
if (localMethodInvokerMap != null && localMethodInvokerMap.size() > 0) {
// 獲取方法名和參數(shù)列表
String methodName = RpcUtils.getMethodName(invocation);
Object[] args = RpcUtils.getArguments(invocation);
// 檢測參數(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 列表，具體的使用場景暫時(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>> iterator = localMethodInvokerMap.values().iterator();
if (iterator.hasNext()) {
// 通過迭代器獲取 Invoker 列表
invokers = iterator.next();
}
}
}
// 返回 Invoker 列表
return invokers == null ? new ArrayList>(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í)未完全搞清楚，我猜測是被路由規(guī)則（用戶可基于 Router 接口實(shí)現(xiàn)自定義路由器）處理后，可能會(huì)得到一個(gè) null。目前僅是猜測，未做驗(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ù)目錄，它需要接受注冊中心配置進(jìn)行動(dòng)態(tài)調(diào)整。因此 RegistryDirectory 實(shí)現(xiàn)了 NotifyListener 接口，通過這個(gè)接口獲取注冊中心變更通知。下面我們來看一下具體的邏輯。
public synchronized void notify(List urls) {
// 定義三個(gè)集合，分別用于存放服務(wù)提供者 url，路由 url，配置器 url
List invokerUrls = new ArrayList();
List routerUrls = new ArrayList();
List configuratorUrls = new ArrayList();
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)) {
// 添加路由器 url
routerUrls.add(url);
} else if (Constants.CONFIGURATORS_CATEGORY.equals(category)
|| Constants.OVERRIDE_PROTOCOL.equals(protocol)) {
// 添加配置器 url
configuratorUrls.add(url);
} else if (Constants.PROVIDERS_CATEGORY.equals(category)) {
// 添加服務(wù)提供者 url
invokerUrls.add(url);
} else {
// 忽略不支持的 category
logger.warn("Unsupported category ...");
}
}
if (configuratorUrls != null && !configuratorUrls.isEmpty()) {
// 將 url 轉(zhuǎn)成 Configurator
this.configurators = toConfigurators(configuratorUrls);
}
if (routerUrls != null && !routerUrls.isEmpty()) {
// 將 url 轉(zhuǎn)成 Router
List routers = toRouters(routerUrls);
if (routers != null) {
setRouters(routers);
}
}
List localConfigurators = this.configurators;
this.overrideDirectoryUrl = directoryUrl;
if (localConfigurators != null && !localConfigurators.isEmpty()) {
for (Configurator configurator : localConfigurators) {
// 配置 overrideDirectoryUrl
this.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 隨注冊中心變化而變化的關(guān)鍵所在。這一塊邏輯比較多，接下來一一進(jìn)行分析。
private void refreshInvoker(List 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 為 true
this.forbidden = true;
this.methodInvokerMap = null;
// 銷毀所有 Invoker
destroyAllInvokers();
} else {
this.forbidden = false;
Map> oldUrlInvokerMap = this.urlInvokerMap;
if (invokerUrls.isEmpty() && this.cachedInvokerUrls != null) {
// 添加緩存 url 到 invokerUrls 中
invokerUrls.addAll(this.cachedInvokerUrls);
} else {
this.cachedInvokerUrls = new HashSet();
// 緩存 invokerUrls
this.cachedInvokerUrls.addAll(invokerUrls);
}
if (invokerUrls.isEmpty()) {
return;
}
// 將 url 轉(zhuǎn)成 Invoker
Map> newUrlInvokerMap = toInvokers(invokerUrls);
// 將 newUrlInvokerMap 轉(zhuǎn)成方法名到 Invoker 列表的映射
Map>> newMethodInvokerMap = toMethodInvokers(newUrlInvokerMap);
// 轉(zhuǎn)換出錯(cuò)，直接打印異常，并返回
if (newUrlInvokerMap == null || newUrlInvokerMap.size() == 0) {
logger.error(new IllegalStateException("urls to invokers error ..."));
return;
}
// 合并多個(gè)組的 Invoker
this.methodInvokerMap = multiGroup ? toMergeMethodInvokerMap(newMethodInvokerMap) : newMethodInvokerMap;
// 保存為本地緩存
this.urlInvokerMap = newUrlInvokerMap;
try {
// 銷毀無用 Invoker
destroyUnusedInvokers(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，得到 的映射關(guān)系。然后進(jìn)一步進(jìn)行轉(zhuǎn)換，得到 。之后進(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)換過程。
private Map> toInvokers(List urls) {
Map> newUrlInvokerMap = new HashMap>();
if (urls == null || urls.isEmpty()) {
return newUrlInvokerMap;
}
Set keys = new HashSet();
// 獲取服務(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(",");
// 檢測服務(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)前 providerUrl
continue;
}
}
// 忽略 empty 協(xié)議
if (Constants.EMPTY_PROTOCOL.equals(providerUrl.getProtocol())) {
continue;
}
// 通過 SPI 檢測服務(wù)端協(xié)議是否被消費(fèi)端支持
if (!ExtensionLoader.getExtensionLoader(Protocol.class).hasExtension(providerUrl.getProtocol())) {
logger.error(new IllegalStateException("Unsupported protocol..."));
continue;
}
// 合并 url
URL url = mergeUrl(providerUrl);
String key = url.toFullString();
if (keys.contains(key)) {
// 忽略重復(fù) url
continue;
}
keys.add(key);
// 本地 Invoker 緩存列表
Map> localUrlInvokerMap = this.urlInvokerMap;
Invoker 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 獲取 Invoker
invoker = new InvokerDelegate(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)行檢測，若服務(wù)消費(fèi)端的配置不支持服務(wù)端的協(xié)議，或服務(wù)端 url 協(xié)議頭為 empty 時(shí)，toInvokers 均會(huì)忽略服務(wù)提供方 url。必要的檢測做完后，緊接著是合并 url，然后訪問緩存，嘗試獲取與 url 對(duì)應(yīng)的 invoker。如果緩存命中，直接將 Invoker 存入 newUrlInvokerMap 中即可。如果未命中，則需要新建 Invoker。Invoker 是通過 Protocol 的 refer 方法創(chuàng)建的，這個(gè)我在上一篇文章中已經(jīng)分析過了，這里就不贅述了。
toInvokers 方法返回的是 映射關(guān)系表，接下來還要對(duì)這個(gè)結(jié)果進(jìn)行進(jìn)一步處理，得到方法名到 Invoker 列表的映射關(guān)系。這個(gè)過程由 toMethodInvokers 方法完成，如下：
private Map>> toMethodInvokers(Map> invokersMap) {
// 方法名 -> Invoker 列表
Map>> newMethodInvokerMap = new HashMap>>();
List> invokersList = new ArrayList>();
if (invokersMap != null && invokersMap.size() > 0) {
for (Invoker 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> methodInvokers = newMethodInvokerMap.get(method);
if (methodInvokers == null) {
methodInvokers = new ArrayList>();
newMethodInvokerMap.put(method, methodInvokers);
}
// 存儲(chǔ) Invoker 到列表中
methodInvokers.add(invoker);
}
}
}
}
invokersList.add(invoker);
}
}
// 進(jìn)行服務(wù)級(jí)別路由，參考：https://github.com/apache/incubator-dubbo/pull/749
List> 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> 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(newMethodInvokerMap.keySet())) {
List> 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ù)的合并邏輯。
private Map>> toMergeMethodInvokerMap(Map>> methodMap) {
Map>> result = new HashMap>>();
// 遍歷入?yún)? for (Map.Entry>> entry : methodMap.entrySet()) {
String method = entry.getKey();
List> invokers = entry.getValue();
// group -> Invoker 列表
Map>> groupMap = new HashMap>>();
// 遍歷 Invoker 列表
for (Invoker invoker : invokers) {
// 獲取分組配置
String group = invoker.getUrl().getParameter(Constants.GROUP_KEY, "");
List> groupInvokers = groupMap.get(group);
if (groupInvokers == null) {
groupInvokers = new ArrayList>();
// 緩存 > 到 groupMap 中
groupMap.put(group, groupInvokers);
}
// 存儲(chǔ) invoker 到 groupInvokers
groupInvokers.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> groupInvokers = new ArrayList>();
for (List> groupList : groupMap.values()) {
// 通過集群類合并每個(gè)分組對(duì)應(yīng)的 Invoker 列表
groupInvokers.add(cluster.join(new StaticDirectory(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é)束。
接下來我們再來看一下 Invoker 列表刷新邏輯的最后一個(gè)動(dòng)作 -- 刪除無用 Invoker。如下：
private void destroyUnusedInvokers(Map> oldUrlInvokerMap, Map> newUrlInvokerMap) {
if (newUrlInvokerMap == null || newUrlInvokerMap.size() == 0) {
destroyAllInvokers();
return;
}
List deleted = null;
if (oldUrlInvokerMap != null) {
// 獲取新生成的 Invoker 列表
Collection> newInvokers = newUrlInvokerMap.values();
// 遍歷老的 映射表
for (Map.Entry> entry : oldUrlInvokerMap.entrySet()) {
// 檢測 newInvokers 中是否包含老的 Invoker
if (!newInvokers.contains(entry.getValue())) {
if (deleted == null) {
deleted = new ArrayList();
}
// 若不包含，則將老的 Invoker 對(duì)應(yīng)的 url 存入 deleted 列表中
deleted.add(entry.getKey());
}
}
}
if (deleted != null) {
// 遍歷 deleted 集合，并到老的 映射關(guān)系表查出 Invoker，銷毀之
for (String url : deleted) {
if (url != null) {
// 從 oldUrlInvokerMap 中移除 url 對(duì)應(yīng)的 Invoker
Invoker invoker = oldUrlInvokerMap.remove(url);
if (invoker != null) {
try {
// 銷毀 Invoker
invoker.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é)。如下：
檢測入?yún)⑹欠駜H包含一個(gè) url，且 url 協(xié)議頭為 empty
若第一步檢測結(jié)果為 true，表示禁用所有服務(wù)，此時(shí)銷毀所有的 Invoker
若第一步檢測結(jié)果為 false，此時(shí)將入?yún)⑥D(zhuǎn)為 Invoker 列表
對(duì)將上一步邏輯刪除的結(jié)果進(jìn)行進(jìn)一步處理，得到方法名到 Invoker 的映射關(guān)系表
合并多組 Invoker
銷毀無用 Invoker
Invoker 的刷新邏輯還是比較復(fù)雜的，大家在看的過程中多寫點(diǎn) demo 進(jìn)行調(diào)試。好了，本節(jié)就到這。
4. 總結(jié)
本篇文章對(duì) Dubbo 服務(wù)目錄進(jìn)行了較為詳細(xì)的分析，篇幅主要集中在 RegistryDirectory 的源碼分析上。分析下來，不由得感嘆，想讓本地服務(wù)目錄和注冊中心保持一致還是需要做很多事情的，并不簡單。服務(wù)目錄是 Dubbo 集群容錯(cuò)的一部分，也是比較基礎(chǔ)的部分，所以大家務(wù)必搞懂。
好了，本篇文章就先到這了。感謝大家閱讀。
本文在知識(shí)共享許可協(xié)議 4.0 下發(fā)布，轉(zhuǎn)載需在明顯位置處注明出處
作者：田小波
本文同步發(fā)布在我的個(gè)人博客：http://www.tianxiaobo.com
本作品采用知識(shí)共享署名-非商業(yè)性使用-禁止演繹 4.0 國際許可協(xié)議進(jìn)行許可。
