Dubbo源碼分析（二）Dubbo服務發布Export

本文轉載自查看原文 2018-11-06 15:16 1009 Dubbo源碼/ Dubbo

服務發布

打印的日志

[INFO ]  com.alibaba.dubbo.config.AbstractConfig {ServiceBean.java:107} -  [DUBBO] The service ready on spring started. service: com.alibaba.dubbo.demo.DemoService, dubbo version: 2.0.0, current host: 127.0.0.1
[INFO ]  com.alibaba.dubbo.config.AbstractConfig {ServiceConfig.java:575} -  [DUBBO] Export dubbo service com.alibaba.dubbo.demo.DemoService to local registry, dubbo version: 2.0.0, current host: 127.0.0.1
[INFO ]  com.alibaba.dubbo.config.AbstractConfig {ServiceConfig.java:535} -  [DUBBO] Export dubbo service com.alibaba.dubbo.demo.DemoService to url dubbo://192.168.31.132:20880/com.alibaba.dubbo.demo.DemoService?anyhost=true&application=demo-provider&dubbo=2.0.0&generic=false&interface=com.alibaba.dubbo.demo.DemoService&methods=sayHello&pid=948&side=provider&timestamp=1538476342717, dubbo version: 2.0.0, current host: 127.0.0.1
[INFO ]  com.alibaba.dubbo.config.AbstractConfig {ServiceConfig.java:546} -  [DUBBO] Register dubbo service com.alibaba.dubbo.demo.DemoService url dubbo://192.168.31.132:20880/com.alibaba.dubbo.demo.DemoService?anyhost=true&application=demo-provider&dubbo=2.0.0&generic=false&interface=com.alibaba.dubbo.demo.DemoService&methods=sayHello&pid=1173&side=provider&timestamp=1538490950321 to registry registry://localhost:2181/com.alibaba.dubbo.registry.RegistryService?application=demo-provider&dubbo=2.0.0&pid=1173&registry=zookeeper&timestamp=1538490950209, dubbo version: 2.0.0, current host: 127.0.0.1
[INFO ]  com.alibaba.dubbo.remoting.transport.AbstractServer {AbstractServer.java:64} -  [DUBBO] Start NettyServer bind /0.0.0.0:20880, export /192.168.31.132:20880, dubbo version: 2.0.0, current host: 127.0.0.
[INFO ]  com.alibaba.dubbo.registry.zookeeper.ZookeeperRegistry {AbstractRegistry.java:223} -  [DUBBO] Load registry store file /Users/isz_pm/.dubbo/dubbo-registry-localhost.cache, data: {
[INFO ]  org.apache.zookeeper.ZooKeeper {ZooKeeper.java:438} - Initiating client connection, connectString=localhost:2181 sessionTimeout=30000 watcher=org.I0Itec.zkclient.ZkClient@26fc13bc

我們通過啟動provider，看下控制台的日志輸出，基本上可以看出Dubbo的服務發布的幾個步驟

Dubbo怎么和Spring融合

Spring 提供了可擴展的Schema，Dubbo是怎么擴展的呢，先看下Dubbo的配置文件：

<?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="zookeeper://localhost:2181"/>
    <!-- 用dubbo協議在20880端口暴露服務 -->
    <dubbo:protocol name="dubbo" port="20880"/>
    <!-- 聲明需要暴露的服務接口 -->
    <bean id="demoService" class="com.alibaba.dubbo.demo.provider.DemoServiceImpl"/>
    <!-- 和本地bean一樣實現服務 -->
    <dubbo:service interface="com.alibaba.dubbo.demo.DemoService" ref="demoService"/>
</beans>

我看可以看到http://code.alibabatech.com/schema/dubbo/dubbo.xsd，從源碼中我們找到：

要想擴展Schema，需要這五個文件，具體的作用就不說了，直接看DubboNamespaceHandler類

public class DubboNamespaceHandler extends NamespaceHandlerSupport {
    static {
        Version.checkDuplicate(DubboNamespaceHandler.class);
    }
    public void init() {
        registerBeanDefinitionParser("application", new DubboBeanDefinitionParser(ApplicationConfig.class, true));
        registerBeanDefinitionParser("module", new DubboBeanDefinitionParser(ModuleConfig.class, true));
        registerBeanDefinitionParser("registry", new DubboBeanDefinitionParser(RegistryConfig.class, true));
        registerBeanDefinitionParser("monitor", new DubboBeanDefinitionParser(MonitorConfig.class, true));
        registerBeanDefinitionParser("provider", new DubboBeanDefinitionParser(ProviderConfig.class, true));
        registerBeanDefinitionParser("consumer", new DubboBeanDefinitionParser(ConsumerConfig.class, true));
        registerBeanDefinitionParser("protocol", new DubboBeanDefinitionParser(ProtocolConfig.class, true));
        registerBeanDefinitionParser("service", new DubboBeanDefinitionParser(ServiceBean.class, true));
        registerBeanDefinitionParser("reference", new DubboBeanDefinitionParser(ReferenceBean.class, false));
        registerBeanDefinitionParser("annotation", new DubboBeanDefinitionParser(AnnotationBean.class, true));
    }
}

這里面初始化了Dubbo配置文件，剩下的就不多說了，直接看ServiceBean中Dubbo是怎么發布服務的

暴露本地服務

我們先看ServiceBean，這個是Dubbo服務發布的入口代碼

public class ServiceBean<T> extends ServiceConfig<T> implements InitializingBean, DisposableBean, ApplicationContextAware, ApplicationListener, BeanNameAware {

這個類實現了很多接口，其中的ApplicationListener接口,實現了onApplicationEvent()事件方法，

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();
            }
        }
    }

