dubbo原始碼解析（三十七）叢集——directory

Dubbo · 發表 2019-02-06 09:04:47

摘要：叢集——directory 目標：介紹dubbo中叢集的目錄，介紹dubbo-cluster下directory包的原始碼。前言我在前面的文章中也提到了Directory可以看成是多個Invoker的集合，Directory 的用途是儲存 Invoker，其實現類 Regist...

叢集——directory

目標：介紹dubbo中叢集的目錄，介紹dubbo-cluster下directory包的原始碼。

前言

我在前面的文章中也提到了Directory可以看成是多個Invoker的集合，Directory 的用途是儲存 Invoker，其實現類 RegistryDirectory 是一個動態服務目錄，可感知註冊中心配置的變化，它所持有的 Inovker 列表會隨著註冊中心內容的變化而變化。每次變化後，RegistryDirectory 會動態增刪 Inovker，那在之前文章中我忽略了RegistryDirectory的原始碼分析，在本文中來補充。

原始碼分析

（一）AbstractDirectory

該類實現了Directory介面，

1.屬性

// logger
private static final Logger logger = LoggerFactory.getLogger(AbstractDirectory.class);

/**
 * url物件
 */
private final URL url;

/**
 * 是否銷燬
 */
private volatile boolean destroyed = false;

/**
 * 消費者端url
 */
private volatile URL consumerUrl;

/**
 * 路由集合
 */
private volatile List<Router> routers;

2.list

@Override
public List<Invoker<T>> list(Invocation invocation) throws RpcException {
// 如果銷燬，則丟擲異常
if (destroyed) {
throw new RpcException("Directory already destroyed .url: " + getUrl());
}
// 呼叫doList來獲得Invoker集合
List<Invoker<T>> invokers = doList(invocation);
// 獲得路由集合
List<Router> localRouters = this.routers; // local reference
if (localRouters != null && !localRouters.isEmpty()) {
// 遍歷路由
for (Router router : localRouters) {
try {
if (router.getUrl() == null || router.getUrl().getParameter(Constants.RUNTIME_KEY, false)) {
// 根據路由規則選擇符合規則的invoker集合
invokers = router.route(invokers, getConsumerUrl(), invocation);
}
} catch (Throwable t) {
logger.error("Failed to execute router: " + getUrl() + ", cause: " + t.getMessage(), t);
}
}
}
return invokers;
}

該方法是生成invoker集合的邏輯實現。其中doList是抽象方法，交由子類來實現。

3.setRouters

protected void setRouters(List<Router> routers) {
// copy list
// 複製路由集合
routers = routers == null ? new ArrayList<Router>() : new ArrayList<Router>(routers);
// append url router
// 獲得路由的配置
String routerkey = url.getParameter(Constants.ROUTER_KEY);
if (routerkey != null && routerkey.length() > 0) {
// 載入路由工廠
RouterFactory routerFactory = ExtensionLoader.getExtensionLoader(RouterFactory.class).getExtension(routerkey);
// 加入集合
routers.add(routerFactory.getRouter(url));
}
// append mock invoker selector
// 加入服務降級路由
routers.add(new MockInvokersSelector());
// 排序
Collections.sort(routers);
this.routers = routers;
}

（二）StaticDirectory

靜態 Directory 實現類，將傳入的 invokers 集合，封裝成靜態的 Directory 物件。

public class StaticDirectory<T> extends AbstractDirectory<T> {

private final List<Invoker<T>> invokers;

public StaticDirectory(List<Invoker<T>> invokers) {
this(null, invokers, null);
}

public StaticDirectory(List<Invoker<T>> invokers, List<Router> routers) {
this(null, invokers, routers);
}

public StaticDirectory(URL url, List<Invoker<T>> invokers) {
this(url, invokers, null);
}

public StaticDirectory(URL url, List<Invoker<T>> invokers, List<Router> routers) {
super(url == null && invokers != null && !invokers.isEmpty() ? invokers.get(0).getUrl() : url, routers);
if (invokers == null || invokers.isEmpty())
throw new IllegalArgumentException("invokers == null");
this.invokers = invokers;
}

@Override
public Class<T> getInterface() {
return invokers.get(0).getInterface();
}

@Override
public boolean isAvailable() {
if (isDestroyed()) {
return false;
}
// 遍歷invokers，如果有一個可用，則可用
for (Invoker<T> invoker : invokers) {
if (invoker.isAvailable()) {
return true;
}
}
return false;
}

@Override
public void destroy() {
if (isDestroyed()) {
return;
}
super.destroy();
// 遍歷invokers，銷燬所有的invoker
for (Invoker<T> invoker : invokers) {
invoker.destroy();
}
// 清除集合
invokers.clear();
}

@Override
protected List<Invoker<T>> doList(Invocation invocation) throws RpcException {

return invokers;
}

}

