dubbo - 服務(wù)引用過程

開篇

這篇文章的目的是嘗試將dubbo的服務(wù)引用過程盡量描述的清楚些,主要核心在于注冊consumer,訂閱provider并生成reference的invoker對象。

另外,嘗試通過zookeeper訂閱過程回調(diào)描述清楚provider變更時consumer如何感知provider的增刪操作。


注冊中心類依賴圖

dubbo注冊中心類圖

說明:

  • 關(guān)注ZookeeperRegistry、FailbackRegistry、AbastractRegistry的類關(guān)系。
  • 關(guān)注上述三者的依賴關(guān)系由于理解consumer的注冊過程。


consumer初始化過程時序圖

consumer初始化時序圖

說明:

  • 1、consumer端在初始化bean后的afterPropertiesSet()方法中進行初始化動作。

  • 2、consumer注冊自身到zk的/dubbo/xxx_interface/consumer節(jié)點下面

  • 3、consumer訂閱zk的/dubbo/xxx_interface/provider|configor|routers服務(wù)

  • 4、訂閱節(jié)點有變化的時候通知變更

  • 5、所有訂閱的服務(wù)保存在RegistyDirectory目錄中供發(fā)起RPC調(diào)用的時候cluster調(diào)用


整體流程圖總結(jié)

整體流程圖總結(jié)

說明:

該部分以Spring配置及ReferenceBean為入口,主要在ReferenceConfig中進行。

ReferenceConfig依賴RegistryProtocol完成了 "服務(wù)引用者注冊"、"服務(wù)提供者訂閱"和"Invoker創(chuàng)建" 的工作;

ReferenceConfig依賴JavassistProxyFactory完成了 "代理對象生成" 的工作;

2、注冊中心訂閱 & Invoker生成與獲取

該部分主要由RegistryDirectory和FailfastCluster實現(xiàn)。

通過ReferenceConfig調(diào)用RegistryDirectory的subscribe方法,觸發(fā)了對服務(wù)提供者url的訂閱及監(jiān)聽,在監(jiān)聽過程中RegistryDirectory借助DubboProtocol完成了Invoker的創(chuàng)建工作,并保存了服務(wù)引用url和Invoker的關(guān)系;

通過ReferenceConfig調(diào)用FailfastCluster的join方法,完成了對Invoker對象的獲??;

3、生成代理對象

該部分主要由JavassistProxyFactory完成。

以ReferenceConfig調(diào)用JavassistProxyFactory的getProxy方法為入口,傳入Invoker;

新創(chuàng)建了InvokerInvocationHandler,并使用dubbo自己的動態(tài)代理工具Proxy最終生成代理對象T ref;


dubbo 服務(wù)引用過程 - 階段一

說明:

  • 1、ReferenceBean的afterPropertiesSet()方法作為consumer的初始化入口。

  • 2、afterPropertiesSet()內(nèi)部調(diào)用getObject() -> get()執(zhí)行ReferenceConfig的get()方法。

  • 3、ReferenceConfig執(zhí)行g(shù)et() -> init() -> createProxy()進入consumer的代理創(chuàng)建過程。

  • 4、createProxy()方法內(nèi)部遍歷注冊中心生成consumer需要注冊的注冊中心URL地址。

  • 5、關(guān)注refprotocol.refer方法,進入consumer引用provider的邏輯。

public class ReferenceBean<T> extends ReferenceConfig<T> implements 
    FactoryBean, ApplicationContextAware, InitializingBean, DisposableBean {

    @SuppressWarnings({"unchecked"})
    public void afterPropertiesSet() throws Exception {
        // 省略相關(guān)代碼
        Boolean b = isInit();
        if (b == null && getConsumer() != null) {
            b = getConsumer().isInit();
        }
        if (b != null && b.booleanValue()) {
            // 獲取referenceBean入口
            getObject();
        }
    }

    public Object getObject() throws Exception {
        return get();
    }
}



public class ReferenceConfig<T> extends AbstractReferenceConfig {