進入export()方法

public synchronized void export() {
        ···
        if (delay != null && delay > 0) {
            Thread thread = new Thread(new Runnable() {
                public void run() {
                    try {
                        //1.如果需要延遲發布，直接sleep
                        Thread.sleep(delay);
                    } catch (Throwable e) {
                    }
                    doExport();
                }
            });
            thread.setDaemon(true);
            thread.setName("DelayExportServiceThread");
            thread.start();
        } else {
            //調用發布方法
            doExport();
        }
    }

進入doExport()方法，繼續各種check，進入doExportUrls()方法

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

這里registryURL是啥，看debug結果：

loadRegistries()方法中返回了一個URL的List，我們看下這個URL是以registry開頭的，localhost:2181是我們配置在xml中<dubbo:registry address="zookeeper://localhost:2181"/>的這段代碼中獲取的，URL后面都是需要的數據，有沒有理解上一篇中說的一句話，Dubbo是基於URL驅動的
那什么會有多個URL呢，因為我們在xml中可以配置多個dubbo:registry，可以把服務發布到多個注冊中心
調用doExportUrlsFor1Protocol()方法，此時protocolConfig是<dubbo:protocol name="dubbo" port="20880"/>

private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List<URL> registryURLs) {
        ···
        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 = proxyFactory.getInvoker(ref, (Class) interfaceClass, url);

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

先看服務本地發布exportLocal()方法
此時的url=dubbo://192.168.31.132:20880/com.alibaba.dubbo.demo.DemoService?anyhost=true&application=demo-provider&dubbo=2.0.0&generic=false&interface=com.alibaba.dubbo.demo.DemoService&methods=sayHello&pid=843&side=provider&timestamp=1538464205631

private void exportLocal(URL url) {
        if (!Constants.LOCAL_PROTOCOL.equalsIgnoreCase(url.getProtocol())) {
            URL local = URL.valueOf(url.toFullString())
                    .setProtocol(Constants.LOCAL_PROTOCOL)
                    .setHost(NetUtils.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");
        }
    }

繼續組裝url，local=injvm://127.0.0.1/com.alibaba.dubbo.demo.DemoService?anyhost=true&application=demo-provider&dubbo=2.0.0&generic=false&interface=com.alibaba.dubbo.demo.DemoService&methods=sayHello&pid=843&side=provider&timestamp=1538464205631
執行proxyFactory.getInvoker(ref, (Class) interfaceClass, local)，這個時候的proxyFactory是什么？

private static final ProxyFactory proxyFactory = ExtensionLoader.getExtensionLoader(ProxyFactory.class).getAdaptiveExtension();

這個時候proxyFactory是一個動態適配器代理類ProxyFactory$Adpative，我們把這個代理類拿出來

package com.alibaba.dubbo.rpc;
import com.alibaba.dubbo.common.extension.ExtensionLoader;
public class ProxyFactory$Adpative implements com.alibaba.dubbo.rpc.ProxyFactory {
    public com.alibaba.dubbo.rpc.Invoker getInvoker(java.lang.Object arg0, java.lang.Class arg1, com.alibaba.dubbo.common.URL arg2) throws com.alibaba.dubbo.rpc.RpcException {
        if (arg2 == null) throw new IllegalArgumentException("url == null");
        com.alibaba.dubbo.common.URL url = arg2;
        String extName = url.getParameter("proxy", "javassist");
        if(extName == null) throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.rpc.ProxyFactory) name from url(" + url.toString() + ") use keys([proxy])");
        com.alibaba.dubbo.rpc.ProxyFactory extension = (com.alibaba.dubbo.rpc.ProxyFactory)ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.rpc.ProxyFactory.class).getExtension(extName);
        return extension.getInvoker(arg0, arg1, arg2);
    }
    public java.lang.Object getProxy(com.alibaba.dubbo.rpc.Invoker arg0) throws com.alibaba.dubbo.rpc.RpcException {
        if (arg0 == null) throw new IllegalArgumentException("com.alibaba.dubbo.rpc.Invoker argument == null");
        if (arg0.getUrl() == null) throw new IllegalArgumentException("com.alibaba.dubbo.rpc.Invoker argument getUrl() == null");com.alibaba.dubbo.common.URL url = arg0.getUrl();
        String extName = url.getParameter("proxy", "javassist");
        if(extName == null) throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.rpc.ProxyFactory) name from url(" + url.toString() + ") use keys([proxy])");
        com.alibaba.dubbo.rpc.ProxyFactory extension = (com.alibaba.dubbo.rpc.ProxyFactory)ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.rpc.ProxyFactory.class).getExtension(extName);
        return extension.getProxy(arg0);
    }
}