該類我就不多講解，比較簡單。

（三）RegistryDirectory

該類繼承了AbstractDirectory類，是基於註冊中心的動態 Directory 實現類，會根據註冊中心的推送變更 List<Invoker>

1.屬性

private static final Logger logger = LoggerFactory.getLogger(RegistryDirectory.class);

/**
 * cluster實現類物件
 */
private static final Cluster cluster = ExtensionLoader.getExtensionLoader(Cluster.class).getAdaptiveExtension();

/**
 * 路由工廠
 */
private static final RouterFactory routerFactory = ExtensionLoader.getExtensionLoader(RouterFactory.class).getAdaptiveExtension();

/**
 * 配置規則工廠
 */
private static final ConfiguratorFactory configuratorFactory = ExtensionLoader.getExtensionLoader(ConfiguratorFactory.class).getAdaptiveExtension();

/**
 * 服務key
 */
private final String serviceKey; // Initialization at construction time, assertion not null
/**
 * 服務型別
 */
private final Class<T> serviceType; // Initialization at construction time, assertion not null
/**
 * 消費者URL的配置項 Map
 */
private final Map<String, String> queryMap; // Initialization at construction time, assertion not null
/**
 * 原始的目錄 URL
 */
private final URL directoryUrl; // Initialization at construction time, assertion not null, and always assign non null value
/**
 * 服務方法集合
 */
private final String[] serviceMethods;
/**
 * 是否使用多分組
 */
private final boolean multiGroup;
/**
 * 協議
 */
private Protocol protocol; // Initialization at the time of injection, the assertion is not null
/**
 * 註冊中心
 */
private Registry registry; // Initialization at the time of injection, the assertion is not null
/**
 *是否禁止訪問
 */
private volatile boolean forbidden = false;

/**
 * 覆蓋目錄的url
 */
private volatile URL overrideDirectoryUrl; // Initialization at construction time, assertion not null, and always assign non null value

/**
 * override rules
 * Priority: override>-D>consumer>provider
 * Rule one: for a certain provider <ip:port,timeout=100>
 * Rule two: for all providers <* ,timeout=5000>
 * 配置規則陣列
 */
private volatile List<Configurator> configurators; // The initial value is null and the midway may be assigned to null, please use the local variable reference

// Map<url, Invoker> cache service url to invoker mapping.
/**
 * url與服務提供者 Invoker 集合的對映快取
 */
private volatile Map<String, Invoker<T>> urlInvokerMap; // The initial value is null and the midway may be assigned to null, please use the local variable reference

// Map<methodName, Invoker> cache service method to invokers mapping.
/**
 * 方法名和服務提供者 Invoker 集合的對映快取
 */
private volatile Map<String, List<Invoker<T>>> methodInvokerMap; // The initial value is null and the midway may be assigned to null, please use the local variable reference

// Set<invokerUrls> cache invokeUrls to invokers mapping.
/**
 * 服務提供者Invoker 集合快取
 */
private volatile Set<URL> cachedInvokerUrls; // The initial value is null and the midway may be assigned to null, please use the local variable reference

2.toConfigurators

public static List<Configurator> toConfigurators(List<URL> urls) {
// 如果為空，則返回空集合
if (urls == null || urls.isEmpty()) {
return Collections.emptyList();
}

List<Configurator> configurators = new ArrayList<Configurator>(urls.size());
// 遍歷url集合
for (URL url : urls) {
//如果是協議是empty的值，則清空配置集合
if (Constants.EMPTY_PROTOCOL.equals(url.getProtocol())) {
configurators.clear();
break;
}
// 覆蓋的引數集合
Map<String, String> override = new HashMap<String, String>(url.getParameters());
//The anyhost parameter of override may be added automatically, it can't change the judgement of changing url
// 覆蓋的anyhost引數可以自動新增，也不能改變更改url的判斷
override.remove(Constants.ANYHOST_KEY);
// 如果需要覆蓋新增的值為0，則清空配置
if (override.size() == 0) {
configurators.clear();
continue;
}
// 加入配置規則集合
configurators.add(configuratorFactory.getConfigurator(url));
}
// 排序
Collections.sort(configurators);
return configurators;
}

