Dubbo源碼學習--服務發佈（ServiceBean、ServiceConfig）

時間 2019-11-19

標籤 dubbo 源碼學習服務發佈 servicebean serviceconfig 欄目 Dubbo 简体版

原文原文鏈接

前面講過Dubbo SPI拓展機制，經過ExtensionLoader實現可插拔加載拓展，本節將接着分析Dubbo的服務發佈過程。java

以源碼中dubbo-demo模塊做爲切入口一步步走進Dubbo源碼。在 dubbo-demo-provider模塊下配置文件 dubbo-demo-provider.xml中定義了服務提供方、註冊中心、協議及端口、服務接口等信息，以下：spring

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:dubbo="http://code.alibabatech.com/schema/dubbo"
       xmlns="http://www.springframework.org/schema/beans"
       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
    http://code.alibabatech.com/schema/dubbo http://code.alibabatech.com/schema/dubbo/dubbo.xsd">

    <!-- 提供方應用信息，用於計算依賴關係 -->
    <dubbo:application name="demo-provider"/>

    <!-- 使用multicast廣播註冊中心暴露服務地址 -->
    <!--<dubbo:registry address="multicast://224.5.6.7:1234"/>-->
    <dubbo:registry address="zookeeper://192.168.1.197:2181"/>

    <!-- 用dubbo協議在20880端口暴露服務 -->
    <dubbo:protocol name="dubbo" port="20880"/>

    <!-- 和本地bean同樣實現服務 -->
    <bean id="demoService" class="com.alibaba.dubbo.demo.provider.DemoServiceImpl"/>

    <!-- 聲明須要暴露的服務接口 -->
    <dubbo:service interface="com.alibaba.dubbo.demo.DemoService" ref="demoService"/>

</beans>

本人搭建了一個zookeeper服務器，將註冊中心由multicast更改成zookeeper。但是，在Spring中並無定義這些配置節點，配置文件中的內容如何自動加載到內存中、賦值給相應的對象呢？api

dubbo-demo-provider模塊com.alibaba.dubbo.demo.provider.Provider類中 main 方法做爲入口，進行斷點跟蹤。可發現，Dubbo配置文件加載是基於Spring可拓展Schema自定義實現的。Spring可拓展Schema須要實現以下幾步：服務器

定義配置文件相應Java Bean
編寫XSD文件
實現BeanDefinitionParser接口和繼承NamespaceHandlerSupport抽象類
編寫handlers和schemas文件，串聯各部分

關於Spring可拓展Schema機制，請自行Google瞭解。app

在 dubbo-config子模塊 dubbo-config-spring中定義了 spring.handlers 、 spring.schemas 、 dubbo.xsd、 DubboBeanDefinitionParser、 DubboNamespaceHandler文件，配置的Java Bean是在 dubbo-config-api模塊中定義的。spring.handlers配置定義了 Dubbo名空間的處理類，spring.schemas 定義了XSD文件的位置，DubboBeanDefinitionParser實現了BeanDefinitionParser接口，DubboNamespaceHandler繼承了NamespaceHandlerSupport抽象類。dom

spring.handlers文件內容以下：jvm

http\://code.alibabatech.com/schema/dubbo=com.alibaba.dubbo.config.spring.schema.DubboNamespaceHandler

spring.schemas 文件內容以下：ide

http\://code.alibabatech.com/schema/dubbo/dubbo.xsd=META-INF/dubbo.xsd

經過運用Spring Schema機制，實現對自定義配置文件參數注入後，會繼續執行InitializingBean接口的afterPropertiesSet() throws Exception方法。實例對象初始化完成後會執行事件監聽器ApplicationListener接口的void onApplicationEvent(E event)方法。在Dubbo中，ServiceBean類實現了ApplicationListener接口方法。ServiceBean類以下所示：學習

public class ServiceBean<T> extends ServiceConfig<T> implements InitializingBean, DisposableBean, ApplicationContextAware, ApplicationListener, BeanNameAware {
    ......
    public ServiceBean() {
        super();
        this.service = null;
    }

    public ServiceBean(Service service) {
        super(service);
        this.service = service;
    }
    ......

