Dubbo源码解析-Provider服务暴露Export源码解析

上篇我们介绍了ServiceBean初始化和依赖注入过程，地址如下

Dubbo源码-Provider服务端ServiceBean初始化和属性注入-CSDN博客

        本文主要针Dubbo服务端服务Export过程，从dubbo源码角度进行解析。

        Dubbo 服务端暴露细节流程比较长，也是面试过程中比较常问的技术问题，大家可以好好仔细读一下本文。有疑问欢迎留言。

        接着说明，读Dubbo源码最好是先对Spring源码有一定的了解。如果大家需要，我也可以针对Spring框架做一系列源码的解读专栏。

         不过不用担心，如果需要Spring的源码知识，文章中也会进行Spring源码铺垫介绍的。

        如果内容中有没描述清楚的，或者大家在阅读源代码有疑问的，欢迎留言，看到就会及时回复。

        为了更清楚的分析解释源码，源代码中部分不重要的内容可能会删减，保留重要内容方便大家理解。

主要内容

• Dubbo中Provider服务Export源码解析

服务Export源码解析

        我们都知道服务端在启动的时候，会将自己的服务地址注册到注册中心，那么具体的过程和细节是怎么样的呢？接下来，我们从大概流程和细节流程以及源代码角度进行说明和分析。

        为什么有流程总结这一部分，因为好多同学为了应付面试或者觉得细节流程太繁琐或者复杂，只想弄清楚相对简练的过程。所以对服务暴露的细节流程做了一个相对笼统的总结，方便大家理解大概过程，以及应对面试。

Spring知识铺垫

• ServiceBean继承ApplicationListener<ContextRefreshedEvent>，通过监听事件。实现服务发布
• Spring容器启动时，registerListeners（）收集上下文中即成继承ApplicationListener的类
•  Spring容器启动前完成之后，广播事件finishRefresh（）->publishEvent(new ContextRefreshedEvent(this))
•  Spring启动之后，触发ServiceBean的onApplicationEvent()事件，调用export()完成服务暴露

流程总结

服务暴露的核心流程在RegistryProtocol.export中，概括如下

1. ServiceBean继承ApplicationListener。Spring启动后会触发onApplicationEvent事件
2. List<URL> registryURLs = loadRegistries(true)。加载所有Registry，将zookeeper协议URL变为registry协议URL。（多注册中心）
3. 遍历Protocols.（多协议）
4. 收集配置信息到map,把map转化为Dubbo协议URL。
5. 遍历registryURLs.
6. 将DubboUrl地址绑定到registryUrl属性中。key为export.
7. protocol.export(wrapperInvoker)：开始Registry协议URL流转。protocol包装类，调用连。Protocol调用链：QosProtocolWrapper->ProtocolFilterWrapper->ProtocolListenerWrapper->RegistryProtocol
8. QosProtocolWrapper.export():开启QosServer。就是一个Netty服务端(服务统计功能如服务列表，服务在线状态等)
9. RegistryProtocol.export():protocol.export(invokerDelegete):开始Dubbo协议流转
10. doLocalExport(originInvoker):启动netty服务端。Protocol调用链：QosProtocolWrapper->ProtocolFilterWrapper->ProtocolListenerWrapper->DubboProtocol
    1. ProtocolFilterWrapper：根据Filter生成Invoker调用链
    2. DubboProtocol:启动NettyServer：根据filter构建invoker执行链，启动nettyserver,构建handler链
11. 回到RegistryProtocol,完成服务注册：register(registryUrl, registeredProviderUrl)，dubbo协议URL写入到/dubbo/com.*.*/providers/节点下
12. 回到RegistryProtocol:对configurators节点注册事件监听:.subscribe(overrideSubscribeUrl, overrideSubscribeListener);

如果还想再简练一点

• 启动NettyServer.
• 注册服务地址到Providers节点下。
• 订阅configurators节点事件