    public synchronized T get() {

        if (ref == null) {
            init();
        }

        return ref;
    }


    private void init() {

        // 組裝參數(shù)
        Map<String, String> map = new HashMap<String, String>();
        Map<Object, Object> attributes = new HashMap<Object, Object>();
        map.put(Constants.SIDE_KEY, Constants.CONSUMER_SIDE);
        map.put(Constants.DUBBO_VERSION_KEY, Version.getVersion());
        map.put(Constants.TIMESTAMP_KEY, String.valueOf(System.currentTimeMillis()));
        if (ConfigUtils.getPid() > 0) {
            map.put(Constants.PID_KEY, String.valueOf(ConfigUtils.getPid()));
        }        

        // 創(chuàng)建服務(wù)引用的入口
        ref = createProxy(map);

        ConsumerModel consumerModel = new ConsumerModel(getUniqueServiceName(), this, ref, interfaceClass.getMethods());
        ApplicationModel.initConsumerModel(getUniqueServiceName(), consumerModel);
    }



    private T createProxy(Map<String, String> map) {
        if (isJvmRefer) {
            // 省略相關(guān)代碼
        } else {
            // user specified URL, could be peer-to-peer address, 
            // or register center's address. 
            // 處理直連的情況
            if (url != null && url.length() > 0) { 
                String[] us = Constants.SEMICOLON_SPLIT_PATTERN.split(url);
                if (us != null && us.length > 0) {
                    for (String u : us) {
                        URL url = URL.valueOf(u);
                        if (url.getPath() == null || url.getPath().length() == 0) {
                            url = url.setPath(interfaceName);
                        }
                        if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {
                            urls.add(url.addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map)));
                        } else {
                            urls.add(ClusterUtils.mergeUrl(url, map));
                        }
                    }
                }
            } else { 
                // assemble URL from register center's configuration
                // 加載配置的所有注冊中心,拼裝成urls
                List<URL> us = loadRegistries(false);
                if (us != null && us.size() > 0) {
                    for (URL u : us) {
                        URL monitorUrl = loadMonitor(u);
                        if (monitorUrl != null) {
                            map.put(Constants.MONITOR_KEY, URL.encode(monitorUrl.toFullString()));
                        }
                        // 這部分其實把注冊中心和reference的信息進行了合并,后面的url是合并信息
                        urls.add(u.addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map)));
                    }
                }
            }

            // 根據(jù)注冊中心的數(shù)量選擇走分支,這里一般情況走if分支。
            if (urls.size() == 1) {
                // 單注冊中心或者直連的服務(wù)
                invoker = refprotocol.refer(interfaceClass, urls.get(0));
            } else {
                // 多注冊中心,遍歷urls,調(diào)用refProtocol.refer創(chuàng)建遠程的動態(tài)代理Invoker
                List<Invoker<?>> invokers = new ArrayList<Invoker<?>>();
                URL registryURL = null;

                // 多注冊中心用最后一個注冊中心的地址
                for (URL url : urls) {
                    invokers.add(refprotocol.refer(interfaceClass, url));
                    if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {
                        registryURL = url; // use last registry url
                    }
                }

                // 多注冊中心需要通過cluster選擇一個
                if (registryURL != null) { // registry url is available
                    // use AvailableCluster only when register's cluster is available
                    URL u = registryURL.addParameter(Constants.CLUSTER_KEY, AvailableCluster.NAME);
                    invoker = cluster.join(new StaticDirectory(u, invokers));
                } else { // not a registry url
                    invoker = cluster.join(new StaticDirectory(invokers));
                }
            }
        }

        // 創(chuàng)建一個代理
        return (T) proxyFactory.getProxy(invoker);
    }
}


dubbo 服務(wù)引用過程 - 階段二

說明:

  • 1、RegistryProtocol內(nèi)部的doRefer()方法執(zhí)行consumer注冊和provider的訂閱。

  • 2、registry.register()負責(zé)注冊consumer到zookeeper的對應(yīng)節(jié)點路徑上。

  • 3、directory.subscribe()負責(zé)訂閱provider/configurator/routers等節(jié)點路徑信息。