該方法是處理配置規則url集合，轉換覆蓋url對映以便在重新引用時使用，每次傳送所有規則，網址將被重新組裝和計算。

3.destroy

@Override
public void destroy() {
// 如果銷燬了，則返回
if (isDestroyed()) {
return;
}
// unsubscribe.
try {
if (getConsumerUrl() != null && registry != null && registry.isAvailable()) {
// 取消訂閱
registry.unsubscribe(getConsumerUrl(), this);
}
} catch (Throwable t) {
logger.warn("unexpeced error when unsubscribe service " + serviceKey + "from registry" + registry.getUrl(), t);
}
super.destroy(); // must be executed after unsubscribing
try {
// 清空所有的invoker
destroyAllInvokers();
} catch (Throwable t) {
logger.warn("Failed to destroy service " + serviceKey, t);
}
}

該方法是銷燬方法。

4.destroyAllInvokers

private void destroyAllInvokers() {
Map<String, Invoker<T>> localUrlInvokerMap = this.urlInvokerMap; // local reference
// 如果invoker集合不為空
if (localUrlInvokerMap != null) {
// 遍歷
for (Invoker<T> invoker : new ArrayList<Invoker<T>>(localUrlInvokerMap.values())) {
try {
// 銷燬invoker
invoker.destroy();
} catch (Throwable t) {
logger.warn("Failed to destroy service " + serviceKey + " to provider " + invoker.getUrl(), t);
}
}
// 清空集合
localUrlInvokerMap.clear();
}
methodInvokerMap = null;
}

該方法是關閉所有的invoker服務。

5.notify

@Override
public synchronized void notify(List<URL> urls) {
List<URL> invokerUrls = new ArrayList<URL>();
List<URL> routerUrls = new ArrayList<URL>();
List<URL> configuratorUrls = new ArrayList<URL>();
// 遍歷url
for (URL url : urls) {
// 獲得協議
String protocol = url.getProtocol();
// 獲得類別
String category = url.getParameter(Constants.CATEGORY_KEY, Constants.DEFAULT_CATEGORY);
// 如果是路由規則
if (Constants.ROUTERS_CATEGORY.equals(category)
|| Constants.ROUTE_PROTOCOL.equals(protocol)) {
// 則在路由規則集合中加入
routerUrls.add(url);
} else if (Constants.CONFIGURATORS_CATEGORY.equals(category)
|| Constants.OVERRIDE_PROTOCOL.equals(protocol)) {
// 如果是配置規則，則加入配置規則集合
configuratorUrls.add(url);
} else if (Constants.PROVIDERS_CATEGORY.equals(category)) {
// 如果是服務提供者，則加入服務提供者集合
invokerUrls.add(url);
} else {
logger.warn("Unsupported category " + category + " in notified url: " + url + " from registry " + getUrl().getAddress() + " to consumer " + NetUtils.getLocalHost());
}
}
// configurators
if (configuratorUrls != null && !configuratorUrls.isEmpty()) {
// 處理配置規則url集合
this.configurators = toConfigurators(configuratorUrls);
}
// routers
if (routerUrls != null && !routerUrls.isEmpty()) {
// 處理路由規則 URL 集合
List<Router> routers = toRouters(routerUrls);
if (routers != null) { // null - do nothing
// 並且設定路由集合
setRouters(routers);
}
}
List<Configurator> localConfigurators = this.configurators; // local reference
// merge override parameters
this.overrideDirectoryUrl = directoryUrl;
if (localConfigurators != null && !localConfigurators.isEmpty()) {
// 遍歷配置規則集合 逐個進行配置
for (Configurator configurator : localConfigurators) {
this.overrideDirectoryUrl = configurator.configure(overrideDirectoryUrl);
}
}
// providers
// 處理服務提供者 URL 集合
refreshInvoker(invokerUrls);
}

當服務有變化的時候，執行該方法。首先將url根據路由規則、服務提供者和配置規則三種類型分開，分別放入三個集合，然後對每個集合進行修改或者通知

6.refreshInvoker

