Dubbo是阿里巴巴开源的基于Java实现的高性能、透明化的RPC框架。深入了解Dubbo源码,有助于快速定位问题、高效实现自定义拓展。本文以Dubbo服务端初始化过程为例,分析Dubbo怎么从配置转化成可被调用的服务。
<!-- 提供方应用信息,用于计算依赖关系 --> <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节点,如: application
、 protocol
、 registry
、 provider
、 service
等,Dubbo通过实现Spring提供的 NamespaceHandler
接口,向Spring注册 BeanDefinition
解析器,使Spring能识别Dubbo命名空间下的节点,并且通过实现 BeanDefinitionParser
接口,使Spring能解析各节点的具体配置。
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》
一文。
Dubbo服务提供者由 dubbo:service
来定义的,从前面可以看到,Spring把 dubbo:service
解析成一个ServiceBean,ServiceBean实现了 ApplicationListener
和 InitializingBean
接口,
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的 InitializingBean
、 DisposableBean
、 ApplicationListener
等接口,实现了 afterPropertiesSet()
、 destroy()
、 onApplicationEvent()
等典型方法,这里便是Dubbo和Spring整合的关键, 一般第三方框架基本都是通过这几个接口和Spring整合的
。
afterPropertiesSet()
主要用来注入各种 ConfigBean
,便于服务注册过程中各种参数的获取,注意看最后关于延迟发布的几行代码, 大意是如果不延迟,就立即注册和暴露服务
。
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
中加载所有的注册中心,注意 ServiceBean
的 List registries
属性,为注册中心集合。
Step5:(逻辑同上)尝试从 BeanFacotry
中加载一个监控中心,填充 ServiceBean
的 MonitorConfig monitor
属性,如果存在多个 dubbo:monitor
配置,则抛出”Duplicate monitor configs: “。
Step6:(逻辑同上)尝试从 BeanFactory
中加载所有的协议,注意: ServiceBean
的 List protocols
是一个集合,也即一个服务可以通过多种协议暴露给消费者。
Step7:(逻辑同上)设置 ServiceBean
的 path
属性, path
属性存放的是 dubbo:service
的 beanName
(dubbo:service id)。
Step8:如果为启用延迟暴露机制,则调用 export
暴露服务。首先看一下 isDelay
的实现,然后重点分析 export
的实现原理(服务暴露的整个实现原理)。
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
,则获取 ProviderConfig
的 delay
属性,
最后如果还是 null
或配置为 -1
表示延迟暴露服务
。可见Dubbo获取运行参数的层级,便于更精确化的配置各种参数。
通过 supportedApplicationListener
可以猜到服务延迟暴露是通过Spring容器的监听器触发的 。个人更倾向于明确设置 delay=-1
或者所有层级都不配置, 因为如果提早暴露服务,此时其他的Spring bean可能还未初始化完成
,而暴露出去的服务大部分情况下依赖于Spring的其他bean来实现业务功能, 如果提早接收到客户端的请求,难免会出现各种异常
。
@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:service
或 dubbo:provider
的属性 delay
,或配置 delay
为 -1
,都表示启用延迟机制,单位为毫秒,设置为 -1
, 表示等到Spring容器初始化后再暴露服务
。
从这里也可以看出,Dubbo暴露服务的处理入口为:
ServiceBean#export --> ServiceConfig#export
。
从前一节代码分析可知,最后一步是调用 ServiceBean
的父类 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事件机制触发
)。
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
标签也就是 ServiceBean
的 provider
属性为空,调用 appendProperties
方法,填充默认属性,其具体加载顺序:
1. 从系统属性加载对应参数值,参数键:dubbo.provider.属性名,System.getProperty。 2. 从属性配置文件加载对应参数值,可通过系统属性指定属性配置文件: dubbo.properties.file,如果该值未配置,则默认取 dubbo.properties 属性配置文件。 复制代码
Step2:校验 ref
与 interface
属性。如果 ref
是 GenericService
,则为Dubbo的泛化实现,然后验证 interface
接口与 ref
引用的类型是否一致。
Step3: dubbo:service
local机制,已经废弃,被 stub
属性所替换。
Step4:处理本地存根 Stub
。
Step5:校验 ServiceBean
的 application
、 registry
、 protocol
是否为空,并从系统属性(优先)、资源文件中填充其属性。系统属性、资源文件属性的配置如下:
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:校验 stub
、 mock
类的合理性,是否是 interface
的实现类。
Step7:执行 doExportUrls()
方法暴露服务,接下来会重点分析该方法。
Step8:将服务提供者信息注册到 ApplicationModel
实例中。
private void doExportUrls() { List<URL> registryURLs = loadRegistries(true); // @ step1 for (ProtocolConfig protocolConfig : protocols) { doExportUrlsFor1Protocol(protocolConfig, registryURLs); // @ step2 } } 复制代码
Step1:首先遍历 ServiceBean
的 List registries
(所有注册中心的配置信息),然后将地址封装成URL对象,关于注册中心的所有配置属性,最终转换成url的属性(?属性名=属性值),
loadRegistries(true)
,参数的意思: true
,代表服务提供者, false
:代表服务消费者
,如果是服务提供者,则检测注册中心的配置,如果配置了 register=“false”
,则忽略该地址,如果是服务消费者,并配置了 subscribe=“false”
则表示不从该注册中心订阅服务,故也不返回。
Step2:然后遍历配置的所有协议,根据每个协议,向注册中心暴露服务,接下来重点分析 doExportUrlsFor1Protocol
方法的实现细节。
所以,从上面代码,可以看出 Dubbo同一个服务支持多种服务协议、支持向多种注册中心注册 ,很方便同一功能由各种不同实现方式的客户端调用。