资讯专栏INFORMATION COLUMN

dubbo源码解析(三十七)集群——directory

blastz / 647人阅读

摘要:首先将根据路由规则服务提供者和配置规则三种类型分开,分别放入三个集合,然后对每个集合进行修改或者通知设置禁止访问置空关闭所有的关闭禁止访问引用老的传入的为空,说明是路由规则或配置规则发生改变,此时是空的,直接使用。

集群——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 routers;
2.list
@Override
public List> list(Invocation invocation) throws RpcException {
    // 如果销毁,则抛出异常
    if (destroyed) {
        throw new RpcException("Directory already destroyed .url: " + getUrl());
    }
    // 调用doList来获得Invoker集合
    List> invokers = doList(invocation);
    // 获得路由集合
    List 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 routers) {
    // copy list
    // 复制路由集合
    routers = routers == null ? new ArrayList() : new ArrayList(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 extends AbstractDirectory {

    private final List> invokers;

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

    public StaticDirectory(List> invokers, List routers) {
        this(null, invokers, routers);
    }

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

    public StaticDirectory(URL url, List> invokers, List 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 getInterface() {
        return invokers.get(0).getInterface();
    }

    @Override
    public boolean isAvailable() {
        if (isDestroyed()) {
            return false;
        }
        // 遍历invokers,如果有一个可用,则可用
        for (Invoker invoker : invokers) {
            if (invoker.isAvailable()) {
                return true;
            }
        }
        return false;
    }

    @Override
    public void destroy() {
        if (isDestroyed()) {
            return;
        }
        super.destroy();
        // 遍历invokers,销毁所有的invoker
        for (Invoker invoker : invokers) {
            invoker.destroy();
        }
        // 清除集合
        invokers.clear();
    }

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

        return invokers;
    }

}

该类我就不多讲解,比较简单。

(三)RegistryDirectory

该类继承了AbstractDirectory类,是基于注册中心的动态 Directory 实现类,会根据注册中心的推送变更 List

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 serviceType; // Initialization at construction time, assertion not null
/**
 * 消费者URL的配置项 Map
 */
private final Map 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 
 * Rule two: for all providers <* ,timeout=5000>
 * 配置规则数组
 */
private volatile List configurators; // The initial value is null and the midway may be assigned to null, please use the local variable reference

// Map cache service url to invoker mapping.
/**
 * url与服务提供者 Invoker 集合的映射缓存
 */
private volatile Map> urlInvokerMap; // The initial value is null and the midway may be assigned to null, please use the local variable reference

// Map cache service method to invokers mapping.
/**
 * 方法名和服务提供者 Invoker 集合的映射缓存
 */
private volatile Map>> methodInvokerMap; // The initial value is null and the midway may be assigned to null, please use the local variable reference

// Set cache invokeUrls to invokers mapping.
/**
 * 服务提供者Invoker 集合缓存
 */
private volatile Set 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 toConfigurators(List urls) {
    // 如果为空,则返回空集合
    if (urls == null || urls.isEmpty()) {
        return Collections.emptyList();
    }

    List configurators = new ArrayList(urls.size());
    // 遍历url集合
    for (URL url : urls) {
        //如果是协议是empty的值,则清空配置集合
        if (Constants.EMPTY_PROTOCOL.equals(url.getProtocol())) {
            configurators.clear();
            break;
        }
        // 覆盖的参数集合
        Map override = new HashMap(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> localUrlInvokerMap = this.urlInvokerMap; // local reference
    // 如果invoker集合不为空
    if (localUrlInvokerMap != null) {
        // 遍历
        for (Invoker invoker : new ArrayList>(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 urls) {
    List invokerUrls = new ArrayList();
    List routerUrls = new ArrayList();
    List configuratorUrls = new ArrayList();
    // 遍历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 routers = toRouters(routerUrls);
        if (routers != null) { // null - do nothing
            // 并且设置路由集合
            setRouters(routers);
        }
    }
    List 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 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> oldUrlInvokerMap = this.urlInvokerMap; // local reference
        // 传入的 invokerUrls 为空,说明是路由规则或配置规则发生改变,此时 invokerUrls 是空的,直接使用 cachedInvokerUrls 。
        if (invokerUrls.isEmpty() && this.cachedInvokerUrls != null) {
            invokerUrls.addAll(this.cachedInvokerUrls);
        } else {
            // 否则把所有的invokerUrls加入缓存
            this.cachedInvokerUrls = new HashSet();
            this.cachedInvokerUrls.addAll(invokerUrls);//Cached invoker urls, convenient for comparison
        }
        // 如果invokerUrls为空,则直接返回
        if (invokerUrls.isEmpty()) {
            return;
        }
        // 将传入的 invokerUrls ,转成新的 urlInvokerMap
        Map> newUrlInvokerMap = toInvokers(invokerUrls);// Translate url list to Invoker map
        // 转换出新的 methodInvokerMap
        Map>> 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>> toMergeMethodInvokerMap(Map>> methodMap) {
    // 循环方法,按照 method + group 聚合 Invoker 集合
    Map>> result = new HashMap>>();
    // 遍历方法集合
    for (Map.Entry>> entry : methodMap.entrySet()) {
        // 获得方法
        String method = entry.getKey();
        // 获得invoker集合
        List> invokers = entry.getValue();
        // 获得组集合
        Map>> groupMap = new HashMap>>();
        // 遍历invoker集合
        for (Invoker invoker : invokers) {
            // 获得url携带的组配置
            String group = invoker.getUrl().getParameter(Constants.GROUP_KEY, "");
            // 获得该组对应的invoker集合
            List> groupInvokers = groupMap.get(group);
            // 如果为空,则新创建一个,然后加入集合
            if (groupInvokers == null) {
                groupInvokers = new ArrayList>();
                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> groupInvokers = new ArrayList>();
            // 遍历组
            for (List> groupList : groupMap.values()) {
                // 每次从集群中选择一个invoker加入groupInvokers
                groupInvokers.add(cluster.join(new StaticDirectory(groupList)));
            }
            // 加入需要返回的集合
            result.put(method, groupInvokers);
        } else {
            result.put(method, invokers);
        }
    }
    return result;
}

该方法是通过按照 method + group 来聚合 Invoker 集合。

8.toRouters
private List toRouters(List urls) {
    List routers = new ArrayList();
    // 如果为空,则直接返回空集合
    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> toInvokers(List urls) {
    Map> newUrlInvokerMap = new HashMap>();
    // 如果为空,则返回空集合
    if (urls == null || urls.isEmpty()) {
        return newUrlInvokerMap;
    }
    Set keys = new HashSet();
    // 获得引用服务的协议
    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> localUrlInvokerMap = this.urlInvokerMap; // local reference
        Invoker 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(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 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>> toMethodInvokers(Map> invokersMap) {
    Map>> newMethodInvokerMap = new HashMap>>();
    // According to the methods classification declared by the provider URL, the methods is compatible with the registry to execute the filtered methods
    List> invokersList = new ArrayList>();
    if (invokersMap != null && invokersMap.size() > 0) {
        // 遍历调用者列表
        for (Invoker 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> methodInvokers = newMethodInvokerMap.get(method);
                            if (methodInvokers == null) {
                                methodInvokers = new ArrayList>();
                                newMethodInvokerMap.put(method, methodInvokers);
                            }
                            methodInvokers.add(invoker);
                        }
                    }
                }
            }
            invokersList.add(invoker);
        }
    }
    // 根据路由规则,匹配合适的 Invoker 集合。
    List> 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> 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(newMethodInvokerMap.keySet())) {
        List> 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> oldUrlInvokerMap, Map> newUrlInvokerMap) {
    if (newUrlInvokerMap == null || newUrlInvokerMap.size() == 0) {
        destroyAllInvokers();
        return;
    }
    // check deleted invoker
    // 记录已经删除的invoker
    List deleted = null;
    if (oldUrlInvokerMap != null) {
        Collection> newInvokers = newUrlInvokerMap.values();
        // 遍历旧的invoker集合
        for (Map.Entry> entry : oldUrlInvokerMap.entrySet()) {
            if (!newInvokers.contains(entry.getValue())) {
                if (deleted == null) {
                    deleted = new ArrayList();
                }
                // 加入该invoker
                deleted.add(entry.getKey());
            }
        }
    }

    if (deleted != null) {
        // 遍历需要删除的invoker  url集合
        for (String url : deleted) {
            if (url != null) {
                // 移除该url
                Invoker 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> 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> invokers = null;
    Map>> 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>> iterator = localMethodInvokerMap.values().iterator();
            if (iterator.hasNext()) {
                invokers = iterator.next();
            }
        }
    }
    return invokers == null ? new ArrayList>(0) : invokers;
}

该方法是通过会话域来获得Invoker集合。

后记
该部分相关的源码解析地址:https://github.com/CrazyHZM/i...

该文章讲解了集群中关于directory实现的部分,关键是RegistryDirectory,其中涉及到众多方法,需要好好品味。接下来我将开始对集群模块关于loadbalance部分进行讲解。

文章版权归作者所有,未经允许请勿转载,若此文章存在违规行为,您可以联系管理员删除。

转载请注明本文地址:https://www.ucloud.cn/yun/77406.html

相关文章

  • dubbo源码解析三十九)集群——merger

    摘要:源码分析一创建该类实现了接口,是分组集合的集群实现。三工厂类,获得指定类型的对象。后记该部分相关的源码解析地址该文章讲解了集群中关于分组聚合实现的部分。接下来我将开始对集群模块关于路由部分进行讲解。 集群——merger 目标:介绍dubbo中集群的分组聚合,介绍dubbo-cluster下merger包的源码。 前言 按组合并返回结果 ,比如菜单服务,接口一样,但有多种实现,用gro...

    lscho 评论0 收藏0
  • dubbo源码解析三十五)集群——cluster

    摘要:失败安全,出现异常时,直接忽略。失败自动恢复,在调用失败后,返回一个空结果给服务提供者。源码分析一该类实现了接口,是集群的抽象类。 集群——cluster 目标:介绍dubbo中集群容错的几种模式,介绍dubbo-cluster下support包的源码。 前言 集群容错还是很好理解的,就是当你调用失败的时候所作出的措施。先来看看有哪些模式: showImg(https://segmen...

    gself 评论0 收藏0
  • dubbo源码解析(四十八)异步化改造

    摘要:大揭秘异步化改造目标从源码的角度分析的新特性中对于异步化的改造原理。看源码解析四十六消费端发送请求过程讲到的十四的,在以前的逻辑会直接在方法中根据配置区分同步异步单向调用。改为关于可以参考源码解析十远程通信层的六。 2.7大揭秘——异步化改造 目标:从源码的角度分析2.7的新特性中对于异步化的改造原理。 前言 dubbo中提供了很多类型的协议,关于协议的系列可以查看下面的文章: du...

    lijinke666 评论0 收藏0
  • dubbo源码解析(四十六)消费端发送请求过程

    摘要:可以参考源码解析二十四远程调用协议的八。十六的该类也是用了适配器模式,该类主要的作用就是增加了心跳功能,可以参考源码解析十远程通信层的四。二十的可以参考源码解析十七远程通信的一。 2.7大揭秘——消费端发送请求过程 目标:从源码的角度分析一个服务方法调用经历怎么样的磨难以后到达服务端。 前言 前一篇文章讲到的是引用服务的过程,引用服务无非就是创建出一个代理。供消费者调用服务的相关方法。...

    fish 评论0 收藏0
  • dubbo源码解析三十八)集群——LoadBalance

    摘要:集群目标介绍中集群的负载均衡,介绍下包的源码。源码分析一该类实现了接口,是负载均衡的抽象类,提供了权重计算的功能。四该类是负载均衡基于一致性的逻辑实现。 集群——LoadBalance 目标:介绍dubbo中集群的负载均衡,介绍dubbo-cluster下loadBalance包的源码。 前言 负载均衡,说的通俗点就是要一碗水端平。在这个时代,公平是很重要的,在网络请求的时候同样是这个...

    不知名网友 评论0 收藏0

发表评论

0条评论

blastz

|高级讲师

TA的文章

阅读更多
最新活动
阅读需要支付1元查看
<