private void refreshInvoker(List<URL> invokerUrls) {
if (invokerUrls != null && invokerUrls.size() == 1 && invokerUrls.get(0) != null
&& Constants.EMPTY_PROTOCOL.equals(invokerUrls.get(0).getProtocol())) {
// 設定禁止訪問
this.forbidden = true; // Forbid to access
// methodInvokerMap 置空
this.methodInvokerMap = null; // Set the method invoker map to null
// 關閉所有的invoker
destroyAllInvokers(); // Close all invokers
} else {
// 關閉禁止訪問
this.forbidden = false; // Allow to access
// 引用老的 urlInvokerMap
Map<String, Invoker<T>> oldUrlInvokerMap = this.urlInvokerMap; // local reference
// 傳入的 invokerUrls 為空，說明是路由規則或配置規則發生改變，此時 invokerUrls 是空的，直接使用 cachedInvokerUrls 。
if (invokerUrls.isEmpty() && this.cachedInvokerUrls != null) {
invokerUrls.addAll(this.cachedInvokerUrls);
} else {
// 否則把所有的invokerUrls加入快取
this.cachedInvokerUrls = new HashSet<URL>();
this.cachedInvokerUrls.addAll(invokerUrls);//Cached invoker urls, convenient for comparison
}
// 如果invokerUrls為空，則直接返回
if (invokerUrls.isEmpty()) {
return;
}
// 將傳入的 invokerUrls ，轉成新的 urlInvokerMap
Map<String, Invoker<T>> newUrlInvokerMap = toInvokers(invokerUrls);// Translate url list to Invoker map
// 轉換出新的 methodInvokerMap
Map<String, List<Invoker<T>>> newMethodInvokerMap = toMethodInvokers(newUrlInvokerMap); // Change method name to map Invoker Map
// state change
// If the calculation is wrong, it is not processed.
// 如果為空，則列印錯誤日誌並且返回
if (newUrlInvokerMap == null || newUrlInvokerMap.size() == 0) {
logger.error(new IllegalStateException("urls to invokers error .invokerUrls.size :" + invokerUrls.size() + ", invoker.size :0. urls :" + invokerUrls.toString()));
return;
}
// 若服務引用多 group ，則按照 method + group 聚合 Invoker 集合
this.methodInvokerMap = multiGroup ? toMergeMethodInvokerMap(newMethodInvokerMap) : newMethodInvokerMap;
this.urlInvokerMap = newUrlInvokerMap;
try {
// 銷燬不再使用的 Invoker 集合
destroyUnusedInvokers(oldUrlInvokerMap, newUrlInvokerMap); // Close the unused Invoker
} catch (Exception e) {
logger.warn("destroyUnusedInvokers error. ", e);
}
}
}

該方法是處理服務提供者 URL 集合。根據 invokerURL 列表轉換為 invoker 列表。轉換規則如下：

如果 url 已經被轉換為 invoker ，則不在重新引用，直接從快取中獲取，注意如果 url 中任何一個引數變更也會重新引用。
如果傳入的 invoker 列表不為空，則表示最新的 invoker 列表。
如果傳入的 invokerUrl 列表是空，則表示只是下發的 override 規則或 route 規則，需要重新交叉對比，決定是否需要重新引用。

7.toMergeMethodInvokerMap

private Map<String, List<Invoker<T>>> toMergeMethodInvokerMap(Map<String, List<Invoker<T>>> methodMap) {
// 迴圈方法，按照 method + group 聚合 Invoker 集合
Map<String, List<Invoker<T>>> result = new HashMap<String, List<Invoker<T>>>();
// 遍歷方法集合
for (Map.Entry<String, List<Invoker<T>>> entry : methodMap.entrySet()) {
// 獲得方法
String method = entry.getKey();
// 獲得invoker集合
List<Invoker<T>> invokers = entry.getValue();
// 獲得組集合
Map<String, List<Invoker<T>>> groupMap = new HashMap<String, List<Invoker<T>>>();
// 遍歷invoker集合
for (Invoker<T> invoker : invokers) {
// 獲得url攜帶的組配置
String group = invoker.getUrl().getParameter(Constants.GROUP_KEY, "");
// 獲得該組對應的invoker集合
List<Invoker<T>> groupInvokers = groupMap.get(group);
// 如果為空，則新建立一個，然後加入集合
if (groupInvokers == null) {
groupInvokers = new ArrayList<Invoker<T>>();
groupMap.put(group, groupInvokers);
}
groupInvokers.add(invoker);
}
// 如果只有一個組
if (groupMap.size() == 1) {
// 返回該組的invoker集合
result.put(method, groupMap.values().iterator().next());
} else if (groupMap.size() > 1) {
// 如果不止一個組
List<Invoker<T>> groupInvokers = new ArrayList<Invoker<T>>();
// 遍歷組
for (List<Invoker<T>> groupList : groupMap.values()) {
// 每次從叢集中選擇一個invoker加入groupInvokers
groupInvokers.add(cluster.join(new StaticDirectory<T>(groupList)));
}
// 加入需要返回的集合
result.put(method, groupInvokers);
} else {
result.put(method, invokers);
}
}
return result;
}

