Dubbo實踐（十四）生產者發布服務

Export發布服務流程

Dubbo協議向注冊中心發布服務：當服務提供方，向dubbo協議的注冊中心發布服務的時候，是如何獲取，創建注冊中心的，如何注冊以及訂閱服務的，下面我們來分析其流程。

看如下配置發布服務：

<!-- 指定了哪種的注冊中心，是基於zookeeper協議的，指定了注冊中心的地址以及端口號 -->
<dubbo:registry protocol="zookeeper" client="zkclient" address="localhost:2181" group="dubbo"/>
<bean id="demoService" class="com.alibaba.dubbo.demo.provider.DemoServiceImpl"/> 
<!-- 發布DemoService服務，服務的實現為DemoServiceImpl -->
<dubbo:service interface="com.alibaba.dubbo.demo.DemoService" ref="demoService"/> 

 

每個<dubbo:service/>在Spring內部都會生成一個ServiceBean實例，ServiceBean的實例化過程中調用export方法來暴露服務

1. 通過loadRegistries獲取注冊中心registryUrls

registry://localhost:2181/com.alibaba.dubbo.registry.RegistryService?application=demo-provider&client=zkclient&dubbo=2.6.1&group=dubbo&logger=slf4j&pid=8012&registry=zookeeper&timestamp=1532198612419

用統一數據模型URL表示：

• protocol=registry　　 　　 表示一個注冊中心URL
• 注冊中心地址　　　　　　localhost:2181
• 調用注冊中心的服務　　   RegistryService
• 注冊中心協議　　　　　   registry=zookeeper
• 。。。。。。

 2. 構建發布服務的URL

dubbo://192.168.56.1:20882/org.apache.dubbo.demo.DemoService?anyhost=true&application=demo-provider&buffer=8192&dispatcher=all&dubbo=2.6.1&executes=3&generic=false&interface=org.apache.dubbo.demo.DemoService&iothreads=2&loadbalance=leastactive&logger=slf4j&methods=sayHello&pid=8012&queues=300&retries=0&side=provider&threadpool=fixed&threads=10&timeout=300&timestamp=1532198751165

• 發布協議　　　　　　protocol =dubbo
• 服務提供者的地址　　192.168.56.1:20882
• 發布的服務　　　　com.alibaba.dubbo.demo.DemoService
• 。。。。。。

3. 遍歷registryUrls向注冊中心注冊服務

給每個registryUrl添加屬性key為export，value為上面的發布服務url得到如下registryUrl

registry://localhost:2181/com.alibaba.dubbo.registry.RegistryService?application=demo-provider&client=zkclient&dubbo=2.6.1&group=dubbo&logger=slf4j&pid=8012&registry=zookeeper&timestamp=1532198612419&export=dubbo%3a%2f%2f192.168.56.1%3a20882%2forg.apache.dubbo.demo.DemoService%3fanyhost%3dtrue%26application%3ddemo-provider%26buffer%3d8192%26dispatcher%3dall%26dubbo%3d2.6.1%26executes%3d3%26generic%3dfalse%26interface%3dorg.apache.dubbo.demo.DemoService%26iothreads%3d2%26loadbalance%3dleastactive%26logger%3dslf4j%26methods%3dsayHello%26pid%3d8012%26queues%3d300%26retries%3d0%26side%3dprovider%26threadpool%3dfixed%26threads%3d10%26timeout%3d300%26timestamp%3d1532198751165

 

4. 由發布的服務實例，服務接口以及registryUrl為參數，通過代理工廠proxyFactory獲取Invoker對象，Invoker對象是dubbo的核心模型，其他對象都向它靠攏或者轉換成它。

5. 通過Protocol對象暴露服務protocol.export(invoker)

• 通過DubboProtocol暴露服務的監聽;
• 通過RegistryProtocol將服務地址發布到注冊中心，並訂閱此服務。

