Dubbo源碼解析（三十七）集羣——directory

集羣——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 列表。轉換規則以下：

1. 若是 url 已經被轉換爲 invoker ，則不在從新引用，直接從緩存中獲取，注意若是 url 中任何一個參數變動也會從新引用。
2. 若是傳入的 invoker 列表不爲空，則表示最新的 invoker 列表。
3. 若是傳入的 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) {
        // 遍歷須要刪除的invoker  url集合
        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部分進行講解。