該方法是通過按照 method + group 來聚合 Invoker 集合。

8.toRouters

private List<Router> toRouters(List<URL> urls) {
List<Router> routers = new ArrayList<Router>();
// 如果為空，則直接返回空集合
if (urls == null || urls.isEmpty()) {
return routers;
}
if (urls != null && !urls.isEmpty()) {
// 遍歷url集合
for (URL url : urls) {
// 如果為empty協議，則直接跳過
if (Constants.EMPTY_PROTOCOL.equals(url.getProtocol())) {
continue;
}
// 獲得路由規則
String routerType = url.getParameter(Constants.ROUTER_KEY);
if (routerType != null && routerType.length() > 0) {
// 設定協議
url = url.setProtocol(routerType);
}
try {
// 獲得路由
Router router = routerFactory.getRouter(url);
if (!routers.contains(router))
// 加入集合
routers.add(router);
} catch (Throwable t) {
logger.error("convert router url to router error, url: " + url, t);
}
}
}
return routers;
}

該方法是對url集合進行路由的解析，返回路由集合。

9.toInvokers

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>();
// 獲得引用服務的協議
String queryProtocols = this.queryMap.get(Constants.PROTOCOL_KEY);
// 遍歷url
for (URL providerUrl : urls) {
// If protocol is configured at the reference side, only the matching protocol is selected
// 如果在參考側配置協議，則僅選擇匹配協議
if (queryProtocols != null && queryProtocols.length() > 0) {
boolean accept = false;
// 分割協議
String[] acceptProtocols = queryProtocols.split(",");
// 遍歷協議
for (String acceptProtocol : acceptProtocols) {
// 如果匹配，則是接受的協議
if (providerUrl.getProtocol().equals(acceptProtocol)) {
accept = true;
break;
}
}
if (!accept) {
continue;
}
}
// 如果協議是empty，則跳過
if (Constants.EMPTY_PROTOCOL.equals(providerUrl.getProtocol())) {
continue;
}
// 如果該協議不是dubbo支援的，則列印錯誤日誌，跳過
if (!ExtensionLoader.getExtensionLoader(Protocol.class).hasExtension(providerUrl.getProtocol())) {
logger.error(new IllegalStateException("Unsupported protocol " + providerUrl.getProtocol() + " in notified url: " + providerUrl + " from registry " + getUrl().getAddress() + " to consumer " + NetUtils.getLocalHost()
+ ", supported protocol: " + ExtensionLoader.getExtensionLoader(Protocol.class).getSupportedExtensions()));
continue;
}
// 合併url引數
URL url = mergeUrl(providerUrl);

String key = url.toFullString(); // The parameter urls are sorted
if (keys.contains(key)) { // Repeated url
continue;
}
// 新增到keys
keys.add(key);
// Cache key is url that does not merge with consumer side parameters, regardless of how the consumer combines parameters, if the server url changes, then refer again
// 如果服務端 URL 發生變化，則重新 refer 引用
Map<String, Invoker<T>> localUrlInvokerMap = this.urlInvokerMap; // local reference
Invoker<T> invoker = localUrlInvokerMap == null ? null : localUrlInvokerMap.get(key);
if (invoker == null) { // Not in the cache, refer again
try {
// 判斷是否開啟
boolean enabled = true;
// 獲得enabled配置
if (url.hasParameter(Constants.DISABLED_KEY)) {
enabled = !url.getParameter(Constants.DISABLED_KEY, false);
} else {
enabled = url.getParameter(Constants.ENABLED_KEY, true);
}
// 若開啟，建立 Invoker 物件
if (enabled) {
invoker = new InvokerDelegate<T>(protocol.refer(serviceType, url), url, providerUrl);
}
} catch (Throwable t) {
logger.error("Failed to refer invoker for interface:" + serviceType + ",url:(" + url + ")" + t.getMessage(), t);
}
// 新增到 newUrlInvokerMap 中
if (invoker != null) { // Put new invoker in cache
newUrlInvokerMap.put(key, invoker);
}
} else {
newUrlInvokerMap.put(key, invoker);
}
}
// 清空 keys
keys.clear();
return newUrlInvokerMap;
}