public class RegistryProtocol implements Protocol {

    public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
        // 從url的registryKey獲取注冊中心類型:zookeeper
        url = url.setProtocol(url.getParameter(Constants.REGISTRY_KEY, Constants.DEFAULT_REGISTRY)).removeParameter(Constants.REGISTRY_KEY);
        // 從RegistryFactory獲取注冊器
        Registry registry = registryFactory.getRegistry(url);
        if (RegistryService.class.equals(type)) {
            return proxyFactory.getInvoker((T) registry, type, url);
        }

        // 省略相關(guān)代碼

        // 關(guān)注走doRefer這部分邏輯
        return doRefer(cluster, registry, type, url);
    }


    private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) {
        // 構(gòu)建RegistryDirectory,可以把它理解為注冊資源,其中包含了消費者/服務(wù)/路由等相關(guān)信息,其同時也是回調(diào)監(jiān)聽器
        RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url);
        directory.setRegistry(registry);
        directory.setProtocol(protocol);
        // all attributes of REFER_KEY
        Map<String, String> parameters = new HashMap<String, String>(directory.getUrl().getParameters());

        // 構(gòu)建subscribeUrl信息,主要拼接consumer:xxx的url地址
        URL subscribeUrl = new URL(Constants.CONSUMER_PROTOCOL, parameters.remove(Constants.REGISTER_IP_KEY), 0, type.getName(), parameters);

        if (!Constants.ANY_VALUE.equals(url.getServiceInterface())
                && url.getParameter(Constants.REGISTER_KEY, true)) {
            // 向注冊中心注冊服務(wù)消費者
            registry.register(subscribeUrl.addParameters(Constants.CATEGORY_KEY, Constants.CONSUMERS_CATEGORY,
                    Constants.CHECK_KEY, String.valueOf(false)));
        }

        // 從注冊中心訂閱服務(wù)提供者(即引用的服務(wù))
        directory.subscribe(subscribeUrl.addParameter(Constants.CATEGORY_KEY,
                Constants.PROVIDERS_CATEGORY
                        + "," + Constants.CONFIGURATORS_CATEGORY
                        + "," + Constants.ROUTERS_CATEGORY));

        // 從invoker當(dāng)中選擇其中一個返回
        Invoker invoker = cluster.join(directory);

        // 注冊消費者
        ProviderConsumerRegTable.registerConsuemr(invoker, url, subscribeUrl, directory);
        return invoker;
    }
}


dubbo 服務(wù)引用過程 - 階段三

說明:

  • 1、ZookeeperRegistry的實現(xiàn)consumer的對應(yīng)的節(jié)點的注冊。

  • 2、zkClient.create(toUrlPath(url))負責(zé)創(chuàng)建zookeeper節(jié)點信息。

public abstract class AbstractRegistry implements Registry {

    public void register(URL url) {
        registered.add(url);
    }
}


public abstract class FailbackRegistry extends AbstractRegistry {

    public void register(URL url) {
        if (destroyed.get()){
            return;
        }
        super.register(url);
        failedRegistered.remove(url);
        failedUnregistered.remove(url);
        try {
            // 注冊zookeeper節(jié)點上
            doRegister(url);
        } catch (Exception e) {
          // 省略相關(guān)代碼
        }
    }

    protected abstract void doRegister(URL url);
}


public class ZookeeperRegistry extends FailbackRegistry {

    protected void doRegister(URL url) {
        try {
            zkClient.create(toUrlPath(url), url.getParameter(Constants.DYNAMIC_KEY, true));
        } catch (Throwable e) {
           // 省略相關(guān)代碼
        }
    }
}


dubbo 服務(wù)引用過程 - 階段四

