前言

上篇文章《Dubbo之服務暴露》分析 Dubbo 服務是如何暴露的，本文接著分析 Dubbo 服務的消費流程。主要從以下幾個方面進行分析：註冊中心的暴露；通過註冊中心進行服務消費通知；直連服務進行消費。
服務消費端啟動時，將自身的資訊註冊到註冊中心的目錄，同時還訂閱服務提供方的目錄，當服務提供方的 URL 發生更改時，實時獲取新的資料。

服務消費端流程

下面是一個服務消費的流程圖：

上圖中可以看到，服務消費的流程與服務暴露的流程有點類似逆向的。同樣，Dubbo 服務也是分為兩個大步驟：第一步就是將遠端服務通過Protocol轉換成Invoker(概念在上篇文章中有解釋)。第二步通過動態代理將Invoker

轉換成消費服務需要的介面。

org.apache.dubbo.config.ReferenceConfig類是ReferenceBean的父類，與生產端服務的ServiceBean一樣，存放著解析出來的 XML 和註解資訊。類關係如下：

服務初始化中轉換的入口

當我們消費端呼叫本地介面就能實現遠端服務的呼叫，這是怎麼實現的呢？根據上面的流程圖，來分析消費原理。
在消費端進行初始化時ReferenceConfig#init，會執行ReferenceConfig#createProxy來完成這一系列操作。以下為ReferenceConfig#createProxy主要的程式碼部分：

private T createProxy(Map<String, String> map) {
    // 判斷是否為 Jvm 本地引用
    if (shouldJvmRefer(map)) {
        // 通過 injvm 協議，獲取本地服務
        URL url = new URL(LOCAL_PROTOCOL, LOCALHOST_VALUE, 0, interfaceClass.getName()).addParameters(map);
        invoker = REF_PROTOCOL.refer(interfaceClass, url);
    } else {
        urls.clear();
        // 判斷是否有自定義的直連地址，或註冊中心地址
        if (url != null && url.length() > 0) { 
            String[] us = SEMICOLON_SPLIT_PATTERN.split(url);
            if (us != null && us.length > 0) {
                for (String u : us) {
                    URL url = URL.valueOf(u);
                    if (StringUtils.isEmpty(url.getPath())) {
                        url = url.setPath(interfaceName);
                    }
                    if (UrlUtils.isRegistry(url)) {
                        // 如果是註冊中心Protocol型別，則向地址中新增 refer 服務消費元資料
                        urls.add(url.addParameterAndEncoded(REFER_KEY, StringUtils.toQueryString(map)));
                    } else {
                        // 直連服務提供端
                        urls.add(ClusterUtils.mergeUrl(url, map));
                    }
                }
            }
        } else {
            // 組裝註冊中心的配置
            if (!LOCAL_PROTOCOL.equalsIgnoreCase(getProtocol())) {
                // 檢查配置中心
                checkRegistry();
                List<URL> us = ConfigValidationUtils.loadRegistries(this, false);
                if (CollectionUtils.isNotEmpty(us)) {
                    for (URL u : us) {
                        URL monitorUrl = ConfigValidationUtils.loadMonitor(this, u);
                        if (monitorUrl != null) {
                            // 監控上報資訊
                            map.put(MONITOR_KEY, URL.encode(monitorUrl.toFullString()));
                        }
                        // 註冊中心地址新增 refer 服務消費元資料
                        urls.add(u.addParameterAndEncoded(REFER_KEY, StringUtils.toQueryString(map)));
                    }
                }
            }
        }

        // 只有一條註冊中心資料，即單註冊中心
        if (urls.size() == 1) {
            // 將遠端服務轉化成 Invoker
            invoker = REF_PROTOCOL.refer(interfaceClass, urls.get(0));
        } else {
            // 因為多註冊中心就會存在多個 Invoker，這裡用儲存在 List 中
            List<Invoker<?>> invokers = new ArrayList<Invoker<?>>();
            URL registryURL = null;
            for (URL url : urls) {
                // 將每個註冊中心轉換成 Invoker 資料
                invokers.add(REF_PROTOCOL.refer(interfaceClass, url));
                if (UrlUtils.isRegistry(url)) {
                    // 會覆蓋前遍歷的註冊中心，使用最後一條註冊中心資料
                    registryURL = url;
                }
            }
            if (registryURL != null) {
                // 預設使用 zone-aware 策略來處理多個訂閱
                URL u = registryURL.addParameterIfAbsent(CLUSTER_KEY, ZoneAwareCluster.NAME);
                // 將轉換後的多個 Invoker 合併成一個
                invoker = CLUSTER.join(new StaticDirectory(u, invokers));
            } else {
                invoker = CLUSTER.join(new StaticDirectory(invokers));
            }
        }
    }
    // 利用動態代理，將 Invoker 轉換成本地介面代理
    return (T) PROXY_FACTORY.getProxy(invoker);
}
 