該方法是將url轉換為呼叫者，如果url已被引用，則不會重新引用。

10.mergeUrl

private URL mergeUrl(URL providerUrl) {
// 合併消費端引數
providerUrl = ClusterUtils.mergeUrl(providerUrl, queryMap); // Merge the consumer side parameters

// 合併配置規則
List<Configurator> localConfigurators = this.configurators; // local reference
if (localConfigurators != null && !localConfigurators.isEmpty()) {
for (Configurator configurator : localConfigurators) {
providerUrl = configurator.configure(providerUrl);
}
}

// 不檢查連線是否成功，總是建立 Invoker
providerUrl = providerUrl.addParameter(Constants.CHECK_KEY, String.valueOf(false)); // Do not check whether the connection is successful or not, always create Invoker!

// The combination of directoryUrl and override is at the end of notify, which can't be handled here
// 合併提供者引數，因為 directoryUrl 與 override 合併是在 notify 的最後，這裡不能夠處理
this.overrideDirectoryUrl = this.overrideDirectoryUrl.addParametersIfAbsent(providerUrl.getParameters()); // Merge the provider side parameters

// 1.0版本相容
if ((providerUrl.getPath() == null || providerUrl.getPath().length() == 0)
&& "dubbo".equals(providerUrl.getProtocol())) { // Compatible version 1.0
//fix by tony.chenl DUBBO-44
String path = directoryUrl.getParameter(Constants.INTERFACE_KEY);
if (path != null) {
int i = path.indexOf('/');
if (i >= 0) {
path = path.substring(i + 1);
}
i = path.lastIndexOf(':');
if (i >= 0) {
path = path.substring(0, i);
}
providerUrl = providerUrl.setPath(path);
}
}
return providerUrl;
}

該方法是合併 URL 引數，優先順序為配置規則 > 服務消費者配置 > 服務提供者配置.

11.toMethodInvokers

private Map<String, List<Invoker<T>>> toMethodInvokers(Map<String, Invoker<T>> invokersMap) {
Map<String, List<Invoker<T>>> newMethodInvokerMap = new HashMap<String, List<Invoker<T>>>();
// According to the methods classification declared by the provider URL, the methods is compatible with the registry to execute the filtered methods
List<Invoker<T>> invokersList = new ArrayList<Invoker<T>>();
if (invokersMap != null && invokersMap.size() > 0) {
// 遍歷呼叫者列表
for (Invoker<T> invoker : invokersMap.values()) {
String parameter = invoker.getUrl().getParameter(Constants.METHODS_KEY);
// 按服務提供者 URL 所宣告的 methods 分類
if (parameter != null && parameter.length() > 0) {
// 分割引數得到方法集合
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)) {
// 獲得該方法對應的invoker，如果為空，則建立
List<Invoker<T>> methodInvokers = newMethodInvokerMap.get(method);
if (methodInvokers == null) {
methodInvokers = new ArrayList<Invoker<T>>();
newMethodInvokerMap.put(method, methodInvokers);
}
methodInvokers.add(invoker);
}
}
}
}
invokersList.add(invoker);
}
}
// 根據路由規則，匹配合適的 Invoker 集合。
List<Invoker<T>> newInvokersList = route(invokersList, null);
// 新增 `newInvokersList` 到 `newMethodInvokerMap` 中，表示該服務提供者的全量 Invoker 集合
newMethodInvokerMap.put(Constants.ANY_VALUE, newInvokersList);
if (serviceMethods != null && serviceMethods.length > 0) {
// 迴圈方法，獲得每個方法路由匹配的invoker集合
for (String method : serviceMethods) {
List<Invoker<T>> methodInvokers = newMethodInvokerMap.get(method);
if (methodInvokers == null || methodInvokers.isEmpty()) {
methodInvokers = newInvokersList;
}
newMethodInvokerMap.put(method, route(methodInvokers, method));
}
}
// sort and unmodifiable
// 迴圈排序每個方法的 Invoker 集合，排序
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);
}

該方法是將呼叫者列表轉換為與方法的對映關係。