說明:

  • 1、RegistryDirectory完成consumer對provider端的zk節(jié)點的訂閱及回調(diào)處理。

  • 2、RegistryDirectory的subscribe()方法調(diào)用FailbackRegistry.subscribe()方法。

  • 3、FailbackRegistry.subscribe()內(nèi)部調(diào)用ZookeeperRegistry.doSubscribe()方法。

  • 4、ZookeeperRegistry.doSubscribe()方法內(nèi)部遍歷以下目錄挨個進行訂閱。
    /dubbo/com.alibaba.dubbo.demo.DemoService/providers
    /dubbo/com.alibaba.dubbo.demo.DemoService/configurators
    /dubbo/com.alibaba.dubbo.demo.DemoService/routers

  • 5、new ChildListener()負責(zé)創(chuàng)建zk監(jiān)聽的回調(diào)函數(shù),內(nèi)部childChanged負責(zé)執(zhí)行notify回調(diào)。

  • 6、zkClient.addChildListener()負責(zé)獲取path下所有節(jié)點信息,如providers目錄下所有的provider的列表,即provider的URL。

  • 7、FailbackRegistry.notify()方法用于針對providers/configurators/routers目錄下的urls進行回調(diào)操作,用于初始化對應(yīng)的類似invoker操作。

  • 8、childChanged()方法內(nèi)部的ZookeeperRegistry.this.notify()方法執(zhí)行監(jiān)聽節(jié)點變化并進行重新初始化。

  • 9、 FailbackRegistry.notify()內(nèi)部調(diào)用listener.notify()回調(diào)RegistryDirectory.notify()方法,listener指代RegistryDirectory對象實例。

public class RegistryDirectory<T> extends AbstractDirectory<T> implements NotifyListener {

    public void subscribe(URL url) {
        //todo 關(guān)心這個對象的notify回調(diào)函數(shù)
        setConsumerUrl(url);
        registry.subscribe(url, this);
    }
}


public abstract class FailbackRegistry extends AbstractRegistry {

    public void subscribe(URL url, NotifyListener listener) {
        if (destroyed.get()){
            return;
        }
        super.subscribe(url, listener);
        removeFailedSubscribed(url, listener);
        try {
            // Sending a subscription request to the server side
            doSubscribe(url, listener);
        } catch (Exception e) {
            // 省略相關(guān)代碼
        }
    }

    protected abstract void doSubscribe(URL url, NotifyListener listener);
}


public class ZookeeperRegistry extends FailbackRegistry {

    protected void doSubscribe(final URL url, final NotifyListener listener) {
        try {
            if (Constants.ANY_VALUE.equals(url.getServiceInterface())) {
                // 省略相關(guān)代碼
            } else {
                List<URL> urls = new ArrayList<URL>();
                // path可以取以下的值
                // /dubbo/com.alibaba.dubbo.demo.DemoService/providers
                // /dubbo/com.alibaba.dubbo.demo.DemoService/configurators
                // /dubbo/com.alibaba.dubbo.demo.DemoService/routers
                for (String path : toCategoriesPath(url)) {
                    ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.get(url);
                    if (listeners == null) {
                        zkListeners.putIfAbsent(url, new ConcurrentHashMap<NotifyListener, ChildListener>());
                        listeners = zkListeners.get(url);
                    }
                    ChildListener zkListener = listeners.get(listener);
                    if (zkListener == null) {
                        listeners.putIfAbsent(listener, new ChildListener() {
                            public void childChanged(String parentPath, List<String> currentChilds) {
                                // 內(nèi)部類訪問外部類ZookeeperRegistry實例調(diào)用回調(diào)notify方法
                                ZookeeperRegistry.this.notify(url, listener, toUrlsWithEmpty(url, parentPath, currentChilds));
                            }
                        });
                        zkListener = listeners.get(listener);
                    }
                    zkClient.create(path, false);
                    // 獲取path下的所有子節(jié)點并監(jiān)聽path路徑
                    List<String> children = zkClient.addChildListener(path, zkListener);
                    if (children != null) {
                        // 根據(jù)獲取path路徑下的子節(jié)點名稱,就是provider的URL路徑
                        urls.addAll(toUrlsWithEmpty(url, path, children));
                    }
                }
                // 通過回調(diào)將consumer:xxx的url路徑及對應(yīng)的
                // providers/configurators/routers的路徑進行回調(diào)
                notify(url, listener, urls);
            }
        } catch (Throwable e) {
           // 省略相關(guān)代碼
        }
    }
}