上面轉換的過程中，主要可概括為：先分為本地引用和遠端引用兩類。本地就是以 inJvm 協議的獲取本地服務，這不做過多說明；遠端引用分為直連服務和通過註冊中心。註冊中心分為單註冊中心和多註冊中心的情況，單註冊中心好解決，直接使用即可，多註冊中心時，將轉換後的 Invoker 合併成一個 Invoker。最後通過動態代理將 Invoker 轉換成本地介面代理。

獲取 Invoker 例項

由於本地服務時直接從快取中獲取，這裡就註冊中心的消費進行分析，上面程式碼片段中使用的是REF_PROTOCOL.refer進行轉換，該方法程式碼：

public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
    // 獲取服務的註冊中心url，裡面會設定註冊中心的協議和移除 registry 的引數
    url = getRegistryUrl(url);
    // 獲取註冊中心例項
    Registry registry = registryFactory.getRegistry(url);
    if (RegistryService.class.equals(type)) {
        return proxyFactory.getInvoker((T) registry, type, url);
    }

    // 獲取服務消費元資料
    Map<String, String> qs = StringUtils.parseQueryString(url.getParameterAndDecoded(REFER_KEY));
    // 從服務消費元資料中獲取分組資訊
    String group = qs.get(GROUP_KEY);
    if (group != null && group.length() > 0) {
        if ((COMMA_SPLIT_PATTERN.split(group)).length > 1 || "*".equals(group)) {
            // 執行 Invoker 轉換工作
            return doRefer(getMergeableCluster(), registry, type, url);
        }
    }
    // 執行 Invoker 轉換工作
    return doRefer(cluster, registry, type, url);
}
 

上面主要是獲取服務消費的註冊中心例項和進行服務分組，最後呼叫doRefer方法進行轉換工作，以下為doRefer的程式碼：

private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) {
    // 建立 RegistryDirectory 物件
    RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url);
    // 設定註冊中心
    directory.setRegistry(registry);
    // 設定協議
    directory.setProtocol(protocol);
    // directory.getUrl().getParameters() 是服務消費元資料
    Map<String, String> parameters = new HashMap<String, String>(directory.getUrl().getParameters());
    URL subscribeUrl = new URL(CONSUMER_PROTOCOL, parameters.remove(REGISTER_IP_KEY), 0, type.getName(), parameters);
    if (!ANY_VALUE.equals(url.getServiceInterface()) && url.getParameter(REGISTER_KEY, true)) {
        directory.setRegisteredConsumerUrl(getRegisteredConsumerUrl(subscribeUrl, url));
        // 消費訊息註冊到註冊中心
        registry.register(directory.getRegisteredConsumerUrl());
    }

    directory.buildRouterChain(subscribeUrl);
    // 服務消費者訂閱：服務提供端，動態配置，路由的通知
    directory.subscribe(subscribeUrl.addParameter(CATEGORY_KEY,
            PROVIDERS_CATEGORY + "," + CONFIGURATORS_CATEGORY + "," + ROUTERS_CATEGORY));

    // 多個Invoker合併為一個
    Invoker invoker = cluster.join(directory);
    return invoker;
}

上面實現主要是完成建立 RegistryDirectory 物件，將消費服務元資料註冊到註冊中心，通過 RegistryDirectory 物件裡的資訊，實現服務提供端，動態配置及路由的訂閱相關功能。

RegistryDirectory 這個類實現了 NotifyListener 這個通知監聽介面，當訂閱的服務，配置或路由發生變化時，會接收到通知，進行相應改變：

public synchronized void notify(List<URL> urls) {
    // 將服務提供方配置，路由配置，服務提供方的服務分別以不同的 key 儲存在 Map 中
    Map<String, List<URL>> categoryUrls = urls.stream()
            .filter(Objects::nonNull)
            .filter(this::isValidCategory)
            .filter(this::isNotCompatibleFor26x)
            .collect(Collectors.groupingBy(url -> {
                if (UrlUtils.isConfigurator(url)) {
                    return CONFIGURATORS_CATEGORY;
                } else if (UrlUtils.isRoute(url)) {
                    return ROUTERS_CATEGORY;
                } else if (UrlUtils.isProvider(url)) {
                    return PROVIDERS_CATEGORY;
                }
                return "";
            }));

    // 更新服務提供方配置
    List<URL> configuratorURLs = categoryUrls.getOrDefault(CONFIGURATORS_CATEGORY, Collections.emptyList());
    this.configurators = Configurator.toConfigurators(configuratorURLs).orElse(this.configurators);

    // 更新路由配置
    List<URL> routerURLs = categoryUrls.getOrDefault(ROUTERS_CATEGORY, Collections.emptyList());
    toRouters(routerURLs).ifPresent(this::addRouters);

    // 載入服務提供方的服務資訊
    List<URL> providerURLs = categoryUrls.getOrDefault(PROVIDERS_CATEGORY, Collections.emptyList());
    /**
     * 3.x added for extend URL address
     */
    ExtensionLoader<AddressListener> addressListenerExtensionLoader = ExtensionLoader.getExtensionLoader(AddressListener.class);
    List<AddressListener> supportedListeners = addressListenerExtensionLoader.getActivateExtension(getUrl(), (String[]) null);
    if (supportedListeners != null && !supportedListeners.isEmpty()) {
        for (AddressListener addressListener : supportedListeners) {
            providerURLs = addressListener.notify(providerURLs, getUrl(),this);
        }
    }
    // 重新載入 Invoker 例項
    refreshOverrideAndInvoker(providerURLs);
}