流程细节

        说明：细节流程讲述的可能有点过分细致了，几乎细致到源码解析的每一行。包括属性收集/装配/检测/校验等细节也进行了说明。其实重点关注收集后，服务暴露开始的地方。即重点关注红色标注的部分。

核心逻辑和源码

主要关注源码类RegistryProtocol.export（）方法，该方法中实现了服务发布流程,

• 调用DubboProtocol.export启动NettyServer
• 调用zkRegistry.registry().完成服务的注册。创建providers节点，子节点写入dubbo服务地址
• 调用zkRegistry.subscribe.完成事件订阅。创建configurators节点，创建监听事件，首次notify节点事件。这个监听事件，其实就是DubboAdmin或者Api方式动态修改服务配置的实现原理。

详细过程

        细节其实已经从源代码级别进行分析了，看起来可能比较累。其实也可以跳过前面直接关注红色部分。如果觉得繁琐，也可以直接跳过看源码部分

ServiceBean.export()完成服务暴露具体细节流程

1. 设置配置类属性
   1. .checkoutDefault()
      1. .创建ProviderConfig(@EnableDubboConfig没有创建ProviderConfig的话)
         1. .未配置Provider时，则创建provider
         2. .appendProperties()并设置provider对象属性,遍历set方法。
            •  优先从环境变量取属性System.getProperty()
            • 从dubbo.property文件中获取
   2. .绑定其他配置类属性