public abstract class FailbackRegistry extends AbstractRegistry {

    protected void notify(URL url, NotifyListener listener, List<URL> urls) {
        if (url == null) {
            throw new IllegalArgumentException("notify url == null");
        }
        if (listener == null) {
            throw new IllegalArgumentException("notify listener == null");
        }
        try {
            doNotify(url, listener, urls);
        } catch (Exception t) {
            // 省略相關(guān)代碼
        }
    }

    protected void doNotify(URL url, NotifyListener listener, List<URL> urls) {
        super.notify(url, listener, urls);
    }
}


public abstract class AbstractRegistry implements Registry {

    protected void notify(URL url, NotifyListener listener, List<URL> urls) {

        Map<String, List<URL>> result = new HashMap<String, List<URL>>();
        for (URL u : urls) {
            if (UrlUtils.isMatch(url, u)) {
                String category = u.getParameter(Constants.CATEGORY_KEY, Constants.DEFAULT_CATEGORY);
                List<URL> categoryList = result.get(category);
                if (categoryList == null) {
                    categoryList = new ArrayList<URL>();
                    result.put(category, categoryList);
                }
                categoryList.add(u);
            }
        }

        if (result.size() == 0) {
            return;
        }

        Map<String, List<URL>> categoryNotified = notified.get(url);
        if (categoryNotified == null) {
            notified.putIfAbsent(url, new ConcurrentHashMap<String, List<URL>>());
            categoryNotified = notified.get(url);
        }

        // 遍歷providers/configurators/routers目錄下的所有url進行回調(diào)處理
        for (Map.Entry<String, List<URL>> entry : result.entrySet()) {
            String category = entry.getKey();
            List<URL> categoryList = entry.getValue();
            categoryNotified.put(category, categoryList);
            saveProperties(url);
            // listener是RegistryDirectory對象
            listener.notify(categoryList);
        }
    }
}


dubbo 服務(wù)引用過程 - 階段五

說明:

  • 1、RegistryDirectory.notify()方法調(diào)用refreshInvoker(invokerUrls)實現(xiàn)invoker的創(chuàng)建。

  • 2、refreshInvoker()方法內(nèi)部執(zhí)行兩個動作,分別創(chuàng)建新增的provider的invoker并下線刪除的provider的invoker。

  • 3、toInvokers()內(nèi)部通過protocol.refer()方法創(chuàng)建provider的引用的invoker。

  • 4、destroyUnusedInvokers()內(nèi)部負責(zé)清理下線的provider的invoker。

public class RegistryDirectory<T> extends AbstractDirectory<T> implements NotifyListener {

    public synchronized void notify(List<URL> urls) {
        //todo step1 根據(jù)類別將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
        // merge override parameters
        this.overrideDirectoryUrl = directoryUrl;
        if (localConfigurators != null && localConfigurators.size() > 0) {
            for (Configurator configurator : localConfigurators) {
                this.overrideDirectoryUrl = configurator.configure(overrideDirectoryUrl);
            }
        }
        // todo providers  內(nèi)部的protocol.refer根據(jù)url創(chuàng)建遠程代理Invoker
        refreshInvoker(invokerUrls);
    }