RegistryDirectory#notify裡面最後會重新整理 Invoker 進行重新載入，下面是核心程式碼的實現：

private void refreshOverrideAndInvoker(List<URL> urls) {
    // mock zookeeper://xxx?mock=return null
    overrideDirectoryUrl();
    // 重新整理 invoker 
    refreshInvoker(urls);
}

private void refreshInvoker(List<URL> invokerUrls) {
    Assert.notNull(invokerUrls, "invokerUrls should not be null");

    if (invokerUrls.size() == 1
            && invokerUrls.get(0) != null
            && EMPTY_PROTOCOL.equals(invokerUrls.get(0).getProtocol())) {

        ......

    } else {
        // 重新整理之前的 Invoker
        Map<String, Invoker<T>> oldUrlInvokerMap = this.urlInvokerMap; // local reference
        // 載入新的 Invoker Map
        Map<String, Invoker<T>> newUrlInvokerMap = toInvokers(invokerUrls);// Translate url list to Invoker map
        // 獲取新的 Invokers
        List<Invoker<T>> newInvokers = Collections.unmodifiableList(new ArrayList<>(newUrlInvokerMap.values()));
        // 快取新的 Invokers
        routerChain.setInvokers(newInvokers);
        this.invokers = multiGroup ? toMergeInvokerList(newInvokers) : newInvokers;
        this.urlInvokerMap = newUrlInvokerMap;

        try {
            // 通過新舊 Invokers 對比，銷燬無用的 Invokers
            destroyUnusedInvokers(oldUrlInvokerMap, newUrlInvokerMap); // Close the unused Invoker
        } catch (Exception e) {
            logger.warn("destroyUnusedInvokers error. ", e);
        }
    }
}

獲取重新整理前後的 Invokers，將新的 Invokers 重新快取起來，通過對比，銷燬無用的 Invoker。

上面將 URL 轉換 Invoker 是在RegistryDirectory#toInvokers中進行。

private Map<String, Invoker<T>> toInvokers(List<URL> urls) {
    Map<String, Invoker<T>> newUrlInvokerMap = new HashMap<>();
   
    Set<String> keys = new HashSet<>();
    String queryProtocols = this.queryMap.get(PROTOCOL_KEY);
    for (URL providerUrl : urls) {

        // 過濾消費端不匹配的協議，及非法協議
        ......

        // 合併服務提供端配置資料
        URL url = mergeUrl(providerUrl);
        // 過濾重複的服務提供端配置資料
        String key = url.toFullString();
        if (keys.contains(key)) {
            continue;
        }
        keys.add(key);

        // 快取鍵是不與使用者端引數合併的url，無論使用者如何合併引數，如果伺服器url更改，則再次引用
        Map<String, Invoker<T>> localUrlInvokerMap = this.urlInvokerMap; // local reference
        Invoker<T> invoker = localUrlInvokerMap == null ? null : localUrlInvokerMap.get(key);
        
        // 快取無對應 invoker，再次呼叫 protocol#refer 是否有資料
        if (invoker == null) {
            try {
                boolean enabled = true;
                if (url.hasParameter(DISABLED_KEY)) {
                    enabled = !url.getParameter(DISABLED_KEY, false);
                } else {
                    enabled = url.getParameter(ENABLED_KEY, true);
                }
                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);
            }
            // 將新的 Invoker 快取起來
            if (invoker != null) { // Put new invoker in cache
                newUrlInvokerMap.put(key, invoker);
            }
        } else {
            // 快取裡有資料，則進行重新覆蓋
            newUrlInvokerMap.put(key, invoker);
        }
    }
    keys.clear();
    return newUrlInvokerMap;
}

總結

通過《Dubbo之服務暴露》和本文兩篇文章對 Dubbo 服務暴露和服務消費原理的瞭解。我們可以看到，不管是暴露還是消費，Dubbo 都是以 Invoker 為資料交換主體進行，通過對 Invoker 發起呼叫，實現一個遠端或本地的實現。

個人部落格： https://ytao.top
關注公眾號 【ytao】，更多原創好文

相關推薦