這個時候執行proxyFactory.getInvoker(ref, (Class) interfaceClass, local)，這個時候的local是injvm開頭，所以在getInvoke方法中，extName=javassist,同理執行getExtension("javassist")方法，這個時候應該返回什么呢？是JavassistProxyFactory嗎？不是，因為

我們看到，這個時候有一個包裝類，所以這個時候返回的是StubProxyFactoryWrapper，所以這個時候執行的是StubProxyFactoryWrapper.getInvoker方法

public <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) throws RpcException {
        return proxyFactory.getInvoker(proxy, type, url);
}

先理一下這個方法的三個參數，proxy是接口實現類的引用，type是接口，url是injvm開頭的
這個時候的proxyFactory是什么？我們知道Dubbo的IOC幫我們注入了包裝類需要的實例參數，所以這個時候proxyFactory是JavassistProxyFactory，

public <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) {
        // TODO Wrapper類不能正確處理帶$的類名
        final Wrapper wrapper = Wrapper.getWrapper(proxy.getClass().getName().indexOf('$') < 0 ? proxy.getClass() : type);
        return new AbstractProxyInvoker<T>(proxy, type, url) {
            @Override
            protected Object doInvoke(T proxy, String methodName,
                                      Class<?>[] parameterTypes,
                                      Object[] arguments) throws Throwable {
                return wrapper.invokeMethod(proxy, methodName, parameterTypes, arguments);
            }
        };
    }

先看下Wrapper wrapper = Wrapper.getWrapper()方法，傳入的參數是proxy.getClass()就是我們的實現類DemoServiveImpl的Class，

public static Wrapper getWrapper(Class<?> c) {
        ···
            ret = makeWrapper(c);
            WRAPPER_MAP.put(c, ret);
        }
        return ret;
    }

這個里面的makeWrapper方法就是創建一個動態代理類，我們看下這個方法的部分內容：

private static Wrapper makeWrapper(Class<?> c) {
    ···
        StringBuilder c1 = new StringBuilder("public void setPropertyValue(Object o, String n, Object v){ ");
        StringBuilder c2 = new StringBuilder("public Object getPropertyValue(Object o, String n){ ");
        StringBuilder c3 = new StringBuilder("public Object invokeMethod(Object o, String n, Class[] p, Object[] v) throws " + InvocationTargetException.class.getName() + "{ ");
    ···
    return (Wrapper) wc.newInstance();
    ···

這里生成了invokeMethod方法，最終把這個對象實例化，我們猜想一下，Dubbo的消費端來調用服務端的接口的時候，是不是通過這個代理對象中的invokerMethod方法最終去執行接口實現類的代碼呢？
我們看到 new AbstractProxyInvoker (proxy, type, url)中重寫了一個doInvoke方法，這個doInvoke方法中調用了wrapper.invokeMethod()，這個wrapper就是剛剛生成的wrapper
接着回到ServiceConfig.exportLocal()方法中
執行Exporter<?> exporter = protocol.export(invoker);這個時候protocol是什么？貼代碼

private static final Protocol protocol = ExtensionLoader.getExtensionLoader(Protocol.class).getAdaptiveExtension();

這個時候protocol=Protocol$Adpative，執行Protocol$Adpative.export方法
Protocol extension = (Protocol)ExtensionLoader.getExtensionLoader(Protocol.class).getExtension("injvm")，
這個時候得到的extension什么？是ProtocolListenrWrapper(ProtocolFiterWrapper(InjvmProtocol)),這樣的一個wrapper對象,執行export方法，這里的Fiter會生成8個Fiter鏈，看下InjvmProtocol中的export方法，目的是exporterMap.put(key, this) 此時，key=com.alibaba.dubbo.demo.DemoService, this=InjvmExporter

public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
       return new InjvmExporter<T>(invoker, invoker.getUrl().getServiceKey(), exporterMap);
   }
   InjvmExporter(Invoker<T> invoker, String key, Map<String, Exporter<?>> exporterMap) {
       super(invoker);
       this.key = key;
       this.exporterMap = exporterMap;
       exporterMap.put(key, this);
   }