2. 校验配置信息（application,registry,Protocol,mock,stub）
3. 暴露服务
   1. .收集Registry协议：loadRegistries()根据配置Registies，获取注册的url。（获取注册协议）
      1. 遍历所有registries
      2. .封装属性到map中(applicaiton,registryConfig等)。 appendParameters(map, application);
      3. 生成URL。UrlUtils.parseURLs(address, map)->loadRegistries()根据registry中的Adress地址和map中的key,value。生成url
         1. .zookeeper://192.168.67.139:2184/com.alibaba.dubbo.registry.RegistryService?application=dubbo_provider&dubbo=2.0.2&owner=world&pid=13218&timestamp=1709463351616
      4. URL设registry属性,把协议头换成从zookeeper改为registry
         1. registry://192.168.67.139:2184/com.alibaba.dubbo.registry.RegistryService?application=dubbo_provider&dubbo=2.0.2&owner=world&pid=13882&registry=zookeeper&timestamp=1709465346174
      5. 返回URL。上面的配置属性都会通过注册中心传递给消费者。
   2. 收集Dubbo协议，以及暴露Registry和dubbo协议：doExportUrlsFor1Protocol(protocolConfig, registryURLs)完成服务暴露
      1. .收集配置类到map中application/module/provider/protocolConfig/serviceconfig）
      2. 把methodsConfig/ArgumentConfig配置同样设置到map中
      3. .提前生成接口类对应的代理类，Wrapper.getWrapper(interfaceClass)
      4. .获取接口包含的所有方法，设置key为methods到map中
      5. 置token到map
      6. 获取host:如果Protocol和Privoder配置类都没有配置IP的话，就会调用InetAddress.getLocalHost().getHostAddress();获取主机IP。如果利用容器部署，有可能获取到内网IP，导致消费端掉不通。
      7. .获取端口:如果Protocol和Privoder配置类都没有配置端口的话，通过SPI机制获取Protocol对应的默认端口。（DubboProtocol）
      8. 把map转成URL，DUBBO协议
         1. dubbo://192.168.28.25:29015/com.xiangxue.jack.service.UserService?anyhost=true&application=dubbo_provider&bean.name=com.xiangxue.jack.service.UserService&bind.ip=192.168.28.25&bind.port=29015&default.timeout=5000&dubbo=2.0.2&generic=false&interface=com.xiangxue.jack.service.UserService&methods=doKill,queryUser&owner=world&pid=16437&revision=0.0.1-SNAPSHOT&side=provider&timeout=2000&timestamp=1709472937360
      9. 遍历注册地址，远程调用暴露服务（调用关系AbstractProxyInvoker->wrapper->ServiceClass）
         1. 遍历第一步loadRegistries()中返回的注册协议地址
         2. 加载URL 监控url. monitorUrl = loadMonitor(registryURL)
         3. dubbo协议地址URL绑定monitor属性地址
         4. 获取到invoker对象
            1. 将Dubbo协议URL地址，添加到registry协议URL属性中。key为export
               1. registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString())
            2. 通过SPI获取代理工厂。因为没有@Adaptive注解。通过javassist动态生成一个代理工厂ProxyFactory$Adaptive。
            3. 最终获取到StubProxyFactoryWrapper持有JavassistProxyFactory的代理工厂
            4. 调用代理工厂JavassistProxyFactory.getInvoker（）返回AbstractProxyInvoker。（持有被代理类【ServiceImpl】,接口类，URL(Registry)）
               1. 生成代理类：Wrapper.getWrapper对需要调用的目标类的包装类，通过javassist技术动态生成的
               2. 生成Invoker：AbstractProxyInvoker：是Dubbo最底层的Invoker对象，只有通过他才能调用provider端的目标Service服务方法。持有被代理类，接口类型，RegisterURL
               3. 调用关系AbstractProxyInvoker->wrapper->ServiceClass
            5. 包装Invoker对象为DelegateProviderMetaDataInvoker，持有Invoker对象和ServiceConfig对象

            6. 暴露远程服务：registry协议的export，protocol.export(wrapperInvoker)。【前面都是数据准备过程，接下来才是真正的服务暴露流程】

               1. SPI方式获取Protocol。最终获取到包装类，持有关系：QosProtocolWrapper->ProtocolFilterWrapper->ProtocolListenerWrapper->RegistryProtocol
               2. 开启QosServer。QosProtocolWrapper.export().就是一个Netty服务端(服务统计功能如服务列表，服务在线状态等)
               3. 完成服务注册regist.export():RegistryProtocol.export()
                  1. 启动Server：doLocalExport（）走到DubboProtocl中启动NettyServer
                     1. .创建InvokerDelegete：持有原始Invoker和Dubbo协议URL
                     2.  .服务发布：protocol.export(invokerDelegete)。protocol是SPI动态生成的Protocol$Adaptive，invokerDelegete对应的URL为Dubbo,最终会掉到DubboProtocol。调用关系：QosProtocolWrapper->ProtocolFilterWrapper->ProtocolListenerWrapper->DubboProtocol
                        1. 构建Invoker调用链：ProtocolFilterWrapper.class。SPI获取所有Filters，ProtocolFilterWrapper.buildInvokerChain，通过next指针传递
                           1. 说明：最终执行链的关系：Filter1-》Filter2-〉invokerDelegete->AbstractProxyInvoker->wrapper->ServiceClass
                        2. ProtocolListenerWrapper.export
                           1. DubboProtocol.expot():启动Server
                              1. 获取String url
                              2. 获取key:（服务名称：端口）
                              3. 创建DubboExporter：持有Invoker执行链和key和exporterMap。每个服务都对应一个DubboExporter
                              4. 构建映射关系：Map<String, Exporter<?>> exporterMap。
                              5. openserver():启动服务（每一个@Service修饰的类都会进来，但是同一个IP+端口，只会启动启动一个netty服务端）后面重点讲。
                                 1. 获取Address:ip:port
                                 2. 从serverMap获取ExchangeServer。
                                 3. 如果获取不到则创建createServer(url)：启动netty服务进行双端通信，返回ExchangeServer对象。一个主机只会启动一次NettyServer。源码后面重点讲。
                              6. 总结DubboProtocol.expot()1.完成了Netty服务端的启动，2.建立handler的链条关系
                           2. 服务发布监听包装类，其实就是提供一个扩展，服务发布后给一个通知。ListenerExporterWrapper.持有DubboExporter和List<ExporterListener>。
                     3. .包装ExporterChangeableWrapper类，持有ListenerExporterWrapper和originInvoker【registry协议】
                  2. .registryUrl = getRegistryUrl(originInvoker):获取注册协议zookeeper协议Url
                     1. .zookeeper://192.168.67.139:2184/com.alibaba.dubbo.registry.RegistryService?application=dubbo_provider&dubbo=2.0.2&export=dubbo%3A%2F%2F192.168.3.6%3A20880%2Fcom.xiangxue.jack.async.AsyncService%3Fanyhost%3Dtrue%26application%3Ddubbo_provider%26bean.name%3DServiceBean%3Acom.xiangxue.jack.async.AsyncService%26bind.ip%3D192.168.3.6%26bind.port%3D20880%26dubbo%3D2.0.2%26generic%3Dfalse%26interface%3Dcom.xiangxue.jack.async.AsyncService%26methods%3DasynctoDo%26owner%3Dworld%26pid%3D86931%26revision%3D0.0.1-SNAPSHOT%26side%3Dprovider%26timeout%3D123456%26timestamp%3D1710227953244&owner=world&pid=86931&timestamp=1710227939570
                  3.  获取注册类ZookeeperRegistry。getRegistry(originInvoker);
                  4. .registeredProviderUrl= getRegisteredProviderUrl(originInvoker):获提供者地址（写入接口下provider节点）
                     1. dubbo://192.168.3.6:20880/com.xiangxue.jack.async.AsyncService?anyhost=true&application=dubbo_provider&bean.name=ServiceBean:com.xiangxue.jack.async.AsyncService&dubbo=2.0.2&generic=false&interface=com.xiangxue.jack.async.AsyncService&methods=asynctoDo&owner=world&pid=86931&revision=0.0.1-SNAPSHOT&side=provider&timeout=123456&timestamp=1710227953244
                  5. .建立服务名称和ProviderInvokerWrapper的映射关系[缓存providerInvokers].ProviderInvokerWrapper持有originInvoker, registryUrl, registeredProviderUrl
                  6. .完成服务注册：register(registryUrl, registeredProviderUrl)。把dubbo协议地址注册到Provider节点下，其实就是创建服务节点（providers节点和dubbo服务地址临时节点）;
                     1. SPI获取ZookeeperRegistryFacotry创建Zookeeper,创建zk客户端，注册断线重连监听
                     2.  .ZookeeperRegistry.register（）
                     3. .缓存注册过的URL:registered
                     4.  .创建节点名称：doRegistry(),创建provider持久化节点和dubbo服务地址的临时节点
                  7. 注册监听事件.[以下逻辑是对configurators节点注册事件监听，如果修改了属性则会覆盖客户端的该节点的数据]
                  8. .overrideSubscribeUrl=getSubscribedOverrideUrl(registeredProviderUrl)。获取override的provider协议。后面映射关系可以将provider协议Url理解为接口。（subscribed，notified两个权局缓存）
                     1. .provider://192.168.3.6:20880/com.xiangxue.jack.async.AsyncService?anyhost=true&application=dubbo_provider&bean.name=ServiceBean:com.xiangxue.jack.async.AsyncService&category=configurators&check=false&dubbo=2.0.2&generic=false&interface=com.xiangxue.jack.async.AsyncService&methods=asynctoDo&owner=world&pid=96966&revision=0.0.1-SNAPSHOT&side=provider&timeout=123456&timestamp=1710257665285
                  9. .注册监听事件registry.subscribe(overrideSubscribeUrl, overrideSubscribeListener)订阅configurators节点下数据变更事件,path:/dubbo/com.a.b.service/configurators。
                     1.  .全局变量缓存：subscribed = new ConcurrentHashMap<URL, Set<NotifyListener>>()，缓存overrideSubscribeUrl和NotifyListener
                     2. 建立overrideListener和 ChildListener的映射
                     3. 创建path：/dubbo/com.a.b/configurators
                     4. 注册监听事件
                        1. 建立dubbo的ChildListener事件类和Curator的事件类的映射
                        2. 事件关系：NotifyListener->ChildListener->CuratorWatch
                     5. 当发生overrite时，zk监听器调用curatorWatch的process方法，最终掉到NotifyListener方法，实现动态修改配置参数功能(下节重点将)
                     6. 创建Empty协议：
                        1. empty://192.168.3.6:20880/com.xiangxue.jack.async.AsyncService?anyhost=true&application=dubbo_provider&bean.name=ServiceBean:com.xiangxue.jack.async.AsyncService&category=configurators&check=false&dubbo=2.0.2&generic=false&interface=com.xiangxue.jack.async.AsyncService&methods=asynctoDo&owner=world&pid=26895&revision=0.0.1-SNAPSHOT&side=provider&timeout=123456&timestamp=1710430398847
                     7. notify(url, listener, urls);通知Empty协议或者overide协议。