    private void refreshInvoker(List<URL> invokerUrls) {
        if (invokerUrls != null && invokerUrls.size() == 1 && invokerUrls.get(0) != null
                && Constants.EMPTY_PROTOCOL.equals(invokerUrls.get(0).getProtocol())) {
            // 省略相關(guān)代碼
        } else {
            this.forbidden = false; // Allow to access
            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);//Cached invoker urls, convenient for comparison
            }
            if (invokerUrls.size() == 0) {
                return;
            }
            //todo 根據(jù)url創(chuàng)建遠程代理Invoker
            Map<String, Invoker<T>> newUrlInvokerMap = toInvokers(invokerUrls);// Translate url list to Invoker map
            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;
            }
            this.methodInvokerMap = multiGroup ? toMergeMethodInvokerMap(newMethodInvokerMap) : newMethodInvokerMap;
            this.urlInvokerMap = newUrlInvokerMap;
            try {
                destroyUnusedInvokers(oldUrlInvokerMap, newUrlInvokerMap); // Close the unused Invoker
            } catch (Exception e) {
                logger.warn("destroyUnusedInvokers error. ", e);
            }
        }
    }


    private Map<String, Invoker<T>> toInvokers(List<URL> urls) {
        Map<String, Invoker<T>> newUrlInvokerMap = new HashMap<String, Invoker<T>>();
        
            String key = url.toFullString(); // The parameter urls are sorted
            if (keys.contains(key)) { // Repeated url
                continue;
            }
            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
            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 {
                    // protocol.refer進行invoker創(chuàng)建的流程,待后續(xù)進行分析。
                    if (enabled) {
                        invoker = new InvokerDelegate<T>(protocol.refer(serviceType, url), url, providerUrl);
                    }
                } catch (Throwable t) {
                }
                if (invoker != null) { // Put new invoker in cache
                    newUrlInvokerMap.put(key, invoker);
                }
            } else {
                newUrlInvokerMap.put(key, invoker);
            }
        }
        keys.clear();
        return newUrlInvokerMap;
    }


    private void destroyUnusedInvokers(Map<String, Invoker<T>> oldUrlInvokerMap, Map<String, Invoker<T>> newUrlInvokerMap) {
        if (newUrlInvokerMap == null || newUrlInvokerMap.size() == 0) {
            destroyAllInvokers();
            return;
        }
        // check deleted invoker
        List<String> deleted = null;
        if (oldUrlInvokerMap != null) {
            Collection<Invoker<T>> newInvokers = newUrlInvokerMap.values();
            for (Map.Entry<String, Invoker<T>> entry : oldUrlInvokerMap.entrySet()) {
                if (!newInvokers.contains(entry.getValue())) {
                    if (deleted == null) {
                        deleted = new ArrayList<String>();
                    }
                    deleted.add(entry.getKey());
                }
            }
        }

        if (deleted != null) {
            for (String url : deleted) {
                if (url != null) {
                    Invoker<T> invoker = oldUrlInvokerMap.remove(url);
                    if (invoker != null) {
                        try {
                            invoker.destroy();
                            if (logger.isDebugEnabled()) {
                                logger.debug("destory invoker[" + invoker.getUrl() + "] success. ");
                            }
                        } catch (Exception e) {
                            logger.warn("destory invoker[" + invoker.getUrl() + "] faild. " + e.getMessage(), e);
                        }
                    }
                }
            }
        }
    }
}


dubbo 服務(wù)引用過程 - 階段六

說明:

  • 1、consumer端的RegistryDirectory維持了provider的invoker信息。
public class RegistryDirectory<T> extends AbstractDirectory<T> implements NotifyListener {

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

    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();
    private final String serviceKey; // Initialization at construction time, assertion not null
    private final Class<T> serviceType; // Initialization at construction time, assertion not null
    private final Map<String, String> queryMap; // Initialization at construction time, assertion not null
    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;

    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.
    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.
    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.
    private volatile Set<URL> cachedInvokerUrls; // The initial value is null and the midway may be assigned to null, please use the local variable reference
}
最后編輯于
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成,瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明:文章內(nèi)容(如有圖片或視頻亦包括在內(nèi))由作者上傳并發(fā)布,文章內(nèi)容僅代表作者本人觀點,簡書系信息發(fā)布平臺,僅提供信息存儲服務(wù)。

友情鏈接更多精彩內(nèi)容