最終得到一個ListenerExportWrapper,放到exporters中,到此，本地服務發布完成

[INFO ]  com.alibaba.dubbo.config.AbstractConfig {ServiceConfig.java:575} -  [DUBBO] Export dubbo service com.alibaba.dubbo.demo.DemoService to local registry, dubbo version: 2.0.0, current host: 127.0.0.1

暴露遠程服務

從這段代碼開始讀

Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString()));

registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString())，這個返回一個URL=registry://localhost:2181/com.alibaba.dubbo.registry.RegistryService?application=demo-provider&dubbo=2.0.0&export=dubbo%3A%2F%2F192.168.31.132%3A20880%2Fcom.alibaba.dubbo.demo.DemoService%3Fanyhost%3Dtrue%26application%3Ddemo-provider%26dubbo%3D2.0.0%26generic%3Dfalse%26interface%3Dcom.alibaba.dubbo.demo.DemoService%26methods%3DsayHello%26pid%3D948%26side%3Dprovider%26timestamp%3D1538476342717&pid=948&registry=zookeeper&timestamp=1538476342610
這個時候proxyFactory同樣是Protocol$Adpative，進入getInvoker方法：
ProxyFactory extension = ExtensionLoader.getExtensionLoader(ProxyFactory.class).getExtension("javassist");
這里extension=StubProxyFactoryWrapper包裝類，繼續執行getInvoker()方法，最終在JavassistProxyFactory的getInvoker中 new AbstractProxyInvoker (proxy, type, url)並返回，這個就是我們需要的invoker
繼續執行Exporter<?> exporter = protocol.export(invoker);
此時protocol=Protocol$Adaptive ,執行Protocol$Adaptive.export方法

Protocol extension = (Protocol)ExtensionLoader.getExtensionLoader(Protocol.class).getExtension("registry");
return extension.export(arg0);

同理extension=ProtocolListenrWrapper(ProtocolFiterWrapper(RegistryProtocol)),直接進入RegistryProtocol.export方法中：

public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException {
       //export invoker  先啟動本地監聽服務
       final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker);
       //registry provider
       final Registry registry = getRegistry(originInvoker);
       final URL registedProviderUrl = getRegistedProviderUrl(originInvoker);
       registry.register(registedProviderUrl);
       // 訂閱override數據
       // FIXME 提供者訂閱時，會影響同一JVM即暴露服務，又引用同一服務的的場景，因為subscribed以服務名為緩存的key，導致訂閱信息覆蓋。
       final URL overrideSubscribeUrl = getSubscribedOverrideUrl(registedProviderUrl);
       final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl);
       overrideListeners.put(overrideSubscribeUrl, overrideSubscribeListener);
       registry.subscribe(overrideSubscribeUrl, overrideSubscribeListener);
       //保證每次export都返回一個新的exporter實例
       return new Exporter<T>() {
           public Invoker<T> getInvoker() {
               return exporter.getInvoker();
           }

           public void unexport() {
               try {
                   exporter.unexport();
               } catch (Throwable t) {
                   logger.warn(t.getMessage(), t);
               }
               try {
                   registry.unregister(registedProviderUrl);
               } catch (Throwable t) {
                   logger.warn(t.getMessage(), t);
               }
               try {
                   overrideListeners.remove(overrideSubscribeUrl);
                   registry.unsubscribe(overrideSubscribeUrl, overrideSubscribeListener);
               } catch (Throwable t) {
                   logger.warn(t.getMessage(), t);
               }
           }
       };
   }

進入doLocalExport：

private <T> ExporterChangeableWrapper<T> doLocalExport(final Invoker<T> originInvoker) {
       String key = getCacheKey(originInvoker);
       ExporterChangeableWrapper<T> exporter = (ExporterChangeableWrapper<T>) bounds.get(key);
       if (exporter == null) {
           synchronized (bounds) {
               exporter = (ExporterChangeableWrapper<T>) bounds.get(key);
               if (exporter == null) {
                   final Invoker<?> invokerDelegete = new InvokerDelegete<T>(originInvoker, getProviderUrl(originInvoker));
                   exporter = new ExporterChangeableWrapper<T>((Exporter<T>) protocol.export(invokerDelegete), originInvoker);
                   bounds.put(key, exporter);
               }
           }
       }
       return (ExporterChangeableWrapper<T>) exporter;
   }

此時，我們看到(Exporter ) protocol.export(invokerDelegete)這段代碼，這個時候protocol是什么呢，是Protocol$Adaptive,什么時候時候賦值的呢，是加載擴展點的時候，有個injectExtension方法，依賴注入了protocol
繼續執行Protocol extension = (Protocol)ExtensionLoader.getExtensionLoader(Protocol.class).getExtension("dubbo"),這個extension是ProtocolListenrWrapper(ProtocolFiterWrapper(DubboProtocol))
執行DubboProtocol.export方法：