    public void onApplicationEvent(ApplicationEvent event) {
        if (ContextRefreshedEvent.class.getName().equals(event.getClass().getName())) {
            if (isDelay() && !isExported() && !isUnexported()) {
                if (logger.isInfoEnabled()) {
                    logger.info("The service ready on spring started. service: " + getInterface());
                }
                export();
            }
        }
    }

    ......

    @SuppressWarnings({"unchecked", "deprecation"})
    public void afterPropertiesSet() throws Exception {
        
        ......

        if (!isDelay()) {
            export();
        }
    }
}

實際源碼篇幅過長，這裏只截取關鍵的核心部分。經過上述源碼片斷，能夠看到ServiceBean實現了ApplicationListener接口、InitializingBean接口，同時也繼承了ServiceConfig類。在構造器方法中均調用了父類ServiceConfig的構造方法。this

在加載完自定義配置文件屬性後，會執行afterPropertiesSet方法，根據配置文件中delay屬性判斷是否當即執行export方法。delay屬性是用於標識是否延遲暴露服務，Dubbo中默認延遲暴露。若服務延遲暴露，則繼續初始化示例對象，待對象初始化完成後，執行onApplicationEvent方法，此時會執行export()。

export()是在父類ServiceConfig中定義的，是一個同步方法。進入export()纔是真正的開啓服務發佈之旅。export()源碼以下：

public synchronized void export() {
        if (provider != null) {
            if (export == null) {
                export = provider.getExport();
            }
            if (delay == null) {
                delay = provider.getDelay();
            }
        }
        if (export != null && !export) {
            return;
        }

        if (delay != null && delay > 0) {
            delayExportExecutor.schedule(new Runnable() {
                public void run() {
                    doExport();
                }
            }, delay, TimeUnit.MILLISECONDS);
        } else {
            doExport();
        }
    }

export()方法經過關鍵字synchronized實現同步處理。而且實現了當即發佈和延遲發佈，並經過定時器來實現延遲發佈，延遲發佈時間單位是分鐘。

具體的服務發佈過程交給doExport()處理，doExport()也是一個同步方法，經過synchronized關鍵字實現。追蹤源碼，發現doExport()僅僅是作了服務發佈的前期準備工做，實際的發佈工做交給doExportUrls()方法來完成。doExportUrls()方法源碼以下：

@SuppressWarnings({"unchecked", "rawtypes"})
    private void doExportUrls() {
        List<URL> registryURLs = loadRegistries(true);
        for (ProtocolConfig protocolConfig : protocols) {
            doExportUrlsFor1Protocol(protocolConfig, registryURLs);
        }
    }

源碼中，經過loadRegistries(boolean provider)方法將全部服務URL封裝爲List，獲得服務提供者集合registryURLs。而後，經過for方法遍歷，調用doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List<URL> registryURLs)方法來實現對服務的發佈。doExportUrlsFor1Protocol方法源碼以下：