12.destroyUnusedInvokers

private void destroyUnusedInvokers(Map<String, Invoker<T>> oldUrlInvokerMap, Map<String, Invoker<T>> newUrlInvokerMap) {
if (newUrlInvokerMap == null || newUrlInvokerMap.size() == 0) {
destroyAllInvokers();
return;
}
// check deleted invoker
// 記錄已經刪除的invoker
List<String> deleted = null;
if (oldUrlInvokerMap != null) {
Collection<Invoker<T>> newInvokers = newUrlInvokerMap.values();
// 遍歷舊的invoker集合
for (Map.Entry<String, Invoker<T>> entry : oldUrlInvokerMap.entrySet()) {
if (!newInvokers.contains(entry.getValue())) {
if (deleted == null) {
deleted = new ArrayList<String>();
}
// 加入該invoker
deleted.add(entry.getKey());
}
}
}

if (deleted != null) {
// 遍歷需要刪除的invokerurl集合
for (String url : deleted) {
if (url != null) {
// 移除該url
Invoker<T> invoker = oldUrlInvokerMap.remove(url);
if (invoker != null) {
try {
// 銷燬invoker
invoker.destroy();
if (logger.isDebugEnabled()) {
logger.debug("destroy invoker[" + invoker.getUrl() + "] success. ");
}
} catch (Exception e) {
logger.warn("destroy invoker[" + invoker.getUrl() + "] faild. " + e.getMessage(), e);
}
}
}
}
}
}

該方法是銷燬不再使用的 Invoker 集合。

13.doList

@Override
public List<Invoker<T>> doList(Invocation invocation) {
// 如果禁止訪問，則丟擲異常
if (forbidden) {
// 1. No service provider 2. Service providers are disabled
throw new RpcException(RpcException.FORBIDDEN_EXCEPTION,
"No provider available from registry " + getUrl().getAddress() + " for service " + getConsumerUrl().getServiceKey() + " on consumer " +NetUtils.getLocalHost()
+ " use dubbo version " + Version.getVersion() + ", please check status of providers(disabled, not registered or in blacklist).");
}
List<Invoker<T>> invokers = null;
Map<String, List<Invoker<T>>> localMethodInvokerMap = this.methodInvokerMap; // local reference
if (localMethodInvokerMap != null && localMethodInvokerMap.size() > 0) {
// 獲得方法名
String methodName = RpcUtils.getMethodName(invocation);
// 獲得引數名
Object[] args = RpcUtils.getArguments(invocation);
if (args != null && args.length > 0 && args[0] != null
&& (args[0] instanceof String || args[0].getClass().isEnum())) {
// 根據第一個引數列舉路由
invokers = localMethodInvokerMap.get(methodName + "." + args[0]); // The routing can be enumerated according to the first parameter
}
if (invokers == null) {
// 根據方法名獲得 Invoker 集合
invokers = localMethodInvokerMap.get(methodName);
}
if (invokers == null) {
// 使用全量 Invoker 集合。例如，`#$echo(name)` ，回聲方法
invokers = localMethodInvokerMap.get(Constants.ANY_VALUE);
}
if (invokers == null) {
// 使用 `methodInvokerMap` 第一個 Invoker 集合。防禦性程式設計。
Iterator<List<Invoker<T>>> iterator = localMethodInvokerMap.values().iterator();
if (iterator.hasNext()) {
invokers = iterator.next();
}
}
}
return invokers == null ? new ArrayList<Invoker<T>>(0) : invokers;
}

該方法是通過會話域來獲得Invoker集合。

後記

該部分相關的原始碼解析地址：https://github.com/CrazyHZM/i...

該文章講解了叢集中關於directory實現的部分，關鍵是RegistryDirectory，其中涉及到眾多方法，需要好好品味。接下來我將開始對叢集模組關於loadbalance部分進行講解。

dubbo原始碼解析（三十七）叢集——directory

叢集——directory

前言

原始碼分析

（一）AbstractDirectory

1.屬性

2.list

3.setRouters

（二）StaticDirectory

（三）RegistryDirectory

1.屬性

2.toConfigurators

3.destroy

4.destroyAllInvokers

5.notify

6.refreshInvoker

7.toMergeMethodInvokerMap

8.toRouters

9.toInvokers

10.mergeUrl

11.toMethodInvokers

12.destroyUnusedInvokers

13.doList

後記

您可能也會喜歡…