public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
        URL url = invoker.getUrl();
        // export service.
        String key = serviceKey(url);
        DubboExporter<T> exporter = new DubboExporter<T>(invoker, key, exporterMap);
        exporterMap.put(key, exporter);
        //export an stub service for dispaching event
        Boolean isStubSupportEvent = url.getParameter(Constants.STUB_EVENT_KEY, Constants.DEFAULT_STUB_EVENT);
        Boolean isCallbackservice = url.getParameter(Constants.IS_CALLBACK_SERVICE, false);
        if (isStubSupportEvent && !isCallbackservice) {
            String stubServiceMethods = url.getParameter(Constants.STUB_EVENT_METHODS_KEY);
            if (stubServiceMethods == null || stubServiceMethods.length() == 0) {
                if (logger.isWarnEnabled()) {
                    logger.warn(new IllegalStateException("consumer [" + url.getParameter(Constants.INTERFACE_KEY) +
                            "], has set stubproxy support event ,but no stub methods founded."));
                }
            } else {
                stubServiceMethodsMap.put(url.getServiceKey(), stubServiceMethods);
            }
        }

        openServer(url);

        return exporter;
    }

serviceKey(url), 這里我們得到一個key=com.alibaba.dubbo.demo.DemoService:20880，目的是要exporterMap.put(key, this)// key=com.alibaba.dubbo.demo.DemoService:20880, this=DubboExporter ，這里其中最重要的一點就是將invoker轉化為exporter，專題最后我們會分析一下所有的invoker

終於我們看到了一段代碼，openServer(url),千幸萬苦，我們要開始啟動Netty服務了，先休息一下，寫的好累啊

啟動Netty服務

private void openServer(URL url) {
        // find server.
        String key = url.getAddress();
        //client 也可以暴露一個只有server可以調用的服務。
        boolean isServer = url.getParameter(Constants.IS_SERVER_KEY, true);
        if (isServer) {
            ExchangeServer server = serverMap.get(key);
            if (server == null) {
                serverMap.put(key, createServer(url));
            } else {
                //server支持reset,配合override功能使用
                server.reset(url);
            }
        }
    }

首先得到一個key=ip:端口，再調用createServer(),創建服務,開啟心跳檢測，默認使用 netty。組裝 url

private ExchangeServer createServer(URL url) {
        //默認開啟server關閉時發送readonly事件
        url = url.addParameterIfAbsent(Constants.CHANNEL_READONLYEVENT_SENT_KEY, Boolean.TRUE.toString());
        //默認開啟heartbeat
        url = url.addParameterIfAbsent(Constants.HEARTBEAT_KEY, String.valueOf(Constants.DEFAULT_HEARTBEAT));
        String str = url.getParameter(Constants.SERVER_KEY, Constants.DEFAULT_REMOTING_SERVER);

        if (str != null && str.length() > 0 && !ExtensionLoader.getExtensionLoader(Transporter.class).hasExtension(str))
            throw new RpcException("Unsupported server type: " + str + ", url: " + url);

        url = url.addParameter(Constants.CODEC_KEY, Version.isCompatibleVersion() ? COMPATIBLE_CODEC_NAME : DubboCodec.NAME);
        ExchangeServer server;
        try {
            server = Exchangers.bind(url, requestHandler);
        } catch (RemotingException e) {
            throw new RpcException("Fail to start server(url: " + url + ") " + e.getMessage(), e);
        }
        str = url.getParameter(Constants.CLIENT_KEY);
        if (str != null && str.length() > 0) {
            Set<String> supportedTypes = ExtensionLoader.getExtensionLoader(Transporter.class).getSupportedExtensions();
            if (!supportedTypes.contains(str)) {
                throw new RpcException("Unsupported client type: " + str);
            }
        }
        return server;
    }

進入 Exchangers.bind(url, requestHandler)方法

public static ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException {
       ···
        url = url.addParameterIfAbsent(Constants.CODEC_KEY, "exchange");
        return getExchanger(url).bind(url, handler);
    }

進入getExchanger(url)方法：

public static Exchanger getExchanger(URL url) {
        String type = url.getParameter(Constants.EXCHANGER_KEY, Constants.DEFAULT_EXCHANGER);
        return getExchanger(type);
}
public static Exchanger getExchanger(String type) {
//此時type=header
    return ExtensionLoader.getExtensionLoader(Exchanger.class).getExtension(type);
}

這里返回HeaderExchanger，進入HeaderExchanger.bind方法

public ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException {
        return new HeaderExchangeServer(Transporters.bind(url, new DecodeHandler(new HeaderExchangeHandler(handler))));
}