private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List<URL> registryURLs) {
        String name = protocolConfig.getName();
        if (name == null || name.length() == 0) {
            name = "dubbo";
        }

        Map<String, String> map = new HashMap<String, String>();
        map.put(Constants.SIDE_KEY, Constants.PROVIDER_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()));
        }
        appendParameters(map, application);
        appendParameters(map, module);
        appendParameters(map, provider, Constants.DEFAULT_KEY);
        appendParameters(map, protocolConfig);
        appendParameters(map, this);

        ......

        if (ProtocolUtils.isGeneric(generic)) {
            map.put("generic", generic);
            map.put("methods", Constants.ANY_VALUE);
        } else {
            String revision = Version.getVersion(interfaceClass, version);
            if (revision != null && revision.length() > 0) {
                map.put("revision", revision);
            }

            String[] methods = Wrapper.getWrapper(interfaceClass).getMethodNames();
            if (methods.length == 0) {
                logger.warn("NO method found in service interface " + interfaceClass.getName());
                map.put("methods", Constants.ANY_VALUE);
            } else {
                map.put("methods", StringUtils.join(new HashSet<String>(Arrays.asList(methods)), ","));
            }
        }
        if (!ConfigUtils.isEmpty(token)) {
            if (ConfigUtils.isDefault(token)) {
                map.put("token", UUID.randomUUID().toString());
            } else {
                map.put("token", token);
            }
        }
        if ("injvm".equals(protocolConfig.getName())) {
            protocolConfig.setRegister(false);
            map.put("notify", "false");
        }
        // 導出服務
        String contextPath = protocolConfig.getContextpath();
        if ((contextPath == null || contextPath.length() == 0) && provider != null) {
            contextPath = provider.getContextpath();
        }

        //獲取註冊監聽地址和端口
        String host = this.findConfigedHosts(protocolConfig, registryURLs, map);
        Integer port = this.findConfigedPorts(protocolConfig, name, map);

        //根據以前收集的map數據和地址端口，組裝URL
        URL url = new URL(name, host, port, (contextPath == null || contextPath.length() == 0 ? "" : contextPath + "/") + path, map);

        if (ExtensionLoader.getExtensionLoader(ConfiguratorFactory.class)
                .hasExtension(url.getProtocol())) {
            url = ExtensionLoader.getExtensionLoader(ConfiguratorFactory.class)
                    .getExtension(url.getProtocol()).getConfigurator(url).configure(url);
        }

        String scope = url.getParameter(Constants.SCOPE_KEY);
        //配置爲none不暴露
        if (!Constants.SCOPE_NONE.toString().equalsIgnoreCase(scope)) {

            //配置不是remote的狀況下作本地暴露 (配置爲remote，則表示只暴露遠程服務)
            if (!Constants.SCOPE_REMOTE.toString().equalsIgnoreCase(scope)) {
                exportLocal(url);
            }
            //若是配置不是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.size() > 0
                        && url.getParameter("register", true)) {
                    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()));

                        Exporter<?> exporter = protocol.export(invoker);
                        exporters.add(exporter);
                    }
                } else {
                    //生成代理對象，invoker可看做服務的代理或封裝
                    Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, url);

                    Exporter<?> exporter = protocol.export(invoker);    //此時加載的protocol爲DubboProtocol對象
                    exporters.add(exporter);
                }
            }
        }
        this.urls.add(url);
    }

    @SuppressWarnings({"unchecked", "rawtypes"})
    private void exportLocal(URL url) {
        if (!Constants.LOCAL_PROTOCOL.equalsIgnoreCase(url.getProtocol())) {
            URL local = URL.valueOf(url.toFullString())
                    .setProtocol(Constants.LOCAL_PROTOCOL)
                    .setHost(LOCALHOST)
                    .setPort(0);
            Exporter<?> exporter = protocol.export(
                    proxyFactory.getInvoker(ref, (Class) interfaceClass, local));
            exporters.add(exporter);
            logger.info("Export dubbo service " + interfaceClass.getName() + " to local registry");
        }
    }

方法大致流程以下：

map裝配參數
利用map中的參數構建URL，爲暴露服務作準備。此URL爲Dubbo自定義類型，是final類，實現了Serializable接口
根據範圍選擇是暴露本地服務，仍是暴露遠程服務
根據代理工廠生成服務代理invoker
根據配置的協議，暴露服務代理

值得注意的是：在判斷是否暴露本地服務和遠程服務時，有個簡單的邏輯值得學習，用得很漂亮，使得代碼變得很簡潔。提取核心部分簡化以下：

//配置不是remote的狀況下作本地暴露 (配置爲remote，則表示只暴露遠程服務)
            if (!Constants.SCOPE_REMOTE.toString().equalsIgnoreCase(scope)) {
                exportLocal(url);
            }
            //若是配置不是local則暴露爲遠程服務.(配置爲local，則表示只暴露本地服務)
            if (!Constants.SCOPE_LOCAL.toString().equalsIgnoreCase(scope)) {
                ......
                //exportRemote(url)
                ......
            }

默認狀況下，當scope爲null的時候，會同時暴露本地服務和遠程服務。這個小巧的邏輯技巧值得學習！

通過以上分析：能夠大致瞭解Dubbo RPC服務的發佈過程，可是在整個流程中具體是如何產生服務代理的呢？請聽下回分解：Dubbo RPC服務的發佈之服務代理