Dubbo服务发布过程

落日映苍穹つ 2023-06-16 10:38 110阅读 0赞

在我们对外暴露dubbo服务时，一般使用dubbo提供的@Service注解，或通过XML文件中的<dubbo:service>标签来配置具体的接口属性。dubbo在发布服务前会对指定的包进行扫描，解析出这些属性，将其以ServiceBean的形式保存。

ServiceBean除了继承自ServiceConfig，还实现了以下接口：

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

**InitializingBean**

InitializingBean接口只提供了一个afterPropertiesSet方法，该方法会在bean初始化的时候被调用，它的主要作用是，将配置文件中配置的属性加载到bean里面，如application、registry、service、protocol等。

**DisposableBean**

DisposableBean接口提供了一个destroy方法，当bean被销毁时，spring容器会自动执行该方法，如释放资源。

**ApplicationContextAware**

ApplicationContextAware提供了setApplicationContext方法，当spring容器初始化时，会自动将ApplicationContext注入进来。

**ApplicationListener**

ApplicationListener提供了onApplicationEvent方法，在spring上下文加载完毕后被执行。

ServiceBean实现了该方法，可以看到服务导出export是在spring上下文加载完成后执行的。

@Override
        public void onApplicationEvent(ContextRefreshedEvent event) {
            if (!isExported() && !isUnexported()) {
                if (logger.isInfoEnabled()) {
                    logger.info("The service ready on spring started. service: " + getInterface());
                }
                export();
            }
        }

**BeanNameAware**

提供了setBeanName方法，为beanName赋值。

**ApplicationEventPublisherAware**

提供了setApplicationEventPublisher方法，可以实现异步事件的发送。

**服务发布export**

ServiceBean重写了父类的export方法，增加了一个事件发布，最终还是会调用super.export()

@Override
        public void export() {
            super.export();
            // Publish ServiceBeanExportedEvent
            publishExportEvent();
        }

**ServiceConfig的export方法 **

public synchronized void export() {
            // 检查配置并更新配置
            checkAndUpdateSubConfigs();
    
            // 是否应该发布服务
            if (!shouldExport()) {
                return;
            }
    
            // 是否需要延迟发布
            if (shouldDelay()) {
                DELAY_EXPORT_EXECUTOR.schedule(this::doExport, getDelay(), TimeUnit.MILLISECONDS);
            } else {
                // 发布服务
                doExport();
            }
        }
    
        protected synchronized void doExport() {
            if (unexported) {
                throw new IllegalStateException("The service " + interfaceClass.getName() + " has already unexported!");
            }
            if (exported) {
                return;
            }
            exported = true;
    
            if (StringUtils.isEmpty(path)) {
                path = interfaceName;
            }
            doExportUrls();
        }

经过一系列校验与判断，最终调用了doExportUrls方法，再来看看该方法的逻辑。

private void doExportUrls() {
            // 获取当前服务对应的注册中心实例（支持多注册中心），组装registry URL地址
            List<URL> registryURLs = loadRegistries(true);
            // 如果服务指定多个协议（Dubbo、Rest），则依次发布
            for (ProtocolConfig protocolConfig : protocols) {
                String pathKey = URL.buildKey(getContextPath(protocolConfig).map(p -> p + "/" + path).orElse(path), group, version);
                ProviderModel providerModel = new ProviderModel(pathKey, ref, interfaceClass);
                ApplicationModel.initProviderModel(pathKey, providerModel);
                // 服务发布
                doExportUrlsFor1Protocol(protocolConfig, registryURLs);
            }
        }

**doExportUrlsFor1Protocol**

private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List<URL> registryURLs) {
            String name = protocolConfig.getName();
            if (StringUtils.isEmpty(name)) {
                name = DUBBO;
            }
    
            // 读取配置信息到map，用于组装URL
            Map<String, String> map = new HashMap<String, String>();
            map.put(SIDE_KEY, PROVIDER_SIDE);
    
            appendRuntimeParameters(map);
            appendParameters(map, metrics);
            appendParameters(map, application);
            appendParameters(map, module);
            // remove 'default.' prefix for configs from ProviderConfig
            // appendParameters(map, provider, Constants.DEFAULT_KEY);
            appendParameters(map, provider);
            appendParameters(map, protocolConfig);
            appendParameters(map, this);
           
            ......
    
            // export service
            // 获取当前服务的ip和端口
            String host = this.findConfigedHosts(protocolConfig, registryURLs, map);
            Integer port = this.findConfigedPorts(protocolConfig, name, map);
            // 组装URL
            URL url = new URL(name, host, port, getContextPath(protocolConfig).map(p -> p + "/" + path).orElse(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(SCOPE_KEY);
            // don't export when none is configured
            if (!SCOPE_NONE.equalsIgnoreCase(scope)) {
    
                // export to local if the config is not remote (export to remote only when config is remote)
                if (!SCOPE_REMOTE.equalsIgnoreCase(scope)) {
                    // 发布本地服务
                    exportLocal(url);
                }
                // export to remote if the config is not local (export to local only when config is local)
                // 发布远程服务
                if (!SCOPE_LOCAL.equalsIgnoreCase(scope)) {
                    if (!isOnlyInJvm() && logger.isInfoEnabled()) {
                        logger.info("Export dubbo service " + interfaceClass.getName() + " to url " + url);
                    }
                    if (CollectionUtils.isNotEmpty(registryURLs)) {
                        for (URL registryURL : registryURLs) {
                            //if protocol is only injvm ,not register
                            if (LOCAL_PROTOCOL.equalsIgnoreCase(url.getProtocol())) {
                                continue;
                            }
                            url = url.addParameterIfAbsent(DYNAMIC_KEY, registryURL.getParameter(DYNAMIC_KEY));
                            URL monitorUrl = loadMonitor(registryURL);
                            if (monitorUrl != null) {
                                url = url.addParameterAndEncoded(MONITOR_KEY, monitorUrl.toFullString());
                            }
                            if (logger.isInfoEnabled()) {
                                logger.info("Register dubbo service " + interfaceClass.getName() + " url " + url + " to registry " + registryURL);
                            }
    
                            // For providers, this is used to enable custom proxy to generate invoker
                            String proxy = url.getParameter(PROXY_KEY);
                            if (StringUtils.isNotEmpty(proxy)) {
                                registryURL = registryURL.addParameter(PROXY_KEY, proxy);
                            }
    
                            // 通过动态代理转换成Invoker
                            Invoker<?> invoker = PROXY_FACTORY.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(EXPORT_KEY, url.toFullString()));
                            DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this);
                            
                            // 服务发布
                            Exporter<?> exporter = protocol.export(wrapperInvoker);
                            // 发布后向注册中心注册服务信息
                            exporters.add(exporter);
                        }
                    } else {
                        // 处理没有注册中心的场景，直接发布服务
                        Invoker<?> invoker = PROXY_FACTORY.getInvoker(ref, (Class) interfaceClass, url);
                        DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this);
    
                        Exporter<?> exporter = protocol.export(wrapperInvoker);
                        exporters.add(exporter);
                    }
                    /**
                     * @since 2.7.0
                     * ServiceData Store
                     */
                    MetadataReportService metadataReportService = null;
                    if ((metadataReportService = getMetadataReportService()) != null) {
                        metadataReportService.publishProvider(url);
                    }
                }
            }
            this.urls.add(url);
        }

protocol.export

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

在源码中可以看到，protocol是一个自适应扩展点，export方法被@Adaptive注解修饰，基于Dubbo SPI相关的知识，protocol.export会调用动态生成的代理类Protocol$Adaptive：

public class Protocol$Adaptive implements org.apache.dubbo.rpc.Protocol {
        public void destroy() {
            throw new UnsupportedOperationException("The method public abstract void org.apache.dubbo.rpc.Protocol.destroy() of interface org.apache.dubbo.rpc.Protocol is not adaptive method!");
        }
     
        public int getDefaultPort() {
            throw new UnsupportedOperationException("The method public abstract int org.apache.dubbo.rpc.Protocol.getDefaultPort() of interface org.apache.dubbo.rpc.Protocol is not adaptive method!");
     
        }
     
        public org.apache.dubbo.rpc.Exporter export(org.apache.dubbo.rpc.Invoker arg0) throws org.apache.dubbo.rpc.RpcException {
            if (arg0 == null) 
                throw new IllegalArgumentException("org.apache.dubbo.rpc.Invoker argument == null");
            
            if (arg0.getUrl() == null)
                throw new IllegalArgumentException("org.apache.dubbo.rpc.Invoker argument getUrl() == null");
            
            org.apache.dubbo.common.URL url = arg0.getUrl();
            
            String extName = (url.getProtocol() == null ? "dubbo" : url.getProtocol());
            
            if (extName == null)
                throw new IllegalStateException("Failed to get extension (org.apache.dubbo.rpc.Protocol) name from url (" + url.toString() + ") use keys([protocol])");
            org.apache.dubbo.rpc.Protocol extension = 
                    (org.apache.dubbo.rpc.Protocol) ExtensionLoader.getExtensionLoader(org.apache.dubbo.rpc.Protocol.class).getExtension(extName);
            return extension.export(arg0);
        }
     
        public org.apache.dubbo.rpc.Invoker refer(java.lang.Class arg0, org.apache.dubbo.common.URL arg1) throws org.apache.dubbo.rpc.RpcException {
            if (arg1 == null) 
                throw new IllegalArgumentException("url == null");
            
            org.apache.dubbo.common.URL url = arg1;
            
            String extName = (url.getProtocol() == null ? "dubbo" : url.getProtocol());
            
            if (extName == null)
                throw new IllegalStateException("Failed to get extension (org.apache.dubbo.rpc.Protocol) name from url (" + url.toString() + ") use keys([protocol])");
            
            org.apache.dubbo.rpc.Protocol extension = 
                    (org.apache.dubbo.rpc.Protocol) ExtensionLoader.getExtensionLoader(org.apache.dubbo.rpc.Protocol.class).getExtension(extName);
            return extension.refer(arg0, arg1);
        }
    }

通过export方法可以看到，最终会通过获取静态扩展点的方式来确定真正export的实现类，如果是有注册中心的情况，extName的值为registry，对应的实现类为org.apache.dubbo.registry.integration.RegistryProtocol。

RegistryProtocol.export

@Override
        public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException {
            // 获得注册中心的url:zookeeper://192.168.10.106:2181/...
            URL registryUrl = getRegistryUrl(originInvoker);
            // url to export locally
            // 获得服务提供者的url: dubbo://192.168.10.1:20880/com.xxx.xxx....
            URL providerUrl = getProviderUrl(originInvoker);
    
            // Subscribe the override data
            // 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.
            // 订阅override数据。在admin控制台可以针对服务进行治理，如修改权重、路由机制等
            final URL overrideSubscribeUrl = getSubscribedOverrideUrl(providerUrl);
            final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl, originInvoker);
            overrideListeners.put(overrideSubscribeUrl, overrideSubscribeListener);
    
            providerUrl = overrideUrlWithConfig(providerUrl, overrideSubscribeListener);
            //export invoker
            // 本质是启动一个netty服务
            final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker, providerUrl);
    
            // url to registry
            // 根据invoker中的url获取registry实例
            final Registry registry = getRegistry(originInvoker);
            // 获取要注册到注册中心的url dubbo://ip:port/xxx...
            final URL registeredProviderUrl = getRegisteredProviderUrl(providerUrl, registryUrl);
            ProviderInvokerWrapper<T> providerInvokerWrapper = ProviderConsumerRegTable.registerProvider(originInvoker,
                    registryUrl, registeredProviderUrl);
            //to judge if we need to delay publish
            boolean register = registeredProviderUrl.getParameter("register", true);
            if (register) {
                register(registryUrl, registeredProviderUrl);
                providerInvokerWrapper.setReg(true);
            }
    
            // Deprecated! Subscribe to override rules in 2.6.x or before.
            registry.subscribe(overrideSubscribeUrl, overrideSubscribeListener);
    
            exporter.setRegisterUrl(registeredProviderUrl);
            exporter.setSubscribeUrl(overrideSubscribeUrl);
            //Ensure that a new exporter instance is returned every time export
            return new DestroyableExporter<>(exporter);
        }

doLocalExport方法里再一次调用了protocol.export，而这里的url是providerUrl，因此静态扩展点的extName的值应该为dubbo。

private <T> ExporterChangeableWrapper<T> doLocalExport(final Invoker<T> originInvoker, URL providerUrl) {
            String key = getCacheKey(originInvoker);
    
            return (ExporterChangeableWrapper<T>) bounds.computeIfAbsent(key, s -> {
                Invoker<?> invokerDelegate = new InvokerDelegate<>(originInvoker, providerUrl);
                // 将invoke转换为exporter 并启动netty服务
                return new ExporterChangeableWrapper<>((Exporter<T>) protocol.export(invokerDelegate), originInvoker);
            });
        }

在SPI配置文件org.apache.dubbo.rpc.Protocol可以确定dubbo对应的实现为org.apache.dubbo.rpc.protocol.dubbo.DubboProtocol，然而在进入DubboProtocol实现前，根据dubbo的扩展点加载机制，需要先查找扩展点的包装类，并进行依赖注入

DubboProtocol.export

最终进入DubboProtocol的export方法：

@Override
        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(STUB_EVENT_KEY, DEFAULT_STUB_EVENT);
            Boolean isCallbackservice = url.getParameter(IS_CALLBACK_SERVICE, false);
            if (isStubSupportEvent && !isCallbackservice) {
                String stubServiceMethods = url.getParameter(STUB_EVENT_METHODS_KEY);
                if (stubServiceMethods == null || stubServiceMethods.length() == 0) {
                    if (logger.isWarnEnabled()) {
                        logger.warn(new IllegalStateException("consumer [" + url.getParameter(INTERFACE_KEY) +
                                "], has set stubproxy support event ,but no stub methods founded."));
                    }
    
                } else {
                    stubServiceMethodsMap.put(url.getServiceKey(), stubServiceMethods);
                }
            }
            // 启动服务
            openServer(url);
            // 优化序列化
            optimizeSerialization(url);
    
            return exporter;
        }

openServer

private void openServer(URL url) {
            // find server.
            // 获取host:port 并以此作为服务的key
            String key = url.getAddress();
            //client can export a service which's only for server to invoke
            // 判断是否是服务端
            boolean isServer = url.getParameter(IS_SERVER_KEY, true);
            if (isServer) {
                ExchangeServer server = serverMap.get(key);
                if (server == null) {
                    synchronized (this) {
                        server = serverMap.get(key);
                        if (server == null) {
                            // 如果server没有被缓存，通过createServer创建server
                            serverMap.put(key, createServer(url));
                        }
                    }
                } else {
                    // server supports reset, use together with override
                    server.reset(url);
                }
            }
        }
    
        private ExchangeServer createServer(URL url) {
            url = URLBuilder.from(url)
                    // send readonly event when server closes, it's enabled by default
                    .addParameterIfAbsent(CHANNEL_READONLYEVENT_SENT_KEY, Boolean.TRUE.toString())
                    // enable heartbeat by default
                    .addParameterIfAbsent(HEARTBEAT_KEY, String.valueOf(DEFAULT_HEARTBEAT))
                    .addParameter(CODEC_KEY, DubboCodec.NAME)
                    .build();
            String str = url.getParameter(SERVER_KEY, DEFAULT_REMOTING_SERVER);
    
            if (str != null && str.length() > 0 && !ExtensionLoader.getExtensionLoader(Transporter.class).hasExtension(str)) {
                throw new RpcException("Unsupported server type: " + str + ", url: " + url);
            }
    
            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(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;
        }

总结一下服务发布的流程

1、基于Spring进行配置解析，将配置属性转换成ServiceBean对象。

2、各种判断校验逻辑，保证信息的安全。

3、组装URL(registry://、dubbo://、injvm://)

4、构建invoker对象。

5、RegistryProtocol.export（向注册中心注册URL）

6、各种wrapper的增强（gos、listener、filter）

7、DubboProtocol.export（发布服务）

8、启动一个NettyServer。