這里初始化一個new DecodeHandler(new HeaderExchangeHandler(handler))，進入Transporters.bind()方法,這里進入transporter層

public static Server bind(URL url, ChannelHandler... handlers) throws RemotingException {
        ···
        ChannelHandler handler;
        if (handlers.length == 1) {
            handler = handlers[0];
        } else {
            handler = new ChannelHandlerDispatcher(handlers);
        }
        return getTransporter().bind(url, handler);
    }

進入getTransporter()方法

public static Transporter getTransporter() {
        return ExtensionLoader.getExtensionLoader(Transporter.class).getAdaptiveExtension();
}

老規矩，先看下Transporter接口

@SPI("netty")
public interface Transporter {
    @Adaptive({Constants.SERVER_KEY, Constants.TRANSPORTER_KEY})
    Server bind(URL url, ChannelHandler handler) throws RemotingException;
    
    @Adaptive({Constants.CLIENT_KEY, Constants.TRANSPORTER_KEY})
    Client connect(URL url, ChannelHandler handler) throws RemotingException;

}

@SPI注解，默認netty，兩個方法，一個bind方法，一個connect方法

獲取自適應適配器擴展點Transporter$Adpative，進入Transporter$Adpative.bind()方法，執行

ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.remoting.Transporter.class).getExtension("netty");

執行NettyTransport.bind方法

public Server bind(URL url, ChannelHandler listener) throws RemotingException {
        return new NettyServer(url, listener);
}
public NettyServer(URL url, ChannelHandler handler) throws RemotingException {
        super(url, ChannelHandlers.wrap(handler, ExecutorUtil.setThreadName(url, SERVER_THREAD_POOL_NAME)));
}
public AbstractServer(URL url, ChannelHandler handler) throws RemotingException {
       ···
        try {
            doOpen();
            if (logger.isInfoEnabled()) {
                logger.info("Start " + getClass().getSimpleName() + " bind " + getBindAddress() + ", export " + getLocalAddress());
            }
        } catch (Throwable t) {
            throw new RemotingException(url.toInetSocketAddress(), null, "Failed to bind " + getClass().getSimpleName()
                    + " on " + getLocalAddress() + ", cause: " + t.getMessage(), t);
        }
        //fixme replace this with better method
        DataStore dataStore = ExtensionLoader.getExtensionLoader(DataStore.class).getDefaultExtension();
        executor = (ExecutorService) dataStore.get(Constants.EXECUTOR_SERVICE_COMPONENT_KEY, Integer.toString(url.getPort()));
    }

AbstractServer的構造方法中，進入doOpen();這個是一個抽象方法，具體實現再NettyTransport中，這里可以看到具體的實現都是在各自的擴展類中去實現，這些不同的擴展類會有一些公共的方法，就可以提取出來一個抽象類去實現

protected void doOpen() throws Throwable {
        NettyHelper.setNettyLoggerFactory();
        ExecutorService boss = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerBoss", true));
        ExecutorService worker = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerWorker", true));
        ChannelFactory channelFactory = new NioServerSocketChannelFactory(boss, worker, getUrl().getPositiveParameter(Constants.IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS));
        bootstrap = new ServerBootstrap(channelFactory);

        final NettyHandler nettyHandler = new NettyHandler(getUrl(), this);
        channels = nettyHandler.getChannels();
        bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
            public ChannelPipeline getPipeline() {
                NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyServer.this);
                ChannelPipeline pipeline = Channels.pipeline();
                pipeline.addLast("decoder", adapter.getDecoder());
                pipeline.addLast("encoder", adapter.getEncoder());
                pipeline.addLast("handler", nettyHandler);
                return pipeline;
            }
        });
        // bind
        channel = bootstrap.bind(getBindAddress());
    }

在這里，我們看到了熟悉的netty代碼，設置 NioServerSocketChannelFactory boss worker的線程池線程個數為3，再設置設置編解碼和hander處理類，
回到new HeaderExchangeServer()方法中

public HeaderExchangeServer(Server server) {
        if (server == null) {
            throw new IllegalArgumentException("server == null");
        }
        this.server = server;
        this.heartbeat = server.getUrl().getParameter(Constants.HEARTBEAT_KEY, 0);
        this.heartbeatTimeout = server.getUrl().getParameter(Constants.HEARTBEAT_TIMEOUT_KEY, heartbeat * 3);
        if (heartbeatTimeout < heartbeat * 2) {
            throw new IllegalStateException("heartbeatTimeout < heartbeatInterval * 2");
        }
        //這是一個心跳定時器，采用了線程池，如果斷開就心跳重連。
        startHeatbeatTimer();
}

