Dubbo源碼解析之Directory

注：Dubbo版本是2.5.7

                              

                                                     圖1 AbstractDirectory有倆個實現類

1.StaticDirectory

    StaticDirectory中的List<Invoker>是固定的，源碼以下所示。構造方法中傳入List<Invoker>後，類內部就沒有改變了。

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

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

2.RegistryDirectory

    RegistryDirectory與StaticDirectory不一樣的是，RegistryDirectory中的屬性會變化——註冊中心的服務變更時，會推送到這裏。

    以前的Dubbo集羣容錯架構中，從Directory中獲取List<Invoker>，獲取List<Invoker>與RegistryDirectory的methodInvokerMap有關，以下所示。

public List<Invoker<T>> doList(Invocation invocation) {
    if (forbidden) {
        // 1. 沒有服務提供者 2. 服務提供者被禁用
        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() + ", may be providers disabled or not registered ?");
    }
    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]); // 可根據第一個參數枚舉路由
        }
        if (invokers == null) {
            invokers = localMethodInvokerMap.get(methodName);
        }
        if (invokers == null) {
            invokers = localMethodInvokerMap.get(Constants.ANY_VALUE);
        }
        if (invokers == null) {
            Iterator<List<Invoker<T>>> iterator = localMethodInvokerMap.values().iterator();
            if (iterator.hasNext()) {
                invokers = iterator.next();
            }
        }
    }
    return invokers == null ? new ArrayList<Invoker<T>>(0) : invokers;
}

    那麼methodInvokeMap的值是從哪來的，以下所示，notify(List<URL>)是NotifyListener的方法，方法中傳入一個List<URL>，解析這個List<URL>，configurators的值、routers的值都是在這個方法中更新的，methodInvokerMap是在refreshInvoker中更新的。以下所示。

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>();
    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.size() > 0) {
        this.configurators = toConfigurators(configuratorUrls);
    }
    // routers
    if (routerUrls != null && routerUrls.size() > 0) {
        List<Router> routers = toRouters(routerUrls);
        if (routers != null) { // null - do nothing
            setRouters(routers);
        }
    }
    List<Configurator> localConfigurators = this.configurators; // local reference
    // 合併override參數
    this.overrideDirectoryUrl = directoryUrl;
    if (localConfigurators != null && localConfigurators.size() > 0) {
        for (Configurator configurator : localConfigurators) {
            this.overrideDirectoryUrl = configurator.configure(overrideDirectoryUrl);
        }
    }
    // providers
    refreshInvoker(invokerUrls);
}

    下面咱們來看refreshInvoker中的實現，涉及緩存之類的，可是咱們丟開細節，看主要的，能夠看到this.methodInvokerMap = multiGroup ? toMergeMethodInvokerMap(newMethodInvokerMap) : newMethodInvokerMap。這個方法中還更新了urlInvokerMap。以下所示。

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; // 禁止訪問
        this.methodInvokerMap = null; // 置空列表
        destroyAllInvokers(); // 關閉全部Invoker
    } else {
        this.forbidden = false; // 容許訪問
        Map<String, Invoker<T>> oldUrlInvokerMap = this.urlInvokerMap; // local reference
        if (invokerUrls.size() == 0 && this.cachedInvokerUrls != null) {
            invokerUrls.addAll(this.cachedInvokerUrls);
        } else {
            this.cachedInvokerUrls = new HashSet<URL>();
            this.cachedInvokerUrls.addAll(invokerUrls);//緩存invokerUrls列表，便於交叉對比
        }
        if (invokerUrls.size() == 0) {
            return;
        }
        Map<String, Invoker<T>> newUrlInvokerMap = toInvokers(invokerUrls);// 將URL列表轉成Invoker列表
        Map<String, List<Invoker<T>>> newMethodInvokerMap = toMethodInvokers(newUrlInvokerMap); // 換方法名映射Invoker列表
        // state change
        //若是計算錯誤，則不進行處理.
        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;
        }
        this.methodInvokerMap = multiGroup ? toMergeMethodInvokerMap(newMethodInvokerMap) : newMethodInvokerMap;
        this.urlInvokerMap = newUrlInvokerMap;
        try {
            destroyUnusedInvokers(oldUrlInvokerMap, newUrlInvokerMap); // 關閉未使用的Invoker
        } catch (Exception e) {
            logger.warn("destroyUnusedInvokers error. ", e);
        }
    }
}

3.思考

    1.RegistryDirectory中，methodInvokerMap是否有多線程的問題？由於服務註冊中心變動的同時，會有服務消費者訪問。methodInvokerMapnotify(List<URL> urls)方法上有synchronized；refreshInvoker(List<URL> invokerUrls)中更新methodInvokerMap是對其徹底賦值，而不是修改methodInvokerMap中的元素。基於這倆點來看是沒有線程安全問題的。

    2.RegistryDirectory中爲性能考慮，用了緩存。其實不止這裏，Spring的不少地方都用了緩存。若是咱們的項目業務裏相似的場景，咱們是否能夠添加緩存的功能，哪些框架源碼能夠爲咱們提供借鑑呢？