以上邏輯，在ServiceConfig類中的doExportUrlsFor1Protocol實現：

        String scope = url.getParameter(Constants.SCOPE_KEY);
        // don't export when none is configured
        // 配置為none不暴露
        if (!Constants.SCOPE_NONE.toString().equalsIgnoreCase(scope)) {

            // export to local if the config is not remote (export to remote only when config is remote)
            // 配置不是remote的情況下做本地暴露 (配置為remote，則表示只暴露遠程服務)
            if (!Constants.SCOPE_REMOTE.toString().equalsIgnoreCase(scope)) {
                exportLocal(url);
            }
            // export to remote if the config is not local (export to local only when config is local)
            // 如果配置不是local則暴露為遠程服務.(配置為local，則表示只暴露遠程服務)
            if (!Constants.SCOPE_LOCAL.toString().equalsIgnoreCase(scope)) {
                if (logger.isInfoEnabled()) {
                    logger.info("Export dubbo service " + interfaceClass.getName() + " to url " + url);
                }
                if (registryURLs != null && !registryURLs.isEmpty()) {
                    for (URL registryURL : registryURLs) {
                        url = url.addParameterIfAbsent("dynamic", registryURL.getParameter("dynamic"));
                        URL monitorUrl = loadMonitor(registryURL);
                        if (monitorUrl != null) {
                            url = url.addParameterAndEncoded(Constants.MONITOR_KEY, monitorUrl.toFullString());
                        }
                        if (logger.isInfoEnabled()) {
                            logger.info("Register dubbo service " + interfaceClass.getName() + " url " + url + " to registry " + registryURL);
                        }
                        Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString()));
                        DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this);

                        Exporter<?> exporter = protocol.export(wrapperInvoker);
                        exporters.add(exporter);
                    }
                } else {
                    Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, url);
                    DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this);

                    Exporter<?> exporter = protocol.export(wrapperInvoker);
                    exporters.add(exporter);
                }
            }
        }
        this.urls.add(url);

 

RegistryProtocol.export(Invoker)暴露服務

META-INF/dubbo/internal/com.alibaba.dubbo.rpc.Protocol 文件中配置如下：

registry=com.alibaba.dubbo.registry.integration.RegistryProtocol

因為上一步中url為registry://開頭，因此使用的是RegistryProtocol類。

    public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException {
        //export invoker
        final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker);

        URL registryUrl = getRegistryUrl(originInvoker);

        //registry provider
        final Registry registry = getRegistry(originInvoker);
        final URL registedProviderUrl = getRegistedProviderUrl(originInvoker);

        //to judge to delay publish whether or not
        boolean register = registedProviderUrl.getParameter("register", true);

        ProviderConsumerRegTable.registerProvider(originInvoker, registryUrl, registedProviderUrl);

        if (register) {
            register(registryUrl, registedProviderUrl);
            ProviderConsumerRegTable.getProviderWrapper(originInvoker).setReg(true);
        }

        // Subscribe the override data
        // 訂閱override數據
        // FIXME When the provider subscribes, it will affect the scene : a certain JVM exposes the service and call the same service. Because the subscribed is cached key with the name of the service, it causes the subscription information to cover.
        // FIXME 提供者訂閱時，會影響同一JVM即暴露服務，又引用同一服務的的場景，因為subscribed以服務名為緩存的key，導致訂閱信息覆蓋。
        final URL overrideSubscribeUrl = getSubscribedOverrideUrl(registedProviderUrl);
        final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl, originInvoker);
        overrideListeners.put(overrideSubscribeUrl, overrideSubscribeListener);
        registry.subscribe(overrideSubscribeUrl, overrideSubscribeListener);
        //Ensure that a new exporter instance is returned every time export
        // 保證每次export都返回一個新的exporter實例
        return new DestroyableExporter<T>(exporter, originInvoker, overrideSubscribeUrl, registedProviderUrl);
    }

以上函數中，doLocalExport(originInvoker)就是調用正常的protocol的export過程，進行暴露。

這里我們針對demo的例子，在debug過程中記錄如下信息：

registedProviderUrl

dubbo://192.168.56.1:20882/com.alibaba.dubbo.demo.DemoService?anyhost=true&application=demo-provider&dubbo=2.0.0&generic=false&interface=com.alibaba.dubbo.demo.DemoService&loadbalance=roundrobin&methods=sayHello&pid=6324&side=provider&timestamp=1428237661384

 

overrideSubscribeUrl

provider://192.168.56.1:20882/com.alibaba.dubbo.demo.DemoService?anyhost=true&application=demo-provider&category=configurators&check=false&dubbo=2.0.0&generic=false&interface=com.alibaba.dubbo.demo.DemoService&loadbalance=roundrobin&methods=sayHello&pid=6324&side=provider&timestamp=1428237661384

 

注冊時應該是將registedProviderUrl傳遞到注冊中心，注冊中心記錄相應信息。這里我們可以理解為，消費者訪問注冊中心時，根據消費者需要獲得的服務去讀取服務提供者（url）。 而訂閱時，則是根據overrideSubscribeUrl地址和overrideSubscribeListener監聽。overrideSubscribeListener監聽的作用是當提供者的url改變時，重新export。