private void startHeatbeatTimer() {
       stopHeartbeatTimer();
       if (heartbeat > 0) {
       //每隔 heartbeat 時間執行一次,此時默認為60000ms
           heatbeatTimer = scheduled.scheduleWithFixedDelay(
                   new HeartBeatTask(new HeartBeatTask.ChannelProvider() {
                       public Collection<Channel> getChannels() {
                       //獲取channels
                           return Collections.unmodifiableCollection(
                                   HeaderExchangeServer.this.getChannels());
                       }
                   }, heartbeat, heartbeatTimeout),
                   heartbeat, heartbeat, TimeUnit.MILLISECONDS);
       }
   }

到此，Netty服務已經啟動

連接注冊中心

我們回到Registry.export()方法中，

public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException {
        //export invoker
        final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker);
        //registry provider  現在我們要從這里開始連接注冊中心
        final Registry registry = getRegistry(originInvoker);
        final URL registedProviderUrl = getRegistedProviderUrl(originInvoker);
        registry.register(registedProviderUrl);
        // 訂閱override數據
        // FIXME 提供者訂閱時，會影響同一JVM即暴露服務，又引用同一服務的的場景，因為subscribed以服務名為緩存的key，導致訂閱信息覆蓋。
        final URL overrideSubscribeUrl = getSubscribedOverrideUrl(registedProviderUrl);
        final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl);
        overrideListeners.put(overrideSubscribeUrl, overrideSubscribeListener);
        registry.subscribe(overrideSubscribeUrl, overrideSubscribeListener);
        //保證每次export都返回一個新的exporter實例
        return new Exporter<T>() {
            public Invoker<T> getInvoker() {
                return exporter.getInvoker();
            }

            public void unexport() {
                try {
                    exporter.unexport();
                } catch (Throwable t) {
                    logger.warn(t.getMessage(), t);
                }
                try {
                    registry.unregister(registedProviderUrl);
                } catch (Throwable t) {
                    logger.warn(t.getMessage(), t);
                }
                try {
                    overrideListeners.remove(overrideSubscribeUrl);
                    registry.unsubscribe(overrideSubscribeUrl, overrideSubscribeListener);
                } catch (Throwable t) {
                    logger.warn(t.getMessage(), t);
                }
            }
        };
    }

進入getRegistry()方法,根據invoker的地址獲取registry實例

private Registry getRegistry(final Invoker<?> originInvoker) {
        URL registryUrl = originInvoker.getUrl();
        if (Constants.REGISTRY_PROTOCOL.equals(registryUrl.getProtocol())) {
            String protocol = registryUrl.getParameter(Constants.REGISTRY_KEY, Constants.DEFAULT_DIRECTORY);
            registryUrl = registryUrl.setProtocol(protocol).removeParameter(Constants.REGISTRY_KEY);
        }
        return registryFactory.getRegistry(registryUrl);
}

此時，registryUrl=zookeeper://localhost:2181/com.alibaba.dubbo.registry.RegistryService?application=demo-provider&dubbo=2.0.0&export=dubbo%3A%2F%2F192.168.31.132%3A20880%2Fcom.alibaba.dubbo.demo.DemoService%3Fanyhost%3Dtrue%26application%3Ddemo-provider%26dubbo%3D2.0.0%26generic%3Dfalse%26interface%3Dcom.alibaba.dubbo.demo.DemoService%26methods%3DsayHello%26pid%3D1122%26side%3Dprovider%26timestamp%3D1538487268569&pid=1122&timestamp=1538487268461
此時registryFactory=RegistryFactory$Adaptive ,執行getRegistry方法
ExtensionLoader.getExtensionLoader(RegistryFactory.class).getExtension("zookeeper");這里得到ZookeeperRegistryFactory，執行ZookeeperRegistryFactory.getRegistry()

public Registry getRegistry(URL url) {
        url = url.setPath(RegistryService.class.getName())
                .addParameter(Constants.INTERFACE_KEY, RegistryService.class.getName())
                .removeParameters(Constants.EXPORT_KEY, Constants.REFER_KEY);
        String key = url.toServiceString();
        // 鎖定注冊中心獲取過程，保證注冊中心單一實例
        LOCK.lock();
        try {
            Registry registry = REGISTRIES.get(key);
            if (registry != null) {
                return registry;
            }
            registry = createRegistry(url);
            if (registry == null) {
                throw new IllegalStateException("Can not create registry " + url);
            }
            REGISTRIES.put(key, registry);
            return registry;
        } finally {
            // 釋放鎖
            LOCK.unlock();
        }
    }

進入createRegistry()方法：