4. 为QOS统计数据做准备.看服务列表，服务在线状态
5. publishExportEvent（）发布服务暴露事件ServiceBeanExportedEvent，即发布一个服务已经暴露的通知。消费端会通过ApplicationListener关注此事件类型

源码分析

       由于provider发布流程代码较长，且流程中其中属性收集，校验等逻辑不是重点。以下我们主要介绍服务发布流程中的核心代码。即RegistryProtocol.export()

• 核心流程RegistryProtocol.export().主要做了三件事。启动NettryServer,注册dubbo服务，订阅configurators节点事件

#RegistryProtocol.export
public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException {
        //export invoker
        //这里会启动server，走到DubboProtocol中去启动server
        final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker);

        //获取注册协议
        URL registryUrl = getRegistryUrl(originInvoker);

        //registry provider
        final Registry registry = getRegistry(originInvoker);
        final URL registeredProviderUrl = getRegisteredProviderUrl(originInvoker);

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

        //建立服务名称和invoke的映射关系
        ProviderConsumerRegTable.registerProvider(originInvoker, registryUrl, registeredProviderUrl);

        if (register) {
            //这里完成了服务注册和事件监听
            register(registryUrl, registeredProviderUrl);
            //设置是否注册标识ProviderConsumerRegTable.getProviderWrapper(originInvoker).setReg(true);
        }

        //以下逻辑是对configurators节点注册事件监听，如果修改了属性则会覆盖客户端的该节点的数据
        final URL overrideSubscribeUrl = getSubscribedOverrideUrl(registeredProviderUrl);
        //zookeeper事件触发后，最终回调的listener
        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
        return new DestroyableExporter<T>(exporter, originInvoker, overrideSubscribeUrl, registeredProviderUrl);
    }

DubboProtocol.export。启动nettyServer.

#RegistryProtocol
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) {
                    //真正invoker的静态代理
                    final Invoker<?> invokerDelegete = new InvokerDelegete<T>(originInvoker, getProviderUrl(originInvoker));
                    //这里的协议头是dubbo了,这里会调用DubboProtocol.export()
                    exporter = new ExporterChangeableWrapper<T>((Exporter<T>) protocol.export(invokerDelegete), originInvoker);
                    bounds.put(key, exporter);
                }
            }
        }
        return exporter;
    }

#DubboProtocol

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 dispatching 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);
            }
        }
        //启动server，核心代码
        openServer(url);
        optimizeSerialization(url);
        return exporter;
    }



private void openServer(URL url) {
        // find server.
        //这个方法会根据@Service注解的类有多少而近来多次，但是同一个ip的netty服务端只会创建一次
        String key = url.getAddress();
        //client can export a service which's only for server to invoke
        boolean isServer = url.getParameter(Constants.IS_SERVER_KEY, true);
        if (isServer) {
            ExchangeServer server = serverMap.get(key);
            if (server == null) {
                //开启server
                serverMap.put(key, createServer(url));
            } else {
                // server supports reset, use together with override
                server.reset(url);
            }
        }
    }

总结：上面内容中，每个从业务流程和源码角度进行了详细分析，如果大家有疑问或者对文章排版任何方面有建议都可以留言评论，看到都会及时回复大家。

知识总结，分享不易，全文手敲，欢迎大家关注点赞评论收藏。