转载

聊聊Dubbo(九):核心源码-服务端启动流程1

Dubbo是阿里巴巴开源的基于Java实现的高性能、透明化的RPC框架。深入了解Dubbo源码,有助于快速定位问题、高效实现自定义拓展。本文以Dubbo服务端初始化过程为例,分析Dubbo怎么从配置转化成可被调用的服务。

以典型的服务端结合Spring配置为例:

<!-- 提供方应用信息,用于计算依赖关系 -->  
<dubbo:application name="demo-provider"/>  
<!-- 用dubbo协议在20880端口暴露服务 -->  
<dubbo:protocol name="dubbo" port="20880"/>  
<!-- 使用zookeeper注册中心暴露服务地址 -->  
<dubbo:registry address="zookeeper://127.0.0.1:1234" id="registry"/>  
<!-- 默认的服务端配置 -->  
<dubbo:provider registry="registry" retries="0" timeout="5000"/>  
<!-- 和本地bean一样实现服务 -->  
<bean id="demoService" class="com.alibaba.dubbo.demo.provider.DemoServiceImpl"/>  
<!-- 声明需要暴露的服务接口 -->  
<dubbo:service interface="com.alibaba.dubbo.demo.DemoService" ref="demoService"/>  
复制代码

在Dubbo命名空间下定义了一系列XML节点,如: applicationprotocolregistryproviderservice 等,Dubbo通过实现Spring提供的 NamespaceHandler 接口,向Spring注册 BeanDefinition 解析器,使Spring能识别Dubbo命名空间下的节点,并且通过实现 BeanDefinitionParser 接口,使Spring能解析各节点的具体配置。

DubboNamespaceHandler#init() ,源码如下:

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));
}
复制代码

由以上代码可以看出, 各个节点最终被转化为各种Bean,配置的各种属性也被转化为Bean的属性 。从Bean的类型可以看出,大部分Bean只用于提供Dubbo的运行参数,只有 ServiceBean 才是本文服务发布分析入口。

备注: DubboNamespaceHandler.java & DubboBeanDefinitionParser.java 源码分析,请参考 《☆聊聊Dubbo(四):核心源码-切入Spring》 一文。

1 ServiceBean 核心入口

Dubbo服务提供者由 dubbo:service 来定义的,从前面可以看到,Spring把 dubbo:service 解析成一个ServiceBean,ServiceBean实现了 ApplicationListenerInitializingBean 接口, ServiceBean有个核心方法 export ,在这个方法中初始化服务提供者并且暴露远程服务。这个方法在bean初始化或容器中所有bean刷新完毕时被调用 ,根据 provider 的延迟设置决定,如果设置了延迟( delay 属性)则在bean初始化结束之后调用,否则在刷新事件中被调用, 默认会延迟 export ,即在所有bean的刷新结束被调用

com.alibaba.dubbo.config.spring.ServiceBean 类,源码如下:

public class ServiceBean<T> extends ServiceConfig<T> implements InitializingBean, DisposableBean, ApplicationContextAware, ApplicationListener, BeanNameAware {  
    ...
    public void afterPropertiesSet() {}
    ...
    public void onApplicationEvent(ApplicationEvent event) {}
    ...
    public void destroy() {}
}
复制代码

ServiceBean 实现了Spring的 InitializingBeanDisposableBeanApplicationListener 等接口,实现了 afterPropertiesSet()destroy()onApplicationEvent() 等典型方法,这里便是Dubbo和Spring整合的关键, 一般第三方框架基本都是通过这几个接口和Spring整合的

afterPropertiesSet() 主要用来注入各种 ConfigBean ,便于服务注册过程中各种参数的获取,注意看最后关于延迟发布的几行代码, 大意是如果不延迟,就立即注册和暴露服务

ServiceBean#afterPropertiesSet(),源码如下:

public void afterPropertiesSet() throws Exception {
        // @ step1
        if (getProvider() == null) { 
            // BeanFactoryUtils.beansOfTypeIncludingAncestors 究竟做了什么?
            // 返回指定类型和子类型的所有bean,若该bean factory 是一个继承类型的beanFactory,这个方法也会获取祖宗factory中定义的指定类型的bean。
            Map<String, ProviderConfig> providerConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ProviderConfig.class, false, false);
            if (providerConfigMap != null && providerConfigMap.size() > 0) {
                Map<String, ProtocolConfig> protocolConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ProtocolConfig.class, false, false);
                if ((protocolConfigMap == null || protocolConfigMap.size() == 0)
                        && providerConfigMap.size() > 1) { // backward compatibility
                    List<ProviderConfig> providerConfigs = new ArrayList<ProviderConfig>();
                    for (ProviderConfig config : providerConfigMap.values()) {
                        if (config.isDefault() != null && config.isDefault()) {
                            providerConfigs.add(config);
                        }
                    }
                    if (!providerConfigs.isEmpty()) {
                        setProviders(providerConfigs);
                    }
                } else {
                    ProviderConfig providerConfig = null;
                    for (ProviderConfig config : providerConfigMap.values()) {
                        if (config.isDefault() == null || config.isDefault()) {
                            if (providerConfig != null) {
                                throw new IllegalStateException("Duplicate provider configs: " + providerConfig + " and " + config);
                            }
                            providerConfig = config;
                        }
                    }
                    if (providerConfig != null) {
                        setProvider(providerConfig);
                    }
                }
            }
        }
        // @ step2
        if (getApplication() == null
                && (getProvider() == null || getProvider().getApplication() == null)) {
            Map<String, ApplicationConfig> applicationConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ApplicationConfig.class, false, false);
            if (applicationConfigMap != null && applicationConfigMap.size() > 0) {
                ApplicationConfig applicationConfig = null;
                for (ApplicationConfig config : applicationConfigMap.values()) {
                    if (config.isDefault() == null || config.isDefault()) {
                        if (applicationConfig != null) {
                            throw new IllegalStateException("Duplicate application configs: " + applicationConfig + " and " + config);
                        }
                        applicationConfig = config;
                    }
                }
                if (applicationConfig != null) {
                    setApplication(applicationConfig);
                }
            }
        }
        // @ step3
        if (getModule() == null
                && (getProvider() == null || getProvider().getModule() == null)) {
            Map<String, ModuleConfig> moduleConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ModuleConfig.class, false, false);
            if (moduleConfigMap != null && moduleConfigMap.size() > 0) {
                ModuleConfig moduleConfig = null;
                for (ModuleConfig config : moduleConfigMap.values()) {
                    if (config.isDefault() == null || config.isDefault()) {
                        if (moduleConfig != null) {
                            throw new IllegalStateException("Duplicate module configs: " + moduleConfig + " and " + config);
                        }
                        moduleConfig = config;
                    }
                }
                if (moduleConfig != null) {
                    setModule(moduleConfig);
                }
            }
        }
        // @ step4
        if ((getRegistries() == null || getRegistries().isEmpty())
                && (getProvider() == null || getProvider().getRegistries() == null || getProvider().getRegistries().isEmpty())
                && (getApplication() == null || getApplication().getRegistries() == null || getApplication().getRegistries().isEmpty())) {
            Map<String, RegistryConfig> registryConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, RegistryConfig.class, false, false);
            if (registryConfigMap != null && registryConfigMap.size() > 0) {
                List<RegistryConfig> registryConfigs = new ArrayList<RegistryConfig>();
                for (RegistryConfig config : registryConfigMap.values()) {
                    if (config.isDefault() == null || config.isDefault()) {
                        registryConfigs.add(config);
                    }
                }
                if (!registryConfigs.isEmpty()) {
                    super.setRegistries(registryConfigs);
                }
            }
        }
        // @ step5
        if (getMonitor() == null
                && (getProvider() == null || getProvider().getMonitor() == null)
                && (getApplication() == null || getApplication().getMonitor() == null)) {
            Map<String, MonitorConfig> monitorConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, MonitorConfig.class, false, false);
            if (monitorConfigMap != null && monitorConfigMap.size() > 0) {
                MonitorConfig monitorConfig = null;
                for (MonitorConfig config : monitorConfigMap.values()) {
                    if (config.isDefault() == null || config.isDefault()) {
                        if (monitorConfig != null) {
                            throw new IllegalStateException("Duplicate monitor configs: " + monitorConfig + " and " + config);
                        }
                        monitorConfig = config;
                    }
                }
                if (monitorConfig != null) {
                    setMonitor(monitorConfig);
                }
            }
        }
        // @ step6
        if ((getProtocols() == null || getProtocols().isEmpty())
                && (getProvider() == null || getProvider().getProtocols() == null || getProvider().getProtocols().isEmpty())) {
            Map<String, ProtocolConfig> protocolConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ProtocolConfig.class, false, false);
            if (protocolConfigMap != null && protocolConfigMap.size() > 0) {
                List<ProtocolConfig> protocolConfigs = new ArrayList<ProtocolConfig>();
                for (ProtocolConfig config : protocolConfigMap.values()) {
                    if (config.isDefault() == null || config.isDefault()) {
                        protocolConfigs.add(config);
                    }
                }
                if (!protocolConfigs.isEmpty()) {
                    super.setProtocols(protocolConfigs);
                }
            }
        }
        // @ step7
        if (getPath() == null || getPath().length() == 0) {
            if (beanName != null && beanName.length() > 0
                    && getInterface() != null && getInterface().length() > 0
                    && beanName.startsWith(getInterface())) {
                setPath(beanName);
            }
        }
        // @ step8
        if (!isDelay()) {
            export();
        }
    }
复制代码

Step1:如果 provider 为空,说明 dubbo:service 标签未设置 provider 属性,则尝试从 BeanFactory 中查询 dubbo:provider 实例,如果存在一个 dubbo:provider 标签,则取该实例,如果存在多个 dubbo:provider 配置则 provider 属性不能为空,否则抛出异常:“Duplicate provider configs”。

Step2:如果 application 为空,说明 dubbo:service 标签未设置 application 属性,则尝试从 BeanFactory 中查询 dubbo:application 实例,如果存在一个 dubbo:application 标签,则取该实例,如果存在多个 dubbo:application 配置,则抛出异常:“Duplicate application configs”。

Step3:如果 module 为空,说明 dubbo:service 标签未设置 module 属性,则尝试从 BeanFactory 中查询 dubbo:module 实例,如果存在一个 dubbo:module 标签,则取该实例,如果存在多个 dubbo:module ,则抛出异常:“Duplicate module configs”。

Step4:(逻辑同上)尝试从 BeanFactory 中加载所有的注册中心,注意 ServiceBeanList registries 属性,为注册中心集合。

Step5:(逻辑同上)尝试从 BeanFacotry 中加载一个监控中心,填充 ServiceBeanMonitorConfig monitor 属性,如果存在多个 dubbo:monitor 配置,则抛出”Duplicate monitor configs: “。

Step6:(逻辑同上)尝试从 BeanFactory 中加载所有的协议,注意: ServiceBeanList protocols 是一个集合,也即一个服务可以通过多种协议暴露给消费者。

Step7:(逻辑同上)设置 ServiceBeanpath 属性, path 属性存放的是 dubbo:servicebeanName (dubbo:service id)。

Step8:如果为启用延迟暴露机制,则调用 export 暴露服务。首先看一下 isDelay 的实现,然后重点分析 export 的实现原理(服务暴露的整个实现原理)。

ServiceBean#isDelay(),源码如下:

private boolean isDelay() {
        Integer delay = getDelay();
        ProviderConfig provider = getProvider();
        if (delay == null && provider != null) {
            delay = provider.getDelay();
        }
        return supportedApplicationListener && (delay == null || delay == -1);
    }
复制代码

先从 ServiceConfig 获取 delay 属性,如果为 null ,则获取 ProviderConfigdelay 属性, 最后如果还是 null 或配置为 -1 表示延迟暴露服务 。可见Dubbo获取运行参数的层级,便于更精确化的配置各种参数。

通过 supportedApplicationListener 可以猜到服务延迟暴露是通过Spring容器的监听器触发的 。个人更倾向于明确设置 delay=-1 或者所有层级都不配置, 因为如果提早暴露服务,此时其他的Spring bean可能还未初始化完成 ,而暴露出去的服务大部分情况下依赖于Spring的其他bean来实现业务功能, 如果提早接收到客户端的请求,难免会出现各种异常

ServiceBean#onApplicationEvent(),源码如下:

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

如果有设置 dubbo:servicedubbo:provider 的属性 delay ,或配置 delay-1 ,都表示启用延迟机制,单位为毫秒,设置为 -1表示等到Spring容器初始化后再暴露服务

从这里也可以看出,Dubbo暴露服务的处理入口为: ServiceBean#export --> ServiceConfig#export

2 ServiceConfig 暴露服务

从前一节代码分析可知,最后一步是调用 ServiceBean 的父类 ServiceConfig#export 方法暴露服务。

2.1 第一步:ServiceConfig#export 暴露服务

调用链:ServiceBean#afterPropertiesSet --> ServiceConfig#export

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

        if (delay != null && delay > 0) {    // @ step2
            delayExportExecutor.schedule(new Runnable() {
                @Override
                public void run() {
                    doExport();
                }
            }, delay, TimeUnit.MILLISECONDS);
        } else {
            doExport();    // @ step3
        }
    }
复制代码

Step1:判断是否暴露服务,由 dubbo:service export=“true|false” 来指定。

Step2:如果启用了 delay 机制,如果 delay 大于0,表示延迟多少毫秒后暴露服务,使用 ScheduledExecutorService 延迟调度,最终调用 doExport 方法。

Step3:执行具体的暴露逻辑 doExport ,需要大家留意: delay=-1 的处理逻辑( 基于Spring事件机制触发 )。

2.2 第二步:ServiceConfig#doExport 暴露服务

调用链:ServiceBean#afterPropertiesSet --> ServiceConfig#export --> ServiceConfig#doExport

protected synchronized void doExport() {
        if (unexported) {
            throw new IllegalStateException("Already unexported!");
        }
        if (exported) {
            return;
        }
        exported = true;
        if (interfaceName == null || interfaceName.length() == 0) {
            throw new IllegalStateException("<dubbo:service interface=/"/" /> interface not allow null!");
        }
        checkDefault(); // @ step1
        if (provider != null) {
            if (application == null) {
                application = provider.getApplication();
            }
            if (module == null) {
                module = provider.getModule();
            }
            if (registries == null) {
                registries = provider.getRegistries();
            }
            if (monitor == null) {
                monitor = provider.getMonitor();
            }
            if (protocols == null) {
                protocols = provider.getProtocols();
            }
        }
        if (module != null) {
            if (registries == null) {
                registries = module.getRegistries();
            }
            if (monitor == null) {
                monitor = module.getMonitor();
            }
        }
        if (application != null) {
            if (registries == null) {
                registries = application.getRegistries();
            }
            if (monitor == null) {
                monitor = application.getMonitor();
            }
        }
        if (ref instanceof GenericService) { // @ step2
            interfaceClass = GenericService.class;
            if (StringUtils.isEmpty(generic)) {
                generic = Boolean.TRUE.toString();
            }
        } else {
            try {
                interfaceClass = Class.forName(interfaceName, true, Thread.currentThread()
                        .getContextClassLoader());
            } catch (ClassNotFoundException e) {
                throw new IllegalStateException(e.getMessage(), e);
            }
            checkInterfaceAndMethods(interfaceClass, methods);
            checkRef();
            generic = Boolean.FALSE.toString();
        }
        if (local != null) { // @ step3
            if ("true".equals(local)) {
                local = interfaceName + "Local";
            }
            Class<?> localClass;
            try {
                localClass = ClassHelper.forNameWithThreadContextClassLoader(local);
            } catch (ClassNotFoundException e) {
                throw new IllegalStateException(e.getMessage(), e);
            }
            if (!interfaceClass.isAssignableFrom(localClass)) {
                throw new IllegalStateException("The local implementation class " + localClass.getName() + " not implement interface " + interfaceName);
            }
        }
        if (stub != null) { // @ step4
            if ("true".equals(stub)) {
                stub = interfaceName + "Stub";
            }
            Class<?> stubClass;
            try {
                stubClass = ClassHelper.forNameWithThreadContextClassLoader(stub);
            } catch (ClassNotFoundException e) {
                throw new IllegalStateException(e.getMessage(), e);
            }
            if (!interfaceClass.isAssignableFrom(stubClass)) {
                throw new IllegalStateException("The stub implementation class " + stubClass.getName() + " not implement interface " + interfaceName);
            }
        }
        checkApplication(); // @ step5
        checkRegistry();
        checkProtocol();
        appendProperties(this);

        checkStubAndMock(interfaceClass); // @ step6
        if (path == null || path.length() == 0) {
            path = interfaceName;
        }
        doExportUrls(); // @ step7

        ProviderModel providerModel = new ProviderModel(getUniqueServiceName(), ref, interfaceClass); // @ step8
        ApplicationModel.initProviderModel(getUniqueServiceName(), providerModel);
    }

    private void checkDefault() { // @ step1
        if (provider == null) {
            provider = new ProviderConfig();
        }
        appendProperties(provider);
    }
复制代码

Step1:如果 dubbo:servce 标签也就是 ServiceBeanprovider 属性为空,调用 appendProperties 方法,填充默认属性,其具体加载顺序:

1. 从系统属性加载对应参数值,参数键:dubbo.provider.属性名,System.getProperty。 

2. 从属性配置文件加载对应参数值,可通过系统属性指定属性配置文件: dubbo.properties.file,如果该值未配置,则默认取 dubbo.properties 属性配置文件。
复制代码

Step2:校验 refinterface 属性。如果 refGenericService ,则为Dubbo的泛化实现,然后验证 interface 接口与 ref 引用的类型是否一致。

Step3: dubbo:service local机制,已经废弃,被 stub 属性所替换。

Step4:处理本地存根 Stub

Step5:校验 ServiceBeanapplicationregistryprotocol 是否为空,并从系统属性(优先)、资源文件中填充其属性。系统属性、资源文件属性的配置如下:

application dubbo.application.属性名,例如 dubbo.application.name 

registry dubbo.registry.属性名,例如 dubbo.registry.address 

protocol dubbo.protocol.属性名,例如 dubbo.protocol.port 

service dubbo.service.属性名,例如 dubbo.service.stub
复制代码

Step6:校验 stubmock 类的合理性,是否是 interface 的实现类。

Step7:执行 doExportUrls() 方法暴露服务,接下来会重点分析该方法。

Step8:将服务提供者信息注册到 ApplicationModel 实例中。

2.3 第三步:ServiceConfig#doExportUrls 暴露服务

调用链:ServiceBean#afterPropertiesSet --> ServiceConfig#export --> ServiceConfig#doExport --> ServiceConfig#doExportUrls

private void doExportUrls() {  
    List<URL> registryURLs = loadRegistries(true); // @ step1
    for (ProtocolConfig protocolConfig : protocols) {
        doExportUrlsFor1Protocol(protocolConfig, registryURLs); // @ step2
    }
}
复制代码

Step1:首先遍历 ServiceBeanList registries (所有注册中心的配置信息),然后将地址封装成URL对象,关于注册中心的所有配置属性,最终转换成url的属性(?属性名=属性值), loadRegistries(true) ,参数的意思: true ,代表服务提供者, false :代表服务消费者 ,如果是服务提供者,则检测注册中心的配置,如果配置了 register=“false” ,则忽略该地址,如果是服务消费者,并配置了 subscribe=“false” 则表示不从该注册中心订阅服务,故也不返回。

Step2:然后遍历配置的所有协议,根据每个协议,向注册中心暴露服务,接下来重点分析 doExportUrlsFor1Protocol 方法的实现细节。

所以,从上面代码,可以看出 Dubbo同一个服务支持多种服务协议、支持向多种注册中心注册 ,很方便同一功能由各种不同实现方式的客户端调用。

原文  https://juejin.im/post/5bcedc1b51882576e31028f7
正文到此结束
Loading...