public Registry createRegistry(URL url) {
       return new ZookeeperRegistry(url, zookeeperTransporter);
}
public ZookeeperRegistry(URL url, ZookeeperTransporter zookeeperTransporter) {
       super(url);
       if (url.isAnyHost()) {
           throw new IllegalStateException("registry address == null");
       }
       String group = url.getParameter(Constants.GROUP_KEY, DEFAULT_ROOT);
       if (!group.startsWith(Constants.PATH_SEPARATOR)) {
           group = Constants.PATH_SEPARATOR + group;
       }
       //設置根節點
       this.root = group;
       zkClient = zookeeperTransporter.connect(url);
       zkClient.addStateListener(new StateListener() {
           public void stateChanged(int state) {
               if (state == RECONNECTED) {
                   try {
                       recover();
                   } catch (Exception e) {
                       logger.error(e.getMessage(), e);
                   }
               }
           }
       });
   }

首先AbstractRegistry抽象類中，調用loadProperties()目的把注冊的服務緩存到本地

然后創建連接，zkClient = zookeeperTransporter.connect(url);進入ZkclientZookeeperTransporter.connect方法

public ZookeeperClient connect(URL url) {
       return new ZkclientZookeeperClient(url);
}
public ZkclientZookeeperClient(URL url) {
       super(url);
       //創建zk連接
       client = new ZkClient(url.getBackupAddress());
       //訂閱的目標：連接斷開，重連
       client.subscribeStateChanges(new IZkStateListener() {
           public void handleStateChanged(KeeperState state) throws Exception {
               ZkclientZookeeperClient.this.state = state;
               if (state == KeeperState.Disconnected) {
                   stateChanged(StateListener.DISCONNECTED);
               } else if (state == KeeperState.SyncConnected) {
                   stateChanged(StateListener.CONNECTED);
               }
           }

           public void handleNewSession() throws Exception {
               stateChanged(StateListener.RECONNECTED);
           }
       });
}

創建zk連接后，返回到ZookeeperRegistry的構造方法中，

zkClient.addStateListener(new StateListener() {
           public void stateChanged(int state) {
               if (state == RECONNECTED) {
                   try {
                       recover();
                   } catch (Exception e) {
                       logger.error(e.getMessage(), e);
                   }
               }
           }
       });

recover方法作用是連接失敗重連
回到ResigstryProtocol.export()方法，執行registry.register(registedProviderUrl);
調用 FailbackRegistry 類中的 register. 為什么呢?因為 ZookeeperRegistry 這個類中並沒有 register 這個方法，但是他的父類 FailbackRegistry 中存在 register 方法，而這個類又重寫了 AbstractRegistry 類中的 register 方法。所以我們可以直接定位大 FailbackRegistry 這個類中的 register 方法中

public FailbackRegistry(URL url) {
       super(url);
       int retryPeriod = url.getParameter(Constants.REGISTRY_RETRY_PERIOD_KEY, Constants.DEFAULT_REGISTRY_RETRY_PERIOD);
       this.retryFuture = retryExecutor.scheduleWithFixedDelay(new Runnable() {
           public void run() {
               // 檢測並連接注冊中心
               try {
               //失敗重連
                   retry();
               } catch (Throwable t) { // 防御性容錯
                   logger.error("Unexpected error occur at failed retry, cause: " + t.getMessage(), t);
               }
           }
       }, retryPeriod, retryPeriod, TimeUnit.MILLISECONDS);
   }

FailbackRegistry，從名字上來看，是一個失敗重試機制
調用父類的register方法，講當前url添加到緩存集合中
調用 doRegister 方法，這個方法很明顯，是一個抽象方法，會由ZookeeperRegistry 子類實現

protected void doRegister(URL url) {
        try {
            zkClient.create(toUrlPath(url), url.getParameter(Constants.DYNAMIC_KEY, true));
        } catch (Throwable e) {
            throw new RpcException("Failed to register " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
        }
}

在這里想zk注冊服務，最終實現注冊的是AbstractZookeeperClient.create方法

public void create(String path, boolean ephemeral) {
        int i = path.lastIndexOf('/');
        if (i > 0) {
            create(path.substring(0, i), false);
        }
        if (ephemeral) {
            createEphemeral(path);
        } else {
            createPersistent(path);
        }
    }

后續的注冊監聽的代碼就不分析了，服務端去zk注冊一個監聽，當zk節點發生變化時，通知服務端處理
至此，Dubbo服務的發布源碼分析完成

總結

從Dubbo官網上找到一張服務暴露的時序圖

免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。

猜您在找 dubbo源碼分析二：服務發布 dubbo源碼之四——dubbo服務發布 Dubbo源碼分析系列---服務的發布 Dubbo源碼學習--服務是如何發布的 Dubbo源碼分析（三）Dubbo的服務引用Refer dubbo源碼—service export Dubbo 源碼分析 - 服務導出 dubbo(2.5.3)源碼之服務發布與注冊 Dubbo源碼分析（五）Dubbo調用鏈-服務端 Dubbo源碼分析（三）：Dubbo之服務端（Service）