Spring与Dubbo整合原理与源码分析
【1】注解@EnableDubbo
@Target({ElementType.TYPE})@Retention(RetentionPolicy.RUNTIME)@Inherited@Documented@EnableDubboConfig // @EnableDubboConfig注解用来将properties文件中的配置项转化为对应的Bean@DubboComponentScan // @DubboComponentScan注解用来扫描服务提供者和引用者(@Service与@Reference)public @interface EnableDubbo { @AliasFor(annotation = DubboComponentScan.class, attribute = "basePackages") String[] scanBasePackages() default {}; @AliasFor(annotation = DubboComponentScan.class, attribute = "basePackageClasses") Class[] scanBasePackageClasses() default {}; @AliasFor(annotation = EnableDubboConfig.class, attribute = "multiple") boolean multipleConfig() default true;}
【2】注解@EnableDubboConfig
@Target({ElementType.TYPE})@Retention(RetentionPolicy.RUNTIME)@Inherited@Documented@Import(DubboConfigConfigurationRegistrar.class)public @interface EnableDubboConfig { boolean multiple() default true;}
1）DubboConfigConfigurationRegistrar类的作用
//因为实现了ImportBeanDefinitionRegistrar接口，spring容器就会实例化该类，并且调用其registerBeanDefinitions方法；public class DubboConfigConfigurationRegistrar implements ImportBeanDefinitionRegistrar { @Override public void registerBeanDefinitions(AnnotationMetadata importingClassMetadata, BeanDefinitionRegistry registry) { //执行DubboConfigConfigurationRegistrar; AnnotationAttributes attributes = AnnotationAttributes.fromMap( importingClassMetadata.getAnnotationAttributes(EnableDubboConfig.class.getName())); boolean multiple = attributes.getBoolean("multiple"); //默认值是true // Single Config Bindings registerBeans(registry, DubboConfigConfiguration.Single.class); if (multiple) { // Since 2.6.6 https://github.com/apache/dubbo/issues/3193 registerBeans(registry, DubboConfigConfiguration.Multiple.class); } }}
2）registerBeans做了什么
public static void registerBeans(BeanDefinitionRegistry registry, Class... annotatedClasses) { if (ObjectUtils.isEmpty(annotatedClasses)) { return; } ... AnnotatedBeanDefinitionReader reader = new AnnotatedBeanDefinitionReader(registry); ... // 利用Spring中的AnnotatedBeanDefinitionReader来解析annotatedClasses // 会解析该类上的注解，然后进行处理 reader.register(annotatedClasses);}
3）DubboConfigConfiguration类展示
public class DubboConfigConfiguration { /** * Single Dubbo {@link AbstractConfig Config} Bean Binding */ @EnableDubboConfigBindings({ @EnableDubboConfigBinding(prefix = "dubbo.application", type = ApplicationConfig.class), @EnableDubboConfigBinding(prefix = "dubbo.module", type = ModuleConfig.class), @EnableDubboConfigBinding(prefix = "dubbo.registry", type = RegistryConfig.class), @EnableDubboConfigBinding(prefix = "dubbo.protocol", type = ProtocolConfig.class), @EnableDubboConfigBinding(prefix = "dubbo.monitor", type = MonitorConfig.class), @EnableDubboConfigBinding(prefix = "dubbo.provider", type = ProviderConfig.class), @EnableDubboConfigBinding(prefix = "dubbo.consumer", type = ConsumerConfig.class), @EnableDubboConfigBinding(prefix = "dubbo.config-center", type = ConfigCenterBean.class), @EnableDubboConfigBinding(prefix = "dubbo.metadata-report", type = MetadataReportConfig.class), @EnableDubboConfigBinding(prefix = "dubbo.metrics", type = MetricsConfig.class) }) public static class Single {} /** * Multiple Dubbo {@link AbstractConfig Config} Bean Binding */ @EnableDubboConfigBindings({ @EnableDubboConfigBinding(prefix = "dubbo.applications", type = ApplicationConfig.class, multiple = true), @EnableDubboConfigBinding(prefix = "dubbo.modules", type = ModuleConfig.class, multiple = true), @EnableDubboConfigBinding(prefix = "dubbo.registries", type = RegistryConfig.class, multiple = true), @EnableDubboConfigBinding(prefix = "dubbo.protocols", type = ProtocolConfig.class, multiple = true), @EnableDubboConfigBinding(prefix = "dubbo.monitors", type = MonitorConfig.class, multiple = true), @EnableDubboConfigBinding(prefix = "dubbo.providers", type = ProviderConfig.class, multiple = true), @EnableDubboConfigBinding(prefix = "dubbo.consumers", type = ConsumerConfig.class, multiple = true), @EnableDubboConfigBinding(prefix = "dubbo.config-centers", type = ConfigCenterBean.class, multiple = true), @EnableDubboConfigBinding(prefix = "dubbo.metadata-reports", type = MetadataReportConfig.class, multiple = true), @EnableDubboConfigBinding(prefix = "dubbo.metricses", type = MetricsConfig.class, multiple = true) }) public static class Multiple {}}
4）那么必然又会解析到@EnableDubboConfigBindings注解
//又是利用了实现了ImportBeanDefinitionRegistrar接口，在实例化该类会调用其registerBeanDefinitions方法；public class DubboConfigBindingsRegistrar implements ImportBeanDefinitionRegistrar, EnvironmentAware { private ConfigurableEnvironment environment; @Override public void registerBeanDefinitions(AnnotationMetadata importingClassMetadata, BeanDefinitionRegistry registry) { //执行DubboConfigBindingsRegistrar AnnotationAttributes attributes = AnnotationAttributes.fromMap(importingClassMetadata.getAnnotationAttributes(EnableDubboConfigBindings.class.getName())); // 拿到多个@EnableDubboConfigBinding注解 AnnotationAttributes[] annotationAttributes = attributes.getAnnotationArray("value"); DubboConfigBindingRegistrar registrar = new DubboConfigBindingRegistrar(); //将环境变量注入 registrar.setEnvironment(environment); for (AnnotationAttributes element : annotationAttributes) { // 逐个解析@EnableDubboConfigBinding注解，比如@EnableDubboConfigBinding(prefix = "dubbo.application", type = ApplicationConfig.class) registrar.registerBeanDefinitions(element, registry); } } @Override public void setEnvironment(Environment environment) { Assert.isInstanceOf(ConfigurableEnvironment.class, environment); this.environment = (ConfigurableEnvironment) environment; }}
5）registrar.registerBeanDefinitions方法的调用情况
public class DubboConfigBindingRegistrar implements ImportBeanDefinitionRegistrar, EnvironmentAware { private final Log log = LogFactory.getLog(getClass()); private ConfigurableEnvironment environment; @Override public void registerBeanDefinitions(AnnotationMetadata importingClassMetadata, BeanDefinitionRegistry registry) { //执行DubboConfigBindingRegistrar AnnotationAttributes attributes = AnnotationAttributes.fromMap(importingClassMetadata.getAnnotationAttributes(EnableDubboConfigBinding.class.getName())); registerBeanDefinitions(attributes, registry); } protected void registerBeanDefinitions(AnnotationAttributes attributes, BeanDefinitionRegistry registry) { // prefix = "dubbo.application" String prefix = environment.resolvePlaceholders(attributes.getString("prefix")); // type = ApplicationConfig.class Class<? extends AbstractConfig> configClass = attributes.getClass("type"); boolean multiple = attributes.getBoolean("multiple"); //针对配置分别进行注册成Bean对象，方法1 registerDubboConfigBeans(prefix, configClass, multiple, registry); } //方法1，因为Single和Multiple都是走同一套逻辑，采用参数boolean multiple区分 private void registerDubboConfigBeans(String prefix, Class<? extends AbstractConfig> configClass, boolean multiple, BeanDefinitionRegistry registry) { // 从properties文件中根据前缀拿对应的配置项，比如根据dubbo.application前缀， // 就可以拿到如下配置： // dubbo.application.name=dubbo-demo-provider-application // dubbo.application.logger=log4j Map properties = getSubProperties(environment.getPropertySources(), prefix); // 如果没有相关的配置项，则不需要注册BeanDefinition if (CollectionUtils.isEmpty(properties)) { if (log.isDebugEnabled()) { log.debug(...); } return; } // 根据配置项生成beanNames，为什么会有多个？ // 普通情况一个dubbo.application前缀对应一个ApplicationConfig类型的Bean // 特殊情况下（配置两种协议），比如dubbo.protocols对应了： // dubbo.protocols.p1.name=dubbo // dubbo.protocols.p1.port=20880 // dubbo.protocols.p1.host=0.0.0.0 // dubbo.protocols.p2.name=http // dubbo.protocols.p2.port=8082 // dubbo.protocols.p2.host=0.0.0.0 // 那么就需要对应两个ProtocolConfig类型的Bean，那么就需要两个beanName:p1和p2 // 这里就是multiple为true或false的区别，名字的区别，根据multiple用来判断是否从配置项中获取beanName // 如果multiple为false，则看有没有配置id属性，如果没有配置则自动生成一个beanName. Set beanNames = multiple ? resolveMultipleBeanNames(properties) : Collections.singleton(resolveSingleBeanName(properties, configClass, registry)); for (String beanName : beanNames) { // 为每个beanName,注册一个空的BeanDefinition，方法2 registerDubboConfigBean(beanName, configClass, registry); // 为每个bean注册一个DubboConfigBindingBeanPostProcessor的Bean后置处理器，方法3 //这里存在的问题就是对应每一种配置都会产生对应的BeanPostProcessor，最多好像也就是10种左右 //但其实一个就可以做的任务，拓展成多个貌似不太合理，结合处理逻辑都是同一套就很尴尬 registerDubboConfigBindingBeanPostProcessor(prefix, beanName, multiple, registry); } // 注册一个NamePropertyDefaultValueDubboConfigBeanCustomizer的bean registerDubboConfigBeanCustomizers(registry); } //方法2，为对应的配置生成一个beanDefinition，并注入到容器 private void registerDubboConfigBean(String beanName, Class<? extends AbstractConfig> configClass,BeanDefinitionRegistry registry) { BeanDefinitionBuilder builder = rootBeanDefinition(configClass); AbstractBeanDefinition beanDefinition = builder.getBeanDefinition(); registry.registerBeanDefinition(beanName, beanDefinition); // ApplicatinoConfig对象 if (log.isInfoEnabled()) { log.info("...); //日志记录 } } //方法3 private void registerDubboConfigBindingBeanPostProcessor(String prefix, String beanName, boolean multiple,BeanDefinitionRegistry registry) { // 注册一个DubboConfigBindingBeanPostProcessor的Bean // 每个XxConfig的Bean对应一个DubboConfigBindingBeanPostProcessor的Bean // 比如，一个ApplicationConfig对应一个DubboConfigBindingBeanPostProcessor， // 一个ProtocolConfig也会对应一个DubboConfigBindingBeanPostProcessor // 在构造DubboConfigBindingBeanPostProcessor的时候会指定构造方法的值，这样就可以区别开来了 Class processorClass = DubboConfigBindingBeanPostProcessor.class; BeanDefinitionBuilder builder = rootBeanDefinition(processorClass); // 真实的前缀，比如dubbo.registries.r2 String actualPrefix = multiple ? normalizePrefix(prefix) + beanName : prefix; // 添加两个构造方法参数值，所以会调用DubboConfigBindingBeanPostProcessor的两个参数的构造方法 builder.addConstructorArgValue(actualPrefix).addConstructorArgValue(beanName); AbstractBeanDefinition beanDefinition = builder.getBeanDefinition(); beanDefinition.setRole(BeanDefinition.ROLE_INFRASTRUCTURE); registerWithGeneratedName(beanDefinition, registry); if (log.isInfoEnabled()) { log.info(...); } } private void registerDubboConfigBeanCustomizers(BeanDefinitionRegistry registry) { registerInfrastructureBean(registry, BEAN_NAME, NamePropertyDefaultValueDubboConfigBeanCustomizer.class); } @Override public void setEnvironment(Environment environment) { Assert.isInstanceOf(ConfigurableEnvironment.class, environment); this.environment = (ConfigurableEnvironment) environment; } private Set resolveMultipleBeanNames(Map properties) { Set beanNames = new LinkedHashSet(); // 比如dubbo.protocols.p1.name=dubbo的propertyName为p1.name for (String propertyName : properties.keySet()) { // propertyName为p1.name int index = propertyName.indexOf("."); if (index > 0) { // 截取beanName名字为p1 String beanName = propertyName.substring(0, index); beanNames.add(beanName); } } return beanNames; } private String resolveSingleBeanName(Map properties, Class<? extends AbstractConfig> configClass,BeanDefinitionRegistry registry) { // 配置了dubbo.application.id=appl，那么appl就是beanName String beanName = (String) properties.get("id"); // 如果beanName为null，则会进入if分支，由spring自动生成一个beanName,比如org.apache.dubbo.config.ApplicationConfig#0 if (!StringUtils.hasText(beanName)) { BeanDefinitionBuilder builder = rootBeanDefinition(configClass); beanName = BeanDefinitionReaderUtils.generateBeanName(builder.getRawBeanDefinition(), registry); } return beanName; }}
6）单个DubboConfigBindingBeanPostProcessor的展示（删减掉部分不怎么用到的）
public class DubboConfigBindingBeanPostProcessor implements BeanPostProcessor, ApplicationContextAware, InitializingBean, BeanDefinitionRegistryPostProcessor { private final String prefix; private final String beanName; private DubboConfigBinder dubboConfigBinder; .... private List configBeanCustomizers = Collections.emptyList(); .... @Override public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException { // 每个XxConfig对应一个BeanPostProcessor，所以每个DubboConfigBindingBeanPostProcessor只处理对应的beanName if (this.beanName.equals(beanName) && bean instanceof AbstractConfig) { AbstractConfig dubboConfig = (AbstractConfig) bean; // 从properties文件中获取值，并设置到dubboConfig对象中 bind(prefix, dubboConfig); // 设置dubboConfig对象的name属性，设置为beanName customize(beanName, dubboConfig); } return bean; } private void bind(String prefix, AbstractConfig dubboConfig) { dubboConfigBinder.bind(prefix, dubboConfig); if (log.isInfoEnabled()) { log.info(...); } } private void customize(String beanName, AbstractConfig dubboConfig) { for (DubboConfigBeanCustomizer customizer : configBeanCustomizers) { customizer.customize(beanName, dubboConfig); } } ... @Override public void afterPropertiesSet() throws Exception { initDubboConfigBinder(); // 创建DefaultDubboConfigBinder initConfigBeanCustomizers(); } private void initDubboConfigBinder() { if (dubboConfigBinder == null) { try { // 先从Spring容器中获取DubboConfigBinder，默认获取不到 dubboConfigBinder = applicationContext.getBean(DubboConfigBinder.class); } catch (BeansException ignored) { if (log.isDebugEnabled()) { log.debug("DubboConfigBinder Bean can't be found in ApplicationContext."); } // Use Default implementation // 生成一个默认的 dubboConfigBinder = createDubboConfigBinder(applicationContext.getEnvironment()); } } dubboConfigBinder.setIgnoreUnknownFields(ignoreUnknownFields); dubboConfigBinder.setIgnoreInvalidFields(ignoreInvalidFields); } private void initConfigBeanCustomizers() { // 得到之前创建了的NamePropertyDefaultValueDubboConfigBeanCustomizer Collection configBeanCustomizers = beansOfTypeIncludingAncestors(applicationContext, DubboConfigBeanCustomizer.class).values(); this.configBeanCustomizers = new ArrayList(configBeanCustomizers); AnnotationAwareOrderComparator.sort(this.configBeanCustomizers); }...}
【3】注解@DubboComponentScan
@Target(ElementType.TYPE)@Retention(RetentionPolicy.RUNTIME)@Documented@Import(DubboComponentScanRegistrar.class)public @interface DubboComponentScan { String[] value() default {}; String[] basePackages() default {}; Class[] basePackageClasses() default {};}
1）导入的DubboComponentScanRegistrar类做了什么
/又是利用了实现了ImportBeanDefinitionRegistrar接口，在实例化该类会调用其registerBeanDefinitions方法；public class DubboComponentScanRegistrar implements ImportBeanDefinitionRegistrar { @Override public void registerBeanDefinitions(AnnotationMetadata importingClassMetadata, BeanDefinitionRegistry registry) { //执行DubboComponentScanRegistrar // 拿到DubboComponentScan注解所定义的包路径，扫描该package下的类，识别这些类上 Set packagesToScan = getPackagesToScan(importingClassMetadata); // 注册ServiceAnnotationBeanPostProcessor一个Bean // 实现了BeanDefinitionRegistryPostProcessor接口，所以在Spring启动时会调用postProcessBeanDefinitionRegistry方法 // 该方法会进行扫描，扫描@Service注解了的类，然后生成BeanDefinition（会生成两个，一个普通的bean，一个ServiceBean），后续的Spring周期中会生成Bean // 在ServiceBean中会监听ContextRefreshedEvent事件，一旦Spring启动完后，就会进行服务导出 registerServiceAnnotationBeanPostProcessor(packagesToScan, registry); // 注册ReferenceAnnotationBeanPostProcessor // 实现了AnnotationInjectedBeanPostProcessor接口，继而实现了InstantiationAwareBeanPostProcessorAdapter接口 // 所以Spring在启动时，在对属性进行注入时会调用AnnotationInjectedBeanPostProcessor接口中的postProcessPropertyValues方法 // 在这个过程中会按照@Refrence注解的信息去生成一个RefrenceBean对象 registerReferenceAnnotationBeanPostProcessor(registry); } //核心方法1，注册一个对@Service注解处理的 BeanDefinitionRegistryPostProcessor private void registerServiceAnnotationBeanPostProcessor(Set packagesToScan, BeanDefinitionRegistry registry) { // 生成一个RootBeanDefinition，对应的beanClass为ServiceAnnotationBeanPostProcessor.class BeanDefinitionBuilder builder = rootBeanDefinition(ServiceAnnotationBeanPostProcessor.class); // 将包路径作为在构造ServiceAnnotationBeanPostProcessor时调用构造方法时的传入参数 builder.addConstructorArgValue(packagesToScan); builder.setRole(BeanDefinition.ROLE_INFRASTRUCTURE); AbstractBeanDefinition beanDefinition = builder.getBeanDefinition(); BeanDefinitionReaderUtils.registerWithGeneratedName(beanDefinition, registry); } //核心方法2，注册一个对属性赋值处理的AnnotationInjectedBeanPostProcessor且带有ApplicationListener事件监听功能 private void registerReferenceAnnotationBeanPostProcessor(BeanDefinitionRegistry registry) { // Register @Reference Annotation Bean Processor // 注册一个ReferenceAnnotationBeanPostProcessor做为bean，ReferenceAnnotationBeanPostProcessor是一个BeanPostProcessor BeanRegistrar.registerInfrastructureBean(registry,ReferenceAnnotationBeanPostProcessor.BEAN_NAME, ReferenceAnnotationBeanPostProcessor.class); } private Set getPackagesToScan(AnnotationMetadata metadata) { AnnotationAttributes attributes = AnnotationAttributes.fromMap( metadata.getAnnotationAttributes(DubboComponentScan.class.getName())); String[] basePackages = attributes.getStringArray("basePackages"); Class[] basePackageClasses = attributes.getClassArray("basePackageClasses"); String[] value = attributes.getStringArray("value"); // Appends value array attributes Set packagesToScan = new LinkedHashSet(Arrays.asList(value)); packagesToScan.addAll(Arrays.asList(basePackages)); for (Class basePackageClass : basePackageClasses) { packagesToScan.add(ClassUtils.getPackageName(basePackageClass)); } if (packagesToScan.isEmpty()) { return Collections.singleton(ClassUtils.getPackageName(metadata.getClassName())); } return packagesToScan; }}
【4】扫描@Service注解，并且进行处理
汇总说明：实际上便是通过处理器扫描@Service注解的类，生成两个Bean【类对应的普通Bean，与Dubbo中要用到的ServiceBean】
其中ServiceBean，是先根据注解上的信息填充对应的属性，后采用环境变量中获取配置的属性，来完成属性填充。
public class ServiceAnnotationBeanPostProcessor implements BeanDefinitionRegistryPostProcessor, EnvironmentAware,ResourceLoaderAware, BeanClassLoaderAware { ... //核心方法1 @Override public void postProcessBeanDefinitionRegistry(BeanDefinitionRegistry registry) throws BeansException { Set resolvedPackagesToScan = resolvePackagesToScan(packagesToScan); if (!CollectionUtils.isEmpty(resolvedPackagesToScan)) { // 扫描包，进行Bean注册，核心方法2调用 registerServiceBeans(resolvedPackagesToScan, registry); } else { if (logger.isWarnEnabled()) { logger.warn("packagesToScan is empty , ServiceBean registry will be ignored!"); } } } //核心方法2 private void registerServiceBeans(Set packagesToScan, BeanDefinitionRegistry registry) { DubboClassPathBeanDefinitionScanner scanner = new DubboClassPathBeanDefinitionScanner(registry, environment, resourceLoader); BeanNameGenerator beanNameGenerator = resolveBeanNameGenerator(registry); scanner.setBeanNameGenerator(beanNameGenerator); // 扫描被Service注解标注的类 scanner.addIncludeFilter(new AnnotationTypeFilter(Service.class)); scanner.addIncludeFilter(new AnnotationTypeFilter(com.alibaba.dubbo.config.annotation.Service.class)); for (String packageToScan : packagesToScan) { // Registers @Service Bean first // 扫描Dubbo自定义的@Service注解 scanner.scan(packageToScan); // 查找被@Service注解的类的BeanDefinition（无论这个类有没有被@ComponentScan注解标注了） Set beanDefinitionHolders = findServiceBeanDefinitionHolders(scanner, packageToScan, registry, beanNameGenerator); if (!CollectionUtils.isEmpty(beanDefinitionHolders)) { // 扫描到BeanDefinition开始处理它，核心方法3的调用 for (BeanDefinitionHolder beanDefinitionHolder : beanDefinitionHolders) { registerServiceBean(beanDefinitionHolder, registry, scanner); } if (logger.isInfoEnabled()) { logger.info(b...); } } else { if (logger.isWarnEnabled()) { logger.warn(...); } } } } //核心方法3 private void registerServiceBean(BeanDefinitionHolder beanDefinitionHolder, BeanDefinitionRegistry registry, DubboClassPathBeanDefinitionScanner scanner) { // 处理扫描到的每一个BeanDefinition // 1. 得到@Service注解上所配置的参数 // 2. 根据每一个BeanDefinition会再额外的生成一个ServiceBean // 3. 对于每一个被@Service注解的类（服务的实现类），会生成两个Bean，一个服务实现类对应的Bean（普通Bean，和@Component一样），一个ServiceBean（Dubbo中要用到的Bean，因为在ServiceBean中包括了很的Config） // 具体的服务实现类 Class beanClass = resolveClass(beanDefinitionHolder); // @Service可以对服务进行各种配置 Annotation service = findServiceAnnotation(beanClass); AnnotationAttributes serviceAnnotationAttributes = getAnnotationAttributes(service, false, false); // 服务实现类对应的接口 Class interfaceClass = resolveServiceInterfaceClass(serviceAnnotationAttributes, beanClass); // 服务实现类对应的bean的名字，比如：demoServiceImpl String annotatedServiceBeanName = beanDefinitionHolder.getBeanName(); // 生成一个ServiceBean，核心方法4的调用 AbstractBeanDefinition serviceBeanDefinition = buildServiceBeanDefinition(service, serviceAnnotationAttributes, interfaceClass, annotatedServiceBeanName); // ServiceBean Bean name ServiceBean表示服务，我们要使用一个服务应该拿ServiceBean String beanName = generateServiceBeanName(serviceAnnotationAttributes, interfaceClass); if (scanner.checkCandidate(beanName, serviceBeanDefinition)) { // check duplicated candidate bean // 把ServiceBean注册进去，对应的beanName为ServiceBean:org.apache.dubbo.demo.DemoService registry.registerBeanDefinition(beanName, serviceBeanDefinition); if (logger.isInfoEnabled()) { logger.info(..); } } else { if (logger.isWarnEnabled()) { logger.warn(...); } } }... //核心方法4 private AbstractBeanDefinition buildServiceBeanDefinition(Annotation serviceAnnotation,AnnotationAttributes serviceAnnotationAttributes,Class interfaceClass,String annotatedServiceBeanName) { // 生成一个ServiceBean对应的BeanDefinition BeanDefinitionBuilder builder = rootBeanDefinition(ServiceBean.class); AbstractBeanDefinition beanDefinition = builder.getBeanDefinition(); MutablePropertyValues propertyValues = beanDefinition.getPropertyValues(); String[] ignoreAttributeNames = of("provider", "monitor", "application", "module", "registry", "protocol", "interface", "interfaceName", "parameters"); // 把serviceAnnotation中的参数值赋值给ServiceBean的属性 // 如：@Service(test = "test") propertyValues.addPropertyValues(new AnnotationPropertyValuesAdapter(serviceAnnotation, environment, ignoreAttributeNames)); // References "ref" property to annotated-@Service Bean // 如：@Service(protocol = "P1")，这种就是要根据对应的值找到对应的P1的config对象里面的值 // ref属性赋值为另外一个bean, 对应的就是被@Service注解的服务实现类对应的bean addPropertyReference(builder, "ref", annotatedServiceBeanName); // Set interface builder.addPropertyValue("interface", interfaceClass.getName()); // Convert parameters into map builder.addPropertyValue("parameters", convertParameters(serviceAnnotationAttributes.getStringArray("parameters"))); // 配置了methods属性，则给ServiceBean对应的methods属性赋值 // Add methods parameters List methodConfigs = convertMethodConfigs(serviceAnnotationAttributes.get("methods")); if (!methodConfigs.isEmpty()) { builder.addPropertyValue("methods", methodConfigs); } /** * Add {@link org.apache.dubbo.config.ProviderConfig} Bean reference */ String providerConfigBeanName = serviceAnnotationAttributes.getString("provider"); if (StringUtils.hasText(providerConfigBeanName)) { addPropertyReference(builder, "provider", providerConfigBeanName); } /** * Add {@link org.apache.dubbo.config.MonitorConfig} Bean reference */ String monitorConfigBeanName = serviceAnnotationAttributes.getString("monitor"); if (StringUtils.hasText(monitorConfigBeanName)) { addPropertyReference(builder, "monitor", monitorConfigBeanName); } /** * Add {@link org.apache.dubbo.config.ApplicationConfig} Bean reference */ String applicationConfigBeanName = serviceAnnotationAttributes.getString("application"); if (StringUtils.hasText(applicationConfigBeanName)) { addPropertyReference(builder, "application", applicationConfigBeanName); } /** * Add {@link org.apache.dubbo.config.ModuleConfig} Bean reference */ String moduleConfigBeanName = serviceAnnotationAttributes.getString("module"); if (StringUtils.hasText(moduleConfigBeanName)) { addPropertyReference(builder, "module", moduleConfigBeanName); } /** * Add {@link org.apache.dubbo.config.RegistryConfig} Bean reference * 获取注解上配置的注册中心的beanName */ String[] registryConfigBeanNames = serviceAnnotationAttributes.getStringArray("registry"); List registryRuntimeBeanReferences = toRuntimeBeanReferences(registryConfigBeanNames); if (!registryRuntimeBeanReferences.isEmpty()) { builder.addPropertyValue("registries", registryRuntimeBeanReferences); } /** * Add {@link org.apache.dubbo.config.ProtocolConfig} Bean reference */ String[] protocolConfigBeanNames = serviceAnnotationAttributes.getStringArray("protocol"); List protocolRuntimeBeanReferences = toRuntimeBeanReferences(protocolConfigBeanNames); if (!protocolRuntimeBeanReferences.isEmpty()) { builder.addPropertyValue("protocols", protocolRuntimeBeanReferences); } return builder.getBeanDefinition(); }....}
【5】扫描@Reference注解，并且进行处理
1）ReferenceAnnotationBeanPostProcessor类会被调用是基于继承关系
//class ReferenceAnnotationBeanPostProcessor extends AnnotationInjectedBeanPostProcessor//abstract class AnnotationInjectedBeanPostProcessor extends InstantiationAwareBeanPostProcessorAdapter //InstantiationAwareBeanPostProcessorAdapter类便是属性注入时候会调用的//调用AnnotationInjectedBeanPostProcessor抽象类的postProcessPropertyValues方法@Overridepublic PropertyValues postProcessPropertyValues( PropertyValues pvs, PropertyDescriptor[] pds, Object bean, String beanName) throws BeanCreationException { // 寻找需要注入的属性（被@Reference标注的Field） InjectionMetadata metadata = findInjectionMetadata(beanName, bean.getClass(), pvs); try { metadata.inject(bean, beanName, pvs); } catch (BeanCreationException ex) { throw ex; } catch (Throwable ex) { throw new BeanCreationException(beanName, "Injection of @" + getAnnotationType().getSimpleName() + " dependencies is failed", ex); } return pvs;}//最终走回到ReferenceAnnotationBeanPostProcessor类的doGetInjectedBean方法
2）ReferenceAnnotationBeanPostProcessor中的方法
public class ReferenceAnnotationBeanPostProcessor extends AnnotationInjectedBeanPostProcessor implements ApplicationContextAware, ApplicationListener {... // 该方法得到的对象会赋值给@ReferenceBean注解的属性 @Override protected Object doGetInjectedBean(AnnotationAttributes attributes, Object bean, String beanName, Class injectedType,InjectionMetadata.InjectedElement injectedElement) throws Exception { // 得到引入服务的beanName // attributes里存的是@Reference注解中的所配置的属性与值 // injectedType表示引入的是哪个服务接口 // referencedBeanName的值为 ServiceBean:org.apache.dubbo.demo.DemoService 表示得到该服务Bean的beanName // referencedBeanName表示 我现在要引用的这个服务，它导出时对应的ServiceBean的beanName是什么，可以用来判断现在我引用的这个服务是不是我自己导出的 String referencedBeanName = buildReferencedBeanName(attributes, injectedType); // @Reference(methods=[Lorg.apache.dubbo.config.annotation.Method;@39b43d60) org.apache.dubbo.demo.DemoService // 我要生成一个RefrenceBean，对应的beanName， 根据@Reference注解来标识不同 String referenceBeanName = getReferenceBeanName(attributes, injectedType); // 生成一个ReferenceBean对象，方法1 ReferenceBean referenceBean = buildReferenceBeanIfAbsent(referenceBeanName, attributes, injectedType); // 把referenceBean添加到Spring容器中去，方法2 registerReferenceBean(referencedBeanName, referenceBean, attributes, injectedType); cacheInjectedReferenceBean(referenceBean, injectedElement); // 创建一个代理对象，Service中的属性被注入的就是这个代理对象 // 内部会调用referenceBean.get(); ，核心方法1 return getOrCreateProxy(referencedBeanName, referenceBeanName, referenceBean, injectedType); } //方法1 private ReferenceBean buildReferenceBeanIfAbsent(String referenceBeanName, AnnotationAttributes attributes, Class referencedType) throws Exception { ReferenceBean referenceBean = referenceBeanCache.get(referenceBeanName); if (referenceBean == null) { // 生成了一个ReferenceBean对象，attributes是@Reference注解的参数值 ReferenceBeanBuilder beanBuilder = ReferenceBeanBuilder .create(attributes, applicationContext) .interfaceClass(referencedType); referenceBean = beanBuilder.build(); referenceBeanCache.put(referenceBeanName, referenceBean); } else if (!referencedType.isAssignableFrom(referenceBean.getInterfaceClass())) { throw new IllegalArgumentException(...); } return referenceBean; } //方法2 private void registerReferenceBean(String referencedBeanName, ReferenceBean referenceBean, AnnotationAttributes attributes, Class interfaceClass) { ConfigurableListableBeanFactory beanFactory = getBeanFactory(); // @Reference(parameters=[Ljava.lang.String;@72ef8d15) org.apache.dubbo.demo.DemoService // ReferenceBean的beanName，注意这个beanName，它是直接取的@Reference的全信息 // 所以，就算引用的是同一个服务，如果@Reference注解上的信息不同，那么就会生成不同的ReferenceBean String beanName = getReferenceBeanName(attributes, interfaceClass); // 要引入的服务就是本地提供的一个服务 if (existsServiceBean(referencedBeanName)) { // If @Service bean is local one /** * Get the @Service's BeanDefinition from {@link BeanFactory} * Refer to {@link ServiceAnnotationBeanPostProcessor#buildServiceBeanDefinition} */ AbstractBeanDefinition beanDefinition = (AbstractBeanDefinition) beanFactory.getBeanDefinition(referencedBeanName); RuntimeBeanReference runtimeBeanReference = (RuntimeBeanReference) beanDefinition.getPropertyValues().get("ref"); // ServiceBean --- ref // The name of bean annotated @Service String serviceBeanName = runtimeBeanReference.getBeanName(); // register Alias rather than a new bean name, in order to reduce duplicated beans // 如果是本地提供的一个服务，那么就@Reference(parameters=[Ljava.lang.String;@72ef8d15) org.apache.dubbo.demo.DemoService // 的别名是demoService，不需要是ServiceBean的名字 beanFactory.registerAlias(serviceBeanName, beanName); } else { // Remote @Service Bean if (!beanFactory.containsBean(beanName)) { beanFactory.registerSingleton(beanName, referenceBean); } } } //核心方法1 //这里面其实有点绕，因为@Reference其实也相当于做了@Autowired的工作 //能在本地找到，如果不代理的话其实相当于@Autowired注入属性（不会走Dubbo的逻辑），所以包装成代理，让它也走Dubbo的逻辑 private Object getOrCreateProxy(String referencedBeanName, String referenceBeanName, ReferenceBean referenceBean, Class serviceInterfaceType) { //这个其实是判断本地有没有 if (existsServiceBean(referencedBeanName)) { // If the local @Service Bean exists, build a proxy of ReferenceBean //进行代理，让它走Dubbo的逻辑 return newProxyInstance(getClassLoader(), new Class[]{serviceInterfaceType}, wrapInvocationHandler(referenceBeanName, referenceBean)); } else { // ReferenceBean should be initialized and get immediately // 重点，服务引入的地方 return referenceBean.get(); } }...}
【6】图示：
Dubbo服务导出
【0】服务导出要做的几件事情：
1. 确定服务的参数2. 确定服务支持的协议3. 构造服务最终的URL4. 将服务URL注册到注册中心去5. 根据服务支持的不同协议，启动不同的Server，用来接收和处理请求6. 因为Dubbo支持动态配置服务参数，所以服务导出时还需要绑定一个监听器Listener来监听服务的参数是否有修改，如果发现有修改，则需要重新进行导出
【1】核心点记录
ServiceBean的两种暴露服务的方法1.利用InitializingBean接口，调用export()方法（没有监听器的情况下才行）2.利用监听ContextRefreshedEvent事件达到服务暴露
动态代理生成 Invoker 包装成 wrapperInvoker
RegistryProtocol进行注册
DubboProtocol对 Invoker 进行导出，返回一个Exporter
ExchangeServer
【2】ServiceBean是怎么进程服务导出的
//1是利用InitializingBean接口，调用export()方法【主要是调用父类的export()方法和发布ServiceBeanExportedEvent事件】//2是利用监听ContextRefreshedEvent事件达到服务暴露public class ServiceBean extends ServiceConfig implements InitializingBean, DisposableBean, ApplicationContextAware, ApplicationListener, BeanNameAware, ApplicationEventPublisherAware {.... @Override public void setApplicationContext(ApplicationContext applicationContext) { this.applicationContext = applicationContext; // 如果某一个Service是通过Spring暴露的， // 那么当需要获取该服务时就要从Spring容器中进行获取， // 也就是从applicationContext中获取，所以需要把applicationContext添加到SpringExtensionFactory中去 SpringExtensionFactory.addApplicationContext(applicationContext); // 一定要有这一步，不然ServiceBean将接收不到ContextRefreshedEvent事件 supportedApplicationListener = addApplicationListener(applicationContext, this); } //当接收到监听ContextRefreshedEvent事件时候 @Override public void onApplicationEvent(ContextRefreshedEvent event) { // 当前服务没有被导出并且没有卸载，才导出服务 if (!isExported() && !isUnexported()) { if (logger.isInfoEnabled()) { logger.info("The service ready on spring started. service: " + getInterface()); } // 服务导出（服务注册） export(); } } @Override @SuppressWarnings({"unchecked", "deprecation"}) public void afterPropertiesSet() throws Exception { // 如果@Service中没有配置provider if (getProvider() == null) { // 就从Spring容器中找ProviderConfig类型的Bean Map providerConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ProviderConfig.class, false, false); if (providerConfigMap != null && providerConfigMap.size() > 0) { // 从Spring容器中找ProtocolConfig类型的Bean Map protocolConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ProtocolConfig.class, false, false); // 如果存在ProtocolConfig存在，并且存在多个ProviderConfig if (CollectionUtils.isEmptyMap(protocolConfigMap) && providerConfigMap.size() > 1) { // backward compatibility // 如果找到多个，取第一个default等于true的ProviderConfig List providerConfigs = new ArrayList(); 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(...); } providerConfig = config; } } if (providerConfig != null) { setProvider(providerConfig); } } } } if (getApplication() == null && (getProvider() == null || getProvider().getApplication() == null)) { Map 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 (applicationConfig != null) { throw new IllegalStateException(...); } applicationConfig = config; } if (applicationConfig != null) { setApplication(applicationConfig); } } } if (getModule() == null && (getProvider() == null || getProvider().getModule() == null)) { Map 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(...); } moduleConfig = config; } } if (moduleConfig != null) { setModule(moduleConfig); } } } // registryIds代码能看到，但是没找到在哪里能配置 if (StringUtils.isEmpty(getRegistryIds())) { if (getApplication() != null && StringUtils.isNotEmpty(getApplication().getRegistryIds())) { setRegistryIds(getApplication().getRegistryIds()); } if (getProvider() != null && StringUtils.isNotEmpty(getProvider().getRegistryIds())) { setRegistryIds(getProvider().getRegistryIds()); } } if ((CollectionUtils.isEmpty(getRegistries())) && (getProvider() == null || CollectionUtils.isEmpty(getProvider().getRegistries())) && (getApplication() == null || CollectionUtils.isEmpty(getApplication().getRegistries()))) { Map registryConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, RegistryConfig.class, false, false); if (CollectionUtils.isNotEmptyMap(registryConfigMap)) { List registryConfigs = new ArrayList(); if (StringUtils.isNotEmpty(registryIds)) { Arrays.stream(COMMA_SPLIT_PATTERN.split(registryIds)).forEach(id -> { if (registryConfigMap.containsKey(id)) { registryConfigs.add(registryConfigMap.get(id)); } }); } if (registryConfigs.isEmpty()) { for (RegistryConfig config : registryConfigMap.values()) { if (StringUtils.isEmpty(registryIds) && (config.isDefault() == null || config.isDefault().booleanValue())) { registryConfigs.add(config); } } } if (!registryConfigs.isEmpty()) { super.setRegistries(registryConfigs); } } } if (getMetadataReportConfig() == null) { Map metadataReportConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, MetadataReportConfig.class, false, false); if (metadataReportConfigMap != null && metadataReportConfigMap.size() == 1) { super.setMetadataReportConfig(metadataReportConfigMap.values().iterator().next()); } else if (metadataReportConfigMap != null && metadataReportConfigMap.size() > 1) { throw new IllegalStateException("Multiple MetadataReport configs: " + metadataReportConfigMap); } } if (getConfigCenter() == null) { Map configenterMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ConfigCenterConfig.class, false, false); if (configenterMap != null && configenterMap.size() == 1) { super.setConfigCenter(configenterMap.values().iterator().next()); } else if (configenterMap != null && configenterMap.size() > 1) { throw new IllegalStateException("Multiple ConfigCenter found:" + configenterMap); } } if (getMonitor() == null && (getProvider() == null || getProvider().getMonitor() == null) && (getApplication() == null || getApplication().getMonitor() == null)) { Map 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); } } } if (getMetrics() == null) { Map metricsConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, MetricsConfig.class, false, false); if (metricsConfigMap != null && metricsConfigMap.size() > 0) { MetricsConfig metricsConfig = null; for (MetricsConfig config : metricsConfigMap.values()) { if (metricsConfig != null) { throw new IllegalStateException("Duplicate metrics configs: " + metricsConfig + " and " + config); } metricsConfig = config; } if (metricsConfig != null) { setMetrics(metricsConfig); } } } // protocolIds也没看到在哪里配置 if (StringUtils.isEmpty(getProtocolIds())) { if (getProvider() != null && StringUtils.isNotEmpty(getProvider().getProtocolIds())) { setProtocolIds(getProvider().getProtocolIds()); } } if (CollectionUtils.isEmpty(getProtocols()) && (getProvider() == null || CollectionUtils.isEmpty(getProvider().getProtocols()))) { Map protocolConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ProtocolConfig.class, false, false); if (protocolConfigMap != null && protocolConfigMap.size() > 0) { List protocolConfigs = new ArrayList(); if (StringUtils.isNotEmpty(getProtocolIds())) { Arrays.stream(COMMA_SPLIT_PATTERN.split(getProtocolIds())) .forEach(id -> { if (protocolConfigMap.containsKey(id)) { protocolConfigs.add(protocolConfigMap.get(id)); } }); } if (protocolConfigs.isEmpty()) { for (ProtocolConfig config : protocolConfigMap.values()) { if (StringUtils.isEmpty(protocolIds)) { protocolConfigs.add(config); } } } if (!protocolConfigs.isEmpty()) { super.setProtocols(protocolConfigs); } } } if (StringUtils.isEmpty(getPath())) { if (StringUtils.isNotEmpty(beanName) && StringUtils.isNotEmpty(getInterface()) && beanName.startsWith(getInterface())) { setPath(beanName); } } //没有监听事件才做暴露服务 if (!supportedApplicationListener) { export(); } } //服务暴露的核心方法 @Override public void export() { super.export(); // Publish ServiceBeanExportedEvent publishExportEvent(); } private void publishExportEvent() { ServiceBeanExportedEvent exportEvent = new ServiceBeanExportedEvent(this); applicationEventPublisher.publishEvent(exportEvent); } @Override public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) { this.applicationEventPublisher = applicationEventPublisher; }}
1）ServiceConfig类#export()方法
public synchronized void export() { //读取配置并补全（最新最全的配置），方法1 checkAndUpdateSubConfigs(); // 检查服务是否需要导出 if (!shouldExport()) { return; } // 检查是否需要延迟发布 if (shouldDelay()) { DELAY_EXPORT_EXECUTOR.schedule(this::doExport, getDelay(), TimeUnit.MILLISECONDS); } else { // 导出服务，方法2 doExport(); }}
2）方法1：ServiceConfig类#checkAndUpdateSubConfigs()方法
/** * 1. ServiceConfig中的某些属性如果是空的，那么就从ProviderConfig、ModuleConfig、ApplicationConfig中获取 * 2. 从配置中心获取配置，包括应用配置和全局配置 * 3. 从配置中心获取Provider配置 * 4. 从配置中心获取Protocol配置 * 5. 如果ApplicationConfig为空，则构造一个ApplicationConfig * 6. 从配置中心获取Registry配置 * 7. 更新ServiceConfig中的属性为优先级最高的配置 * 8. 更新MetadataReportConfig中的属性为优先级最高的配置 * 9. 检查当前服务是不是一个泛化服务 * 10.检查Stub和Local * 11.检查Mock */public void checkAndUpdateSubConfigs() {// ServiceConfig中的某些属性如果是空的，那么就从ProviderConfig、ModuleConfig、ApplicationConfig中获取（之前生成的配置Bean） completeCompoundConfigs(); // 方法1.1 // 从配置中心获取配置，包括应用配置和全局配置 // 把获取到的配置放入到Environment中的externalConfigurationMap和appExternalConfigurationMap中 // 并刷新所有的Config属性 startConfigCenter(); // 如果没有ProviderConfig对象，则创建一个 checkDefault(); // 如果没有单独的配置protocols，那么就从provider获取配置的协议，添加到的ServiceConfig中去 // 假如程序员在配置文件中配了一个dubbo协议，配置中心的全局配置或应用配置中也配置了一个协议，那么就会被添加到ServiceConfig中 checkProtocol(); checkApplication(); // if protocol is not injvm checkRegistry // 如果protocol不是只有injvm协议，表示服务调用不是只在本机jvm里面调用，那就需要用到注册中心 // 如果protocol是injvm，表示本地调用 if (!isOnlyInJvm()) { checkRegistry(); } // 刷新ServiceConfig，方法1.2 this.refresh(); // 如果配了metadataReportConfig，那么就刷新配置 checkMetadataReport(); if (StringUtils.isEmpty(interfaceName)) { throw new IllegalStateException(" interface not allow null!"); } // 当前服务对应的实现类是一个GenericService，表示没有特定的接口 if (ref instanceof GenericService) { 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); } // 刷新MethodConfig，并判断MethodConfig中对应的方法在接口中是否存在 checkInterfaceAndMethods(interfaceClass, methods); // 实现类是不是该接口类型 checkRef(); generic = Boolean.FALSE.toString(); } // local和stub一样，不建议使用了 if (local != null) { // 如果本地存根为true，则存根类为interfaceName + "Local" if (Boolean.TRUE.toString().equals(local)) { local = interfaceName + "Local"; } // 加载本地存根类 Class localClass; try { localClass = ClassUtils.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) { // 如果本地存根为true，则存根类为interfaceName + "Stub" if (Boolean.TRUE.toString().equals(stub)) { stub = interfaceName + "Stub"; } Class stubClass; try { stubClass = ClassUtils.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); } } // 检查local和stub checkStubAndLocal(interfaceClass); // 检查mock checkMock(interfaceClass);}
3）方法1.1，AbstractInterfaceConfig类#startConfigCenter()方法
void startConfigCenter() { if (configCenter == null) { ConfigManager.getInstance().getConfigCenter().ifPresent(cc -> this.configCenter = cc); } // 如果配置了ConfigCenter if (this.configCenter != null) { // 从其他位置获取配置中心的相关属性信息，比如配置中心地址 // TODO there may have duplicate refresh this.configCenter.refresh(); // 属性更新后，从远程配置中心获取数据(应用配置，全局配置) prepareEnvironment(); } // 从配置中心取到配置数据后，刷新所有的XxConfig中的属性 ConfigManager.getInstance().refreshAll();}private void prepareEnvironment() { if (configCenter.isValid()) { if (!configCenter.checkOrUpdateInited()) { return; } // 动态配置中心，管理台上的配置中心 DynamicConfiguration dynamicConfiguration = getDynamicConfiguration(configCenter.toUrl()); // 如果是zookeeper，获取的就是/dubbo/config/dubbo/dubbo.properties节点中的内容 String configContent = dynamicConfiguration.getProperties(configCenter.getConfigFile(), configCenter.getGroup()); String appGroup = application != null ? application.getName() : null; String appConfigContent = null; if (StringUtils.isNotEmpty(appGroup)) { // 获取的就是/dubbo/config/dubbo-demo-consumer-application/dubbo.properties节点中的内容 // 这里有bug appConfigContent = dynamicConfiguration.getProperties (StringUtils.isNotEmpty(configCenter.getAppConfigFile()) ? configCenter.getAppConfigFile() : configCenter.getConfigFile(), appGroup ); } try { Environment.getInstance().setConfigCenterFirst(configCenter.isHighestPriority()); Environment.getInstance().updateExternalConfigurationMap(parseProperties(configContent)); Environment.getInstance().updateAppExternalConfigurationMap(parseProperties(appConfigContent)); } catch (IOException e) { throw new IllegalStateException(...); } }}
4）方法1.2，AbstractInterfaceConfig类#refresh()方法
// 刷新XxConfig// 一个XxConfig对象的属性可能是有值的，也可能是没有值的，这时需要从其他位置获取属性值,来进行属性的覆盖// 覆盖的优先级，从大到小为系统变量->配置中心应用配置->配置中心全局配置->注解或xml中定义->dubbo.properties文件// 以ServiceConfig为例，ServiceConfig中包括很多属性，比如timeout// 但是在定义一个Service时，如果在注解上没有配置timeout，那么就会其他地方获取timeout的配置// 比如可以从系统变量->配置中心应用配置->配置中心全局配置->注解或xml中定义->dubbo.properties文件// refresh是刷新，将当前ServiceConfig上的set方法所对应的属性更新为优先级最高的值public void refresh() { try { CompositeConfiguration compositeConfiguration = Environment.getInstance().getConfiguration(getPrefix(), getId()); // 表示XxConfig对象本身- AbstractConfig Configuration config = new ConfigConfigurationAdapter(this); //设置顺序， if (Environment.getInstance().isConfigCenterFirst()) { // The sequence would be: SystemConfiguration -> AppExternalConfiguration -> ExternalConfiguration -> AbstractConfig -> PropertiesConfiguration compositeConfiguration.addConfiguration(4, config); } else { // The sequence would be: SystemConfiguration -> AbstractConfig -> AppExternalConfiguration -> ExternalConfiguration -> PropertiesConfiguration compositeConfiguration.addConfiguration(2, config); } // loop methods, get override value and set the new value back to method // Method[] methods = getClass().getMethods(); for (Method method : methods) { // 是不是setXX()方法 if (MethodUtils.isSetter(method)) { // 获取xx配置项的value String value = StringUtils.trim(compositeConfiguration.getString(extractPropertyName(getClass(), method))); // isTypeMatch() is called to avoid duplicate and incorrect update, for example, we have two 'setGeneric' methods in ReferenceConfig. if (StringUtils.isNotEmpty(value) && ClassUtils.isTypeMatch(method.getParameterTypes()[0], value)) { method.invoke(this, ClassUtils.convertPrimitive(method.getParameterTypes()[0], value)); } // 是不是setParameters()方法 } else if (isParametersSetter(method)) { // 获取parameter配置项的value String value = StringUtils.trim(compositeConfiguration.getString(extractPropertyName(getClass(), method))); if (StringUtils.isNotEmpty(value)) { Map map = invokeGetParameters(getClass(), this); map = map == null ? new HashMap() : map; map.putAll(convert(StringUtils.parseParameters(value), "")); invokeSetParameters(getClass(), this, map); } } } } catch (Exception e) { logger.error("Failed to override ", e); }}public CompositeConfiguration getConfiguration(String prefix, String id) { CompositeConfiguration compositeConfiguration = new CompositeConfiguration(); // Config center has the highest priority // JVM环境变量 compositeConfiguration.addConfiguration(this.getSystemConfig(prefix, id)); // 操作系统环境变量 compositeConfiguration.addConfiguration(this.getEnvironmentConfig(prefix, id)); // 配置中心APP配置 compositeConfiguration.addConfiguration(this.getAppExternalConfig(prefix, id)); // 配置中心Global配置 compositeConfiguration.addConfiguration(this.getExternalConfig(prefix, id)); // dubbo.properties中的配置 compositeConfiguration.addConfiguration(this.getPropertiesConfig(prefix, id)); return compositeConfiguration;}
5）方法2，ServiceConfig类#doExport()方法
protected synchronized void doExport() { // 当前服务已经被取消了，就不能再导出了 if (unexported) { throw new IllegalStateException(...); } // 已经导出了，就不再导出了 if (exported) { return; } exported = true; if (StringUtils.isEmpty(path)) { path = interfaceName; } doExportUrls();}@SuppressWarnings({"unchecked", "rawtypes"})private void doExportUrls() { // 得到url，注册服务也是一个服务，所以也会有对应的url，通过调用该url完成服务注册 List registryURLs = loadRegistries(true); // // 遍历每个协议 // 一个协议一个服务 for (ProtocolConfig protocolConfig : protocols) { // path表示服务名 // contextPath表示应用名（可配置） // pathKey = group/contextpath/path:version // 例子：myGroup/user/org.apache.dubbo.demo.DemoService:1.0.1 String pathKey = URL.buildKey(getContextPath(protocolConfig).map(p -> p + "/" + path).orElse(path), group, version); // ProviderModel中存在服务提供者访问路径，实现类，接口，以及接口中的各个方法对应的ProviderMethodModel // ProviderMethodModel表示某一个方法，方法名，所属的服务的， ProviderModel providerModel = new ProviderModel(pathKey, ref, interfaceClass); // ApplicationModel表示应用中有哪些服务提供者和引用了哪些服务 ApplicationModel.initProviderModel(pathKey, providerModel); // 每种协议导出一个单独的服务,注册到各个注册中心 doExportUrlsFor1Protocol(protocolConfig, registryURLs); }}
6）doExportUrlsFor1Protocol方法
private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List registryURLs) { // protocolConfig表示某个协议，registryURLs表示所有的注册中心 // 如果配置的某个协议，没有配置name，那么默认为dubbo String name = protocolConfig.getName(); if (StringUtils.isEmpty(name)) { name = DUBBO; } // 这个map表示服务url的参数 Map map = new HashMap(); map.put(SIDE_KEY, PROVIDER_SIDE); appendRuntimeParameters(map); // 监控中心参数 appendParameters(map, metrics); // 应用相关参数 appendParameters(map, application); // 模块相关参数 appendParameters(map, module); // 提供者相关参数 appendParameters(map, provider); // 协议相关参数 appendParameters(map, protocolConfig); // 服务本身相关参数 appendParameters(map, this); // 服务中某些方法参数 if (CollectionUtils.isNotEmpty(methods)) { for (MethodConfig method : methods) { // 某个方法的配置参数，注意有prefix appendParameters(map, method, method.getName()); String retryKey = method.getName() + ".retry"; // 如果某个方法配置存在xx.retry=false，则改成xx.retry=0 if (map.containsKey(retryKey)) { String retryValue = map.remove(retryKey); if (Boolean.FALSE.toString().equals(retryValue)) { map.put(method.getName() + ".retries", "0"); } } List arguments = method.getArguments(); if (CollectionUtils.isNotEmpty(arguments)) { // 遍历当前方法配置中的参数配置 for (ArgumentConfig argument : arguments) { // 如果配置了type，则遍历当前接口的所有方法，然后找到方法名和当前方法名相等的方法，可能存在多个 // 如果配置了index,则看index对应位置的参数类型是否等于type,如果相等，则向map中存入argument对象中的参数 // 如果没有配置index，那么则遍历方法所有的参数类型，等于type则向map中存入argument对象中的参数 // 如果没有配置type,但配置了index,则把对应位置的argument放入map // convert argument type if (argument.getType() != null && argument.getType().length() > 0) { Method[] methods = interfaceClass.getMethods(); // visit all methods if (methods != null && methods.length > 0) { for (int i = 0; i < methods.length; i++) { String methodName = methods[i].getName(); // target the method, and get its signature if (methodName.equals(method.getName())) { Class[] argtypes = methods[i].getParameterTypes(); // one callback in the method if (argument.getIndex() != -1) { if (argtypes[argument.getIndex()].getName().equals(argument.getType())) { appendParameters(map, argument, method.getName() + "." + argument.getIndex()); } else { throw new IllegalArgumentException(...); } } else { // multiple callbacks in the method for (int j = 0; j < argtypes.length; j++) { Class argclazz = argtypes[j]; if (argclazz.getName().equals(argument.getType())) { appendParameters(map, argument, method.getName() + "." + j); if (argument.getIndex() != -1 && argument.getIndex() != j) { throw new IllegalArgumentException(...); } } } } } } } } else if (argument.getIndex() != -1) { appendParameters(map, argument, method.getName() + "." + argument.getIndex()); } else { throw new IllegalArgumentException(...); } } } } // end of methods for } if (ProtocolUtils.isGeneric(generic)) { map.put(GENERIC_KEY, generic); map.put(METHODS_KEY, ANY_VALUE); } else { String revision = Version.getVersion(interfaceClass, version); if (revision != null && revision.length() > 0) { map.put(REVISION_KEY, revision); } // 通过接口对应的Wrapper，拿到接口中所有的方法名字 String[] methods = Wrapper.getWrapper(interfaceClass).getMethodNames(); if (methods.length == 0) { logger.warn("No method found in service interface " + interfaceClass.getName()); map.put(METHODS_KEY, ANY_VALUE); } else { map.put(METHODS_KEY, StringUtils.join(new HashSet(Arrays.asList(methods)), ",")); } } // Token是为了防止服务被消费者直接调用（伪造http请求） // 主要是存于注册中心，调用时Token匹配成功即算通过（所以要求调用者是通过注册中心获取提供方的信息） if (!ConfigUtils.isEmpty(token)) { if (ConfigUtils.isDefault(token)) { map.put(TOKEN_KEY, UUID.randomUUID().toString()); } else { map.put(TOKEN_KEY, token); } } // export service // 通过该host和port访问该服务 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); // url：http://192.168.40.17:80/org.apache.dubbo.demo.DemoService?anyhost=true&application=dubbo-demo-annotation-provider&bean.name=ServiceBean:org.apache.dubbo.demo.DemoService&bind.ip=192.168.40.17&bind.port=80&deprecated=false&dubbo=2.0.2&dynamic=true&generic=false&interface=org.apache.dubbo.demo.DemoService&methods=sayHello&pid=285072&release=&side=provider×tamp=1585206500409 // 可以通过ConfiguratorFactory，在服务导出时候进行统一配置 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); // scope可能为null，remote, local,none // don't export when none is configured if (!SCOPE_NONE.equalsIgnoreCase(scope)) { // 如果scope为none,则不会进行任何的服务导出，既不会远程，也不会本地 // export to local if the config is not remote (export to remote only when config is remote) if (!SCOPE_REMOTE.equalsIgnoreCase(scope)) { // 如果scope不是remote,则会进行本地导出，会把当前url的protocol改为injvm，然后进行导出 exportLocal(url); } // export to remote if the config is not local (export to local only when config is local) if (!SCOPE_LOCAL.equalsIgnoreCase(scope)) { // 如果scope不是local,则会进行远程导出 if (CollectionUtils.isNotEmpty(registryURLs)) { // 如果有注册中心，则将服务注册到注册中心 for (URL registryURL : registryURLs) { //if protocol is only injvm ,not register // 如果是injvm，则不需要进行注册中心注册 if (LOCAL_PROTOCOL.equalsIgnoreCase(url.getProtocol())) { continue; } // 该服务是否是动态，对应zookeeper上表示是否是临时节点，对应dubbo中的功能就是静态服务 url = url.addParameterIfAbsent(DYNAMIC_KEY, registryURL.getParameter(DYNAMIC_KEY)); // 基于注册中心地址的到监控中心地址，为什么是基于注册中心地址？ URL monitorUrl = loadMonitor(registryURL); // 把监控中心地址添加到服务url中 if (monitorUrl != null) { url = url.addParameterAndEncoded(MONITOR_KEY, monitorUrl.toFullString()); } // 服务的register参数，如果为true，则表示要注册到注册中心 if (logger.isInfoEnabled()) { if (url.getParameter(REGISTER_KEY, true)) { logger.info(...); } else { logger.info(...); } } // For providers, this is used to enable custom proxy to generate invoker // 服务使用的动态代理机制，如果为空则使用javassit String proxy = url.getParameter(PROXY_KEY); if (StringUtils.isNotEmpty(proxy)) { registryURL = registryURL.addParameter(PROXY_KEY, proxy); } // 生成一个当前服务接口的代理对象 // 使用代理生成一个Invoker，Invoker表示服务提供者的代理，可以使用Invoker的invoke方法执行服务 // 对应的url为 registry://127.0.0.1:2181/org.apache.dubbo.registry.RegistryService?application=dubbo-demo-annotation-provider&dubbo=2.0.2&export=http%3A%2F%2F192.168.40.17%3A80%2Forg.apache.dubbo.demo.DemoService%3Fanyhost%3Dtrue%26application%3Ddubbo-demo-annotation-provider%26bean.name%3DServiceBean%3Aorg.apache.dubbo.demo.DemoService%26bind.ip%3D192.168.40.17%26bind.port%3D80%26deprecated%3Dfalse%26dubbo%3D2.0.2%26dynamic%3Dtrue%26generic%3Dfalse%26interface%3Dorg.apache.dubbo.demo.DemoService%26methods%3DsayHello%26pid%3D19472%26release%3D%26side%3Dprovider%26timestamp%3D1585207994860&pid=19472®istry=zookeeper×tamp=1585207994828 // 这个Invoker中包括了服务的实现者、服务接口类、服务的注册地址（针对当前服务的，参数export指定了当前服务） // 此invoker表示一个可执行的服务，调用invoker的invoke()方法即可执行服务,同时此invoker也可用来导出 Invoker invoker = PROXY_FACTORY.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(EXPORT_KEY, url.toFullString())); // DelegateProviderMetaDataInvoker也表示服务提供者，包括了Invoker和服务的配置 DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this); // 使用特定的协议来对服务进行导出，这里的协议为RegistryProtocol，导出成功后得到一个Exporter // 1. 先使用RegistryProtocol进行服务注册 // 2. 注册完了之后，使用DubboProtocol进行导出 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); } // 根据服务url，讲服务的元信息存入元数据中心 MetadataReportService metadataReportService = null; if ((metadataReportService = getMetadataReportService()) != null) { metadataReportService.publishProvider(url); } } } this.urls.add(url);}
7）protocol.export的调用
@Overridepublic Exporter export(final Invoker originInvoker) throws RpcException { // 导出服务 // registry:// ---> RegistryProtocol // zookeeper:// ---> ZookeeperRegistry // dubbo:// ---> DubboProtocol // provider:// ---> // 将registry://xxx?xx=xx®istry=zookeeper 转为---> zookeeper://xxx?xx=xx URL registryUrl = getRegistryUrl(originInvoker); // zookeeper://127.0.0.1:2181/org.apache.dubbo.registry.RegistryService?application=dubbo-demo-provider-application&dubbo=2.0.2&export=dubbo%3A%2F%2F192.168.40.17%3A20880%2Forg.apache.dubbo.demo.DemoService%3Fanyhost%3Dtrue%26application%3Ddubbo-demo-provider-application%26bean.name%3DServiceBean%3Aorg.apache.dubbo.demo.DemoService%26bind.ip%3D192.168.40.17%26bind.port%3D20880%26deprecated%3Dfalse%26dubbo%3D2.0.2%26dynamic%3Dtrue%26generic%3Dfalse%26interface%3Dorg.apache.dubbo.demo.DemoService%26logger%3Dlog4j%26methods%3DsayHello%26pid%3D27656%26release%3D2.7.0%26side%3Dprovider%26timeout%3D3000%26timestamp%3D1590735956489&logger=log4j&pid=27656&release=2.7.0×tamp=1590735956479 // 得到服务提供者url URL providerUrl = getProviderUrl(originInvoker); // dubbo://192.168.40.17:20880/org.apache.dubbo.demo.DemoService?anyhost=true&application=dubbo-demo-provider-application&bean.name=ServiceBean:org.apache.dubbo.demo.DemoService&bind.ip=192.168.40.17&bind.port=20880&deprecated=false&dubbo=2.0.2&dynamic=true&generic=false&interface=org.apache.dubbo.demo.DemoService&logger=log4j&methods=sayHello&pid=27656&release=2.7.0&side=provider&timeout=3000×tamp=1590735956489 // 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. // overrideSubscribeUrl是老版本的动态配置监听url，表示了需要监听的服务以及监听的类型（configurators， 这是老版本上的动态配置） // 在服务提供者url的基础上，生成一个overrideSubscribeUrl，协议为provider://，增加参数category=configurators&check=false final URL overrideSubscribeUrl = getSubscribedOverrideUrl(providerUrl); // 一个overrideSubscribeUrl对应一个OverrideListener，用来监听变化事件，监听到overrideSubscribeUrl的变化后， // OverrideListener就会根据变化进行相应处理，具体处理逻辑看OverrideListener的实现 final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl, originInvoker); overrideListeners.put(overrideSubscribeUrl, overrideSubscribeListener); // 在这个方法里会利用providerConfigurationListener和serviceConfigurationListener去重写providerUrl // providerConfigurationListener表示应用级别的动态配置监听器，providerConfigurationListener是RegistyProtocol的一个属性 // serviceConfigurationListener表示服务级别的动态配置监听器，serviceConfigurationListener是在每暴露一个服务时就会生成一个 // 这两个监听器都是新版本中的监听器 // 新版本监听的zk路径是： // 服务： /dubbo/config/dubbo/org.apache.dubbo.demo.DemoService.configurators节点的内容 // 应用： /dubbo/config/dubbo/dubbo-demo-provider-application.configurators节点的内容 // 注意，要喝配置中心的路径区分开来，配置中心的路径是： // 应用：/dubbo/config/dubbo/org.apache.dubbo.demo.DemoService/dubbo.properties节点的内容 // 全局：/dubbo/config/dubbo/dubbo.properties节点的内容 providerUrl = overrideUrlWithConfig(providerUrl, overrideSubscribeListener); // export invoker // 根据动态配置重写了providerUrl之后，就会调用DubboProtocol或HttpProtocol去进行导出服务了 final ExporterChangeableWrapper exporter = doLocalExport(originInvoker, providerUrl); // url to registry // 得到注册中心-ZookeeperRegistry final Registry registry = getRegistry(originInvoker); // 得到存入到注册中心去的providerUrl,会对服务提供者url中的参数进行简化 final URL registeredProviderUrl = getRegisteredProviderUrl(providerUrl, registryUrl); // 将当前服务提供者Invoker，以及该服务对应的注册中心地址，以及简化后的服务url存入ProviderConsumerRegTable ProviderInvokerWrapper providerInvokerWrapper = ProviderConsumerRegTable.registerProvider(originInvoker, registryUrl, registeredProviderUrl); //to judge if we need to delay publish //是否需要注册到注册中心 boolean register = providerUrl.getParameter(REGISTER_KEY, true); if (register) { // 注册服务，把简化后的服务提供者url注册到registryUrl中去 register(registryUrl, registeredProviderUrl); providerInvokerWrapper.setReg(true); } // 针对老版本的动态配置，需要把overrideSubscribeListener绑定到overrideSubscribeUrl上去进行监听 // 兼容老版本的配置修改，利用overrideSubscribeListener去监听旧版本的动态配置变化 // 监听overrideSubscribeUrl provider://192.168.40.17:20880/org.apache.dubbo.demo.DemoService?anyhost=true&application=dubbo-demo-annotation-provider&bean.name=ServiceBean:org.apache.dubbo.demo.DemoService&bind.ip=192.168.40.17&bind.port=20880&category=configurators&check=false&deprecated=false&dubbo=2.0.2&dynamic=true&generic=false&interface=org.apache.dubbo.demo.DemoService&methods=sayHello&pid=416332&release=&side=provider×tamp=1585318241955 // 那么新版本的providerConfigurationListener和serviceConfigurationListener是在什么时候进行订阅的呢？在这两个类构造的时候 // Deprecated! Subscribe to override rules in 2.6.x or before. // 老版本监听的zk路径是：/dubbo/org.apache.dubbo.demo.DemoService/configurators/override://0.0.0.0/org.apache.dubbo.demo.DemoService?category=configurators&compatible_config=true&dynamic=false&enabled=true&timeout=6000 // 监听的是路径的内容，不是节点的内容 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);}public void register(URL registryUrl, URL registeredProviderUrl) { Registry registry = registryFactory.getRegistry(registryUrl); // 调用FailbackRegistry类的方法再转到ZookeeperRegistry的register方法 registry.register(registeredProviderUrl);}//FailbackRegistry类（进行失败重试）@Overridepublic void register(URL url) { super.register(url); removeFailedRegistered(url); removeFailedUnregistered(url); try { // 这里才会调用ZookeeperRegistry类的方法 doRegister(url); } catch (Exception e) { Throwable t = e; // If the startup detection is opened, the Exception is thrown directly. boolean check = getUrl().getParameter(Constants.CHECK_KEY, true) && url.getParameter(Constants.CHECK_KEY, true) && !CONSUMER_PROTOCOL.equals(url.getProtocol()); boolean skipFailback = t instanceof SkipFailbackWrapperException; if (check || skipFailback) { if (skipFailback) { t = t.getCause(); } throw new IllegalStateException(...); } else { logger.error(...); } addFailedRegistered(url); }}//ZookeeperRegistry的真正注册的地方@Overridepublic void doRegister(URL url) { try { zkClient.create(toUrlPath(url), url.getParameter(DYNAMIC_KEY, true)); } catch (Throwable e) { throw new RpcException("Failed to register " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e); }}
8）doLocalExport方法
@SuppressWarnings("unchecked")private ExporterChangeableWrapper doLocalExport(final Invoker originInvoker, URL providerUrl) { String key = getCacheKey(originInvoker); return (ExporterChangeableWrapper) bounds.computeIfAbsent(key, s -> { Invoker invokerDelegate = new InvokerDelegate(originInvoker, providerUrl); // protocol属性的值是哪来的，是在SPI中注入进来的，是一个代理类 // 这里实际利用的就是DubboProtocol或HttpProtocol去export NettyServer // 为什么需要ExporterChangeableWrapper？方便注销已经被导出的服务 return new ExporterChangeableWrapper((Exporter) protocol.export(invokerDelegate), originInvoker); });}@Overridepublic Exporter export(Invoker invoker) throws RpcException { URL url = invoker.getUrl(); // export service. String key = serviceKey(url); // 构造一个Exporter DubboExporter exporter = new DubboExporter(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(...); } } else { // 服务的stub方法 stubServiceMethodsMap.put(url.getServiceKey(), stubServiceMethods); } } // 开启NettyServer openServer(url); optimizeSerialization(url); return exporter;}private void openServer(URL url) { // find server. String key = url.getAddress(); // 获得ip地址和port， 192.168.40.17:20880 // NettyClient, NettyServer //client can export a service which's only for server to invoke boolean isServer = url.getParameter(IS_SERVER_KEY, true); if (isServer) { // 缓存Server对象 ExchangeServer server = serverMap.get(key); // DCL，Double Check Lock if (server == null) { synchronized (this) { server = serverMap.get(key); if (server == null) { // 创建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(); // 协议的服务器端实现类型，比如：dubbo协议的mina,netty等，http协议的jetty,servlet等，默认为netty 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); } // 通过url绑定端口，和对应的请求处理器 ExchangeServer server; try { // requestHandler是请求处理器，类型为ExchangeHandler // 表示从url的端口接收到请求后，requestHandler来进行处理 server = Exchangers.bind(url, requestHandler); } catch (RemotingException e) { throw new RpcException("Fail to start server(url: " + url + ") " + e.getMessage(), e); } // 协议的客户端实现类型，比如：dubbo协议的mina,netty等 str = url.getParameter(CLIENT_KEY); if (str != null && str.length() > 0) { Set supportedTypes = ExtensionLoader.getExtensionLoader(Transporter.class).getSupportedExtensions(); if (!supportedTypes.contains(str)) { throw new RpcException("Unsupported client type: " + str); } } return server;}9）当数据发生改变时 OverrideListener 监听者的处理
@Overridepublic synchronized void notify(List urls) { List matchedUrls = getMatchedUrls(urls, subscribeUrl.addParameter(CATEGORY_KEY, CONFIGURATORS_CATEGORY)); // No matching results if (matchedUrls.isEmpty()) { return; } // 对发生了变化的url进行过滤，只取url是override协议，或者参数category等于configurators的url this.configurators = Configurator.toConfigurators(classifyUrls(matchedUrls, UrlUtils::isConfigurator)).orElse(configurators); // 根据Override协议修改 doOverrideIfNecessary();}public synchronized void doOverrideIfNecessary() { final Invoker invoker; if (originInvoker instanceof InvokerDelegate) { invoker = ((InvokerDelegate) originInvoker).getInvoker(); } else { invoker = originInvoker; } //The origin invoker 当前服务的原始服务提供者url URL originUrl = RegistryProtocol.this.getProviderUrl(invoker); String key = getCacheKey(originInvoker); ExporterChangeableWrapper exporter = bounds.get(key); if (exporter == null) { logger.warn(new IllegalStateException("error state, exporter should not be null")); return; } //The current, may have been merged many times，当前服务被导出的url URL currentUrl = exporter.getInvoker().getUrl(); //根据configurators修改url，configurators是全量的，并不是某个新增的或删除的，所以是基于原始的url进行修改，并不是基于currentUrl //Merged with this configuration URL newUrl = getConfigedInvokerUrl(configurators, originUrl); newUrl = getConfigedInvokerUrl(providerConfigurationListener.getConfigurators(), newUrl); newUrl = getConfigedInvokerUrl(serviceConfigurationListeners.get(originUrl.getServiceKey()).getConfigurators(), newUrl); // 修改过的url如果和目前的url不相同，则重新按newUrl导出 if (!currentUrl.equals(newUrl)) { RegistryProtocol.this.reExport(originInvoker, newUrl); logger.info(...); }}public void reExport(final Invoker originInvoker, URL newInvokerUrl) { // 根据newInvokerUrl进行导出 // update local exporter ExporterChangeableWrapper exporter = doChangeLocalExport(originInvoker, newInvokerUrl); // 获取准确的ProviderUrl // update registry URL registryUrl = getRegistryUrl(originInvoker); // 对于一个服务提供者url，在注册到注册中心时，会先进行简化，所以如果 final URL registeredProviderUrl = getRegisteredProviderUrl(newInvokerUrl, registryUrl); //decide if we need to re-publish // 根据getServiceKey获取ProviderInvokerWrapper ProviderInvokerWrapper providerInvokerWrapper = ProviderConsumerRegTable.getProviderWrapper(registeredProviderUrl, originInvoker); // 生成一个新的ProviderInvokerWrapper ProviderInvokerWrapper newProviderInvokerWrapper = ProviderConsumerRegTable.registerProvider(originInvoker, registryUrl, registeredProviderUrl); /** * Only if the new url going to Registry is different with the previous one should we do unregister and register. * 如果新的服务提供者url简化后的url和这个服务之前的服务提供者url简化后的url不相等，则需要把新的简化后的服务提供者url注册到注册中心去 */ if (providerInvokerWrapper.isReg() && !registeredProviderUrl.equals(providerInvokerWrapper.getProviderUrl())) { unregister(registryUrl, providerInvokerWrapper.getProviderUrl()); register(registryUrl, registeredProviderUrl); newProviderInvokerWrapper.setReg(true); } exporter.setRegisterUrl(registeredProviderUrl);}@SuppressWarnings("unchecked")private ExporterChangeableWrapper doChangeLocalExport(final Invoker originInvoker, URL newInvokerUrl) { String key = getCacheKey(originInvoker); final ExporterChangeableWrapper exporter = (ExporterChangeableWrapper) bounds.get(key); if (exporter == null) { logger.warn(new IllegalStateException("error state, exporter should not be null")); } else { // 到这里才能真正明白，为什么需要InvokerDelegate // InvokerDelegate表示一个调用者，由invoker+url构成，invoker不变，url可变 final Invoker invokerDelegate = new InvokerDelegate(originInvoker, newInvokerUrl); //这次openServer会走HeaderExchangeServer的reset方法 exporter.setExporter(protocol.export(invokerDelegate)); } return exporter;}//这里面存在重新导出，关闭旧的延迟任务（旧的心跳任务），开启新的心跳任务，但是Netty不会关闭，也不会重启@Overridepublic void reset(URL url) { server.reset(url); try { int currHeartbeat = getHeartbeat(getUrl()); int currIdleTimeout = getIdleTimeout(getUrl()); int heartbeat = getHeartbeat(url); int idleTimeout = getIdleTimeout(url); if (currHeartbeat != heartbeat || currIdleTimeout != idleTimeout) { cancelCloseTask(); startIdleCheckTask(url); } } catch (Throwable t) { logger.error(t.getMessage(), t); }}
【3】汇总
服务导出的入口为ServiceBean中的export()方法，当Spring启动完之后，通过接收Spring的ContextRefreshedEvent事件来触发export()方法的执行。一个ServiceBean对象就表示一个Dubbo服务，ServiceBean对象中的参数就表示服务的参数，比如timeout，该对象的参数值来至@Service注解中所定义的。服务导出主要得做两件事情：1. 根据服务的参数信息，启动对应的网络服务器（netty、tomcat、jetty等），用来接收网络请求2. 将服务的信息注册到注册中心但是在做这两件事情之前得先把服务的参数确定好，因为一个Dubbo服务的参数，除开可以在@Service注解中去配置，还会继承Dubbo服务所属应用（Application）上的配置，
还可以在配置中心或JVM环境变量中去配置某个服务的参数，所以首先要做的是确定好当前服务最终的（优先级最高）的参数值。确定好服务参数之后，就根据所配置的协议启动对应的网络服务器。在启动网络服务器时，并且在网络服务器接收请求的过程中，都可以从服务参数中获取信息，比如最大连接数，线程数，socket超时时间等等。启动完网络服务器之后，就将服务信息注册到注册中心。同时还有向注册中心注册监听器，监听Dubbo的中的动态配置信息变更。
Dubbo服务引入
【0】核心点记录
生成代理对象（代理对象应该包含的功能）{ 1.获取服务提供者列表 2.Mock--------MockClusterInvoker 3.路由筛选 4.负载均衡 5.集群容错------FailoverClusterInvoker 6.构造NettyClient 7.发送数据（Invocation）}代理对象的InvokerMockClusterInvoker Invoker属性塞入FailoverClusterInvoker FailoverClusterInvoker Invoker属性塞入DubboInvoker
【1】服务要怎么引入（ReferenceConfig类#get()方法）
//服务引入的入口方法public synchronized T get() { //读取配置并补全（最新最全的配置） checkAndUpdateSubConfigs(); if (destroyed) { throw new IllegalStateException(...); } if (ref == null) { // 入口 init(); } return ref; // Invoke代理}
【2】检查并拿到最新配置（ReferenceConfig类#checkAndUpdateSubConfigs()方法）
public void checkAndUpdateSubConfigs() { if (StringUtils.isEmpty(interfaceName)) { throw new IllegalStateException(...); } // 填充ReferenceConfig对象中的属性 completeCompoundConfigs(); // 开启配置中心 startConfigCenter(); // get consumer's global configuration checkDefault(); // 刷新ReferenceConfig对象的属性值 this.refresh(); // 设置泛化 if (getGeneric() == null && getConsumer() != null) { setGeneric(getConsumer().getGeneric()); } if (ProtocolUtils.isGeneric(getGeneric())) { interfaceClass = GenericService.class; } else { try { interfaceClass = Class.forName(interfaceName, true, Thread.currentThread().getContextClassLoader()); } catch (ClassNotFoundException e) { throw new IllegalStateException(...); } checkInterfaceAndMethods(interfaceClass, methods); } resolveFile(); checkApplication(); checkMetadataReport();}
【3】初始化生成代理对象（ReferenceConfig类#init()方法）
private void init() { if (initialized) { return; } //准备参数，进行参数配置 checkStubAndLocal(interfaceClass); checkMock(interfaceClass); Map map = new HashMap(); map.put(SIDE_KEY, CONSUMER_SIDE); appendRuntimeParameters(map); if (!ProtocolUtils.isGeneric(getGeneric())) { String revision = Version.getVersion(interfaceClass, version); if (revision != null && revision.length() > 0) { map.put(REVISION_KEY, revision); } String[] methods = Wrapper.getWrapper(interfaceClass).getMethodNames(); if (methods.length == 0) { logger.warn("No method found in service interface " + interfaceClass.getName()); map.put(METHODS_KEY, ANY_VALUE); } else { map.put(METHODS_KEY, StringUtils.join(new HashSet(Arrays.asList(methods)), COMMA_SEPARATOR)); } } map.put(INTERFACE_KEY, interfaceName); appendParameters(map, metrics); appendParameters(map, application); appendParameters(map, module); // remove 'default.' prefix for configs from ConsumerConfig // appendParameters(map, consumer, Constants.DEFAULT_KEY); appendParameters(map, consumer); appendParameters(map, this); Map attributes = null; if (CollectionUtils.isNotEmpty(methods)) { attributes = new HashMap(); for (MethodConfig methodConfig : methods) { appendParameters(map, methodConfig, methodConfig.getName()); String retryKey = methodConfig.getName() + ".retry"; if (map.containsKey(retryKey)) { String retryValue = map.remove(retryKey); if ("false".equals(retryValue)) { map.put(methodConfig.getName() + ".retries", "0"); } } attributes.put(methodConfig.getName(), convertMethodConfig2AsyncInfo(methodConfig)); } } String hostToRegistry = ConfigUtils.getSystemProperty(DUBBO_IP_TO_REGISTRY); if (StringUtils.isEmpty(hostToRegistry)) { hostToRegistry = NetUtils.getLocalHost(); } else if (isInvalidLocalHost(hostToRegistry)) { throw new IllegalArgumentException(...); } map.put(REGISTER_IP_KEY, hostToRegistry); //参数配置完成后去生成代理对象 ref = createProxy(map); String serviceKey = URL.buildKey(interfaceName, group, version); ApplicationModel.initConsumerModel(serviceKey, buildConsumerModel(serviceKey, attributes)); initialized = true;}
【4】生成代理对象（ReferenceConfig类#createProxy()方法）
@SuppressWarnings({"unchecked", "rawtypes", "deprecation"})private T createProxy(Map map) { if (shouldJvmRefer(map)) { // injvm:// URL url = new URL(LOCAL_PROTOCOL, LOCALHOST_VALUE, 0, interfaceClass.getName()).addParameters(map); invoker = REF_PROTOCOL.refer(interfaceClass, url); } else { // 为什么会有urls，因为可以在@Reference的url属性中配置多个url，可以是点对点的服务地址，也可以是注册中心的地址 urls.clear(); // reference retry init will add url to urls, lead to OOM // @Reference中指定了url属性 if (url != null && url.length() > 0) { // user specified URL, could be peer-to-peer address, or register center's address. String[] us = SEMICOLON_SPLIT_PATTERN.split(url); // 用;号切分 if (us != null && us.length > 0) { for (String u : us) { URL url = URL.valueOf(u); if (StringUtils.isEmpty(url.getPath())) { url = url.setPath(interfaceName); } // 如果是注册中心地址，则在url中添加一个refer参数 if (REGISTRY_PROTOCOL.equals(url.getProtocol())) { // map表示消费者端配置的参数 urls.add(url.addParameterAndEncoded(REFER_KEY, StringUtils.toQueryString(map))); } else { // 如果是服务地址 // 有可能url中配置了参数，map中表示的服务消费者消费服务时的参数，所以需要合并 urls.add(ClusterUtils.mergeUrl(url, map)); } } } } else { // assemble URL from register center's configuration // @Reference中的protocol属性表示使用哪个协议调用服务，如果不是本地调用协议injvm://，则把注册中心地址找出来 // 对于injvm://协议已经在之前的逻辑中就已经生成invoke了 // if protocols not injvm checkRegistry if (!LOCAL_PROTOCOL.equalsIgnoreCase(getProtocol())){ checkRegistry(); // 加载注册中心地址 List us = loadRegistries(false); if (CollectionUtils.isNotEmpty(us)) { for (URL u : us) { URL monitorUrl = loadMonitor(u); if (monitorUrl != null) { map.put(MONITOR_KEY, URL.encode(monitorUrl.toFullString())); } // 对于注册中心地址都添加REFER_KEY urls.add(u.addParameterAndEncoded(REFER_KEY, StringUtils.toQueryString(map))); } } if (urls.isEmpty()) { throw new IllegalStateException(...); } } } // 如果只有一个url则直接refer得到一个invoker if (urls.size() == 1) { // RegistryProtocol.refer() 或者 DubboProtocol.refer() invoker = REF_PROTOCOL.refer(interfaceClass, urls.get(0)); } else { // 如果有多个url // 1. 根据每个url，refer得到对应的invoker // 2. 如果这多个urls中存在注册中心url，则把所有invoker整合为RegistryAwareClusterInvoker，该Invoker在调用时，会查看所有Invoker中是否有默认的，如果有则使用默认的Invoker，如果没有，则使用第一个Invoker // 2. 如果这多个urls中不存在注册中心url，则把所有invoker整合为FailoverCluster List<Invoker> invokers = new ArrayList<Invoker>(); URL registryURL = null; // 用来记录urls中最后一个注册中心url for (URL url : urls) { invokers.add(REF_PROTOCOL.refer(interfaceClass, url)); if (REGISTRY_PROTOCOL.equals(url.getProtocol())) { registryURL = url; // use last registry url } } // 如果存在注册中心地址 if (registryURL != null) { // registry url is available // use RegistryAwareCluster only when register's CLUSTER is available URL u = registryURL.addParameter(CLUSTER_KEY, RegistryAwareCluster.NAME); // StaticDirectory表示静态服务目录，里面的invokers是不会变的, 生成一个RegistryAwareCluster // The invoker wrap relation would be: RegistryAwareClusterInvoker(StaticDirectory) -> FailoverClusterInvoker(RegistryDirectory, will execute route) -> Invoker invoker = CLUSTER.join(new StaticDirectory(u, invokers)); } else { // not a registry url, must be direct invoke. // 如果不存在注册中心地址, 生成一个FailoverClusterInvoker invoker = CLUSTER.join(new StaticDirectory(invokers)); } } } if (shouldCheck() && !invoker.isAvailable()) { throw new IllegalStateException(...); } MetadataReportService metadataReportService = null; if ((metadataReportService = getMetadataReportService()) != null) { URL consumerURL = new URL(CONSUMER_PROTOCOL, map.remove(REGISTER_IP_KEY), 0, map.get(INTERFACE_KEY), map); metadataReportService.publishConsumer(consumerURL); } // create service proxy return (T) PROXY_FACTORY.getProxy(invoker);}【5】PROXY_FACTORY.getProxy等过程
//默认采用JavassistProxyFactory来产生代理对象public class JavassistProxyFactory extends AbstractProxyFactory { @Override @SuppressWarnings("unchecked") public T getProxy(Invoker invoker, Class[] interfaces) { return (T) Proxy.getProxy(interfaces).newInstance(new InvokerInvocationHandler(invoker)); } @Override public Invoker getInvoker(T proxy, Class type, URL url) { // TODO Wrapper cannot handle this scenario correctly: the classname contains '$' // 如果现在被代理的对象proxy本身就是一个已经被代理过的对象，那么则取代理类的Wrapper，否则取type（接口）的Wrapper // Wrapper是针对某个类或某个接口的包装类，通过wrapper对象可以更方便的去执行某个类或某个接口的方法 final Wrapper wrapper = Wrapper.getWrapper(proxy.getClass().getName().indexOf('$') < 0 ? proxy.getClass() : type); // proxy是服务实现类 // type是服务接口 // url是一个注册中心url，但同时也记录了 return new AbstractProxyInvoker(proxy, type, url) { @Override protected Object doInvoke(T proxy, String methodName, Class[] parameterTypes, Object[] arguments) throws Throwable { // 执行proxy的method方法 // 执行的proxy实例的方法 // 如果没有wrapper，则要通过原生的反射技术去获取Method对象，然后执行 return wrapper.invokeMethod(proxy, methodName, parameterTypes, arguments); } }; }}public class InvokerInvocationHandler implements InvocationHandler { private static final Logger logger = LoggerFactory.getLogger(InvokerInvocationHandler.class); private final Invoker invoker; public InvokerInvocationHandler(Invoker handler) { this.invoker = handler; } @Override public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { String methodName = method.getName(); Class[] parameterTypes = method.getParameterTypes(); if (method.getDeclaringClass() == Object.class) { return method.invoke(invoker, args); } if ("toString".equals(methodName) && parameterTypes.length == 0) { return invoker.toString(); } if ("hashCode".equals(methodName) && parameterTypes.length == 0) { return invoker.hashCode(); } if ("equals".equals(methodName) && parameterTypes.length == 1) { return invoker.equals(args[0]); } // 这里的recreate方法很重要，他会调用AppResponse的recreate方法， // 如果AppResponse对象中存在exception信息，则此方法中会throw这个异常 return invoker.invoke(new RpcInvocation(method, args)).recreate(); }}
【6】对URL的处理过程
1）从@Reference的url属性中配置多个url，然后采用字符串分割的形式拿出来，包装后塞入urls列表中
2）加载注册中心地址
protected List loadRegistries(boolean provider) { // check && override if necessary List registryList = new ArrayList(); if (CollectionUtils.isNotEmpty(registries)) { for (RegistryConfig config : registries) { String address = config.getAddress(); // 如果注册中心没有配地址，则地址为0.0.0.0 if (StringUtils.isEmpty(address)) { address = ANYHOST_VALUE; } // 如果注册中心的地址不是"N/A" if (!RegistryConfig.NO_AVAILABLE.equalsIgnoreCase(address)) { Map map = new HashMap(); // 把application中的参数放入map中，注意，map中的key是没有prefix的 appendParameters(map, application); // 把config中的参数放入map中，注意，map中的key是没有prefix的 // config是RegistryConfig，表示注册中心 appendParameters(map, config); // 此处path值固定为RegistryService.class.getName()，因为现在是在加载注册中心 map.put(PATH_KEY, RegistryService.class.getName()); // 把dubbo的版本信息和pid放入map中 appendRuntimeParameters(map); // 如果map中如果没有protocol，那么默认为dubbo if (!map.containsKey(PROTOCOL_KEY)) { map.put(PROTOCOL_KEY, DUBBO_PROTOCOL); } // 构造注册中心url，地址+参数 List urls = UrlUtils.parseURLs(address, map); for (URL url : urls) { url = URLBuilder.from(url) .addParameter(REGISTRY_KEY, url.getProtocol()) .setProtocol(REGISTRY_PROTOCOL) .build(); // 到此为止，url的内容大概为： // registry://127.0.0.1:2181/org.apache.dubbo.registry.RegistryService?application=dubbo-demo-annotation-provider&dubbo=2.0.2&pid=269936®istry=zookeeper×tamp=1584886077813 // 该url表示：使用registry协议调用org.apache.dubbo.registry.RegistryService服务 // 参数为application=dubbo-demo-annotation-provider&dubbo=2.0.2&pid=269936®istry=zookeeper×tamp=1584886077813 // 这里是服务提供者和服务消费者区别的逻辑 // 如果是服务提供者，获取register的值，如果为false，表示该服务不注册到注册中心 // 如果是服务消费者，获取subscribe的值，如果为false，表示该引入的服务不订阅注册中心中的数据 if ((provider && url.getParameter(REGISTER_KEY, true)) || (!provider && url.getParameter(SUBSCRIBE_KEY, true))) { registryList.add(url); } } } } } return registryList;}
【7】invoker的包装过程
1）前置说明
//RegistryProtocol实际上会被两个包装类包装ProtocolListenerWrapper与ProtocolFilterWrapper//如ProtocolListenerWrapper的protocol属性存放ProtocolFilterWrapper//ProtocolFilterWrapper的protocol属性存放RegistryProtocol//然后是Cluster，通过接口可知默认是FailoverCluster，但实际上还会有一个包装类MockClusterWrapper//MockClusterWrapper的cluster属性存放FailoverCluster或者RegistryAwareCluster
2）先是调用REF_PROTOCOL.refer
//ProtocolListenerWrapper的处理@Overridepublic Invoker refer(Class type, URL url) throws RpcException { if (REGISTRY_PROTOCOL.equals(url.getProtocol())) { return protocol.refer(type, url); } return new ListenerInvokerWrapper(protocol.refer(type, url), Collections.unmodifiableList( ExtensionLoader.getExtensionLoader(InvokerListener.class) .getActivateExtension(url, INVOKER_LISTENER_KEY)));}//ProtocolFilterWrapper的处理@Overridepublic Invoker refer(Class type, URL url) throws RpcException { if (REGISTRY_PROTOCOL.equals(url.getProtocol())) { return protocol.refer(type, url); } return buildInvokerChain(protocol.refer(type, url), REFERENCE_FILTER_KEY, CommonConstants.CONSUMER);}private static Invoker buildInvokerChain(final Invoker invoker, String key, String group) { Invoker last = invoker; // 根据url获取filter，根据url中的parameters取key为key的value所对应的filter，但是还会匹配group List filters = ExtensionLoader.getExtensionLoader(Filter.class).getActivateExtension(invoker.getUrl(), key, group); // ConsumerContextFilter--->FutureFilter--->MonitorFilter // ConsumerContextFilter用来设置RpcContext if (!filters.isEmpty()) { for (int i = filters.size() - 1; i >= 0; i--) { final Filter filter = filters.get(i); final Invoker next = last; last = new Invoker() { @Override public Class getInterface() { return invoker.getInterface(); } @Override public URL getUrl() { return invoker.getUrl(); } @Override public boolean isAvailable() { return invoker.isAvailable(); } @Override public Result invoke(Invocation invocation) throws RpcException { Result asyncResult; try { // 得到一个异步结果 asyncResult = filter.invoke(next, invocation); } catch (Exception e) { // onError callback if (filter instanceof ListenableFilter) { Filter.Listener listener = ((ListenableFilter) filter).listener(); if (listener != null) { listener.onError(e, invoker, invocation); } } throw e; } return asyncResult; } @Override public void destroy() { invoker.destroy(); } @Override public String toString() { return invoker.toString(); } }; } } return new CallbackRegistrationInvoker(last, filters);}//RegistryProtocol的处理public Invoker refer(Class type, URL url) throws RpcException { // 从registry://的url中获取对应的注册中心，比如zookeeper， 默认为dubbo，dubbo提供了自带的注册中心实现 // url由 registry:// 改变为---> zookeeper:// url = URLBuilder.from(url) .setProtocol(url.getParameter(REGISTRY_KEY, DEFAULT_REGISTRY)) .removeParameter(REGISTRY_KEY) .build(); // 拿到注册中心实现，ZookeeperRegistry Registry registry = registryFactory.getRegistry(url); // 下面这个代码，通过过git历史提交记录是用来解决SimpleRegistry不可用的问题 if (RegistryService.class.equals(type)) { return proxyFactory.getInvoker((T) registry, type, url); } // qs表示 queryString, 表示url中的参数，表示消费者引入服务时所配置的参数 Map qs = StringUtils.parseQueryString(url.getParameterAndDecoded(REFER_KEY)); // group="a,b" or group="*" String group = qs.get(GROUP_KEY); if (group != null && group.length() > 0) { if ((COMMA_SPLIT_PATTERN.split(group)).length > 1 || "*".equals(group)) { // group有多个值，这里的cluster为MergeableCluster return doRefer(getMergeableCluster(), registry, type, url); } } // 这里的cluster是cluster的Adaptive对象 return doRefer(cluster, registry, type, url);}private Invoker doRefer(Cluster cluster, Registry registry, Class type, URL url) { // RegistryDirectory表示动态服务目录，会和注册中心的数据保持同步 // type表示一个服务对应一个RegistryDirectory，url表示注册中心地址 // 在消费端，最核心的就是RegistryDirectory RegistryDirectory directory = new RegistryDirectory(type, url); directory.setRegistry(registry); directory.setProtocol(protocol); // all attributes of REFER_KEY // 引入服务所配置的参数 Map parameters = new HashMap(directory.getUrl().getParameters()); // 消费者url URL subscribeUrl = new URL(CONSUMER_PROTOCOL, parameters.remove(REGISTER_IP_KEY), 0, type.getName(), parameters); if (!ANY_VALUE.equals(url.getServiceInterface()) && url.getParameter(REGISTER_KEY, true)) { directory.setRegisteredConsumerUrl(getRegisteredConsumerUrl(subscribeUrl, url)); // 注册简化后的消费url registry.register(directory.getRegisteredConsumerUrl()); } // 构造路由链,路由链会在引入服务时按路由条件进行过滤 // 路由链是动态服务目录中的一个属性，通过路由链可以过滤某些服务提供者 directory.buildRouterChain(subscribeUrl); // 服务目录需要订阅的几个路径 // 当前所引入的服务的消费应用目录：/dubbo/config/dubbo/dubbo-demo-consumer-application.configurators // 当前所引入的服务的动态配置目录：/dubbo/config/dubbo/org.apache.dubbo.demo.DemoService:1.1.1:g1.configurators // 当前所引入的服务的提供者目录：/dubbo/org.apache.dubbo.demo.DemoService/providers // 当前所引入的服务的老版本动态配置目录：/dubbo/org.apache.dubbo.demo.DemoService/configurators // 当前所引入的服务的老版本路由器目录：/dubbo/org.apache.dubbo.demo.DemoService/routers directory.subscribe(subscribeUrl.addParameter(CATEGORY_KEY, PROVIDERS_CATEGORY + "," + CONFIGURATORS_CATEGORY + "," + ROUTERS_CATEGORY)); // 利用传进来的cluster，join得到invoker, Invoker invoker = cluster.join(directory); ProviderConsumerRegTable.registerConsumer(invoker, url, subscribeUrl, directory); return invoker;}
3）再是调用CLUSTER.join
//MockClusterWrapper的处理public class MockClusterWrapper implements Cluster { private Cluster cluster; public MockClusterWrapper(Cluster cluster) { this.cluster = cluster; } @Override public Invoker join(Directory directory) throws RpcException { return new MockClusterInvoker(directory, this.cluster.join(directory)); }}//有注册中心，RegistryAwareCluster的处理public class RegistryAwareCluster implements Cluster { public final static String NAME = "registryaware"; @Override public Invoker join(Directory directory) throws RpcException { return new RegistryAwareClusterInvoker(directory); }}//没有注册中心，FailoverCluster的处理public class FailoverCluster implements Cluster { public final static String NAME = "failover"; @Override public Invoker join(Directory directory) throws RpcException { return new FailoverClusterInvoker(directory); }}
【8】RegistryProtocol里面的监听过程
//directory.subscribe(subscribeUrl.addParameter(CATEGORY_KEY, PROVIDERS_CATEGORY + "," + CONFIGURATORS_CATEGORY + "," + ROUTERS_CATEGORY));//RegistryDirectory类#subscribe方法public void subscribe(URL url) { setConsumerUrl(url); CONSUMER_CONFIGURATION_LISTENER.addNotifyListener(this); // 监听consumer应用 serviceConfigurationListener = new ReferenceConfigurationListener(this, url); // 监听所引入的服务的动态配置 registry.subscribe(url, this); //老版本的监听}//FailbackRegistry类#subscribe方法@Overridepublic void subscribe(URL url, NotifyListener listener) { super.subscribe(url, listener); removeFailedSubscribed(url, listener); try { // 核心方法 doSubscribe(url, listener); } catch (Exception e) { Throwable t = e; List urls = getCacheUrls(url); if (CollectionUtils.isNotEmpty(urls)) { notify(url, listener, urls); logger.error(...); } else { // If the startup detection is opened, the Exception is thrown directly. boolean check = getUrl().getParameter(Constants.CHECK_KEY, true) && url.getParameter(Constants.CHECK_KEY, true); boolean skipFailback = t instanceof SkipFailbackWrapperException; if (check || skipFailback) { if (skipFailback) { t = t.getCause(); } throw new IllegalStateException("Failed to subscribe " + url + ", cause: " + t.getMessage(), t); } else { logger.error("Failed to subscribe " + url + ", waiting for retry, cause: " + t.getMessage(), t); } } // Record a failed registration request to a failed list, retry regularly // 添加listener，向zk添加监听器时如果报错了，那么会把这个listener添加到failedSubscribed中，并会定时重试（重新注册listener） addFailedSubscribed(url, listener); }}// 父类AbstractRegistry#subscribe方法// 把listener添加到subscribed中，subscribed是一个map, 存的是URL:Set@Overridepublic void subscribe(URL url, NotifyListener listener) { if (url == null) { throw new IllegalArgumentException("subscribe url == null"); } if (listener == null) { throw new IllegalArgumentException("subscribe listener == null"); } if (logger.isInfoEnabled()) { logger.info("Subscribe: " + url); } Set listeners = subscribed.computeIfAbsent(url, n -> new ConcurrentHashSet()); listeners.add(listener);}// 进行订阅，先看父类的subscribe方法@Overridepublic void doSubscribe(final URL url, final NotifyListener listener) { try { if (ANY_VALUE.equals(url.getServiceInterface())) { // 订阅所有服务 String root = toRootPath(); ConcurrentMap listeners = zkListeners.get(url); if (listeners == null) { zkListeners.putIfAbsent(url, new ConcurrentHashMap()); listeners = zkListeners.get(url); } ChildListener zkListener = listeners.get(listener); if (zkListener == null) { listeners.putIfAbsent(listener, (parentPath, currentChilds) -> { for (String child : currentChilds) { child = URL.decode(child); if (!anyServices.contains(child)) { anyServices.add(child); subscribe(url.setPath(child).addParameters(INTERFACE_KEY, child, Constants.CHECK_KEY, String.valueOf(false)), listener); } } }); zkListener = listeners.get(listener); } zkClient.create(root, false); List services = zkClient.addChildListener(root, zkListener); if (CollectionUtils.isNotEmpty(services)) { for (String service : services) { service = URL.decode(service); anyServices.add(service); subscribe(url.setPath(service).addParameters(INTERFACE_KEY, service, Constants.CHECK_KEY, String.valueOf(false)), listener); } } } else { // 单独订阅某一个服务 List urls = new ArrayList(); // 得到真正要监听的zk上的路径, for (String path : toCategoriesPath(url)) { // 根据监听地址去拿listeners，如果没有则生成 ConcurrentMap listeners = zkListeners.get(url); if (listeners == null) { zkListeners.putIfAbsent(url, new ConcurrentHashMap()); listeners = zkListeners.get(url); } // 一个NotifyListener对应一个ChildListener ChildListener zkListener = listeners.get(listener); if (zkListener == null) { // lambda表达式就是监听逻辑， parentPath表示父path,currentChilds表示当前拥有的child, 会调用notify方法进行实际的处理 listeners.putIfAbsent(listener, (parentPath, currentChilds) -> ZookeeperRegistry.this.notify(url, listener, toUrlsWithEmpty(url, parentPath, currentChilds))); zkListener = listeners.get(listener); } // 创建zk上路径 zkClient.create(path, false); // 添加真正跟zk相关的ChildListener,ChildListener中的逻辑就是监听到zk上数据发生了变化后会触发的逻辑 List children = zkClient.addChildListener(path, zkListener); if (children != null) { urls.addAll(toUrlsWithEmpty(url, path, children)); } } // 这里的urls就是从现在所引入的服务的目录下查到的url，比如下面这个三个目录下的路径// "/dubbo/org.apache.dubbo.demo.DemoService/providers"// "/dubbo/org.apache.dubbo.demo.DemoService/configurators"// "/dubbo/org.apache.dubbo.demo.DemoService/routers" notify(url, listener, urls); } } catch (Throwable e) { throw new RpcException("Failed to subscribe " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e); }}
【9】监听器触发时
//RegistryDirectory类#notify方法public synchronized void notify(List urls) { Map<String, List> categoryUrls = urls.stream() .filter(Objects::nonNull) .filter(this::isValidCategory) .filter(this::isNotCompatibleFor26x) .collect(Collectors.groupingBy(url -> { if (UrlUtils.isConfigurator(url)) { return CONFIGURATORS_CATEGORY; } else if (UrlUtils.isRoute(url)) { return ROUTERS_CATEGORY; } else if (UrlUtils.isProvider(url)) { return PROVIDERS_CATEGORY; } return ""; })); // 获取动态配置URL，生成configurators List configuratorURLs = categoryUrls.getOrDefault(CONFIGURATORS_CATEGORY, Collections.emptyList()); this.configurators = Configurator.toConfigurators(configuratorURLs).orElse(this.configurators); // 获取老版本路由URL，生成Router，并添加到路由链中 List routerURLs = categoryUrls.getOrDefault(ROUTERS_CATEGORY, Collections.emptyList()); toRouters(routerURLs).ifPresent(this::addRouters); // 获取服务提供者URL List providerURLs = categoryUrls.getOrDefault(PROVIDERS_CATEGORY, Collections.emptyList()); refreshOverrideAndInvoker(providerURLs);}private void refreshOverrideAndInvoker(List urls) { // mock zookeeper://xxx?mock=return null overrideDirectoryUrl(); refreshInvoker(urls);}// 利用动态配置重写服务目录地址private void overrideDirectoryUrl() { // merge override parameters this.overrideDirectoryUrl = directoryUrl; List localConfigurators = this.configurators; // local reference doOverrideUrl(localConfigurators); List localAppDynamicConfigurators = CONSUMER_CONFIGURATION_LISTENER.getConfigurators(); // local reference doOverrideUrl(localAppDynamicConfigurators); if (serviceConfigurationListener != null) { List localDynamicConfigurators = serviceConfigurationListener.getConfigurators(); // local reference doOverrideUrl(localDynamicConfigurators); }}private void refreshInvoker(List invokerUrls) { Assert.notNull(invokerUrls, "invokerUrls should not be null"); if (invokerUrls.size() == 1 && invokerUrls.get(0) != null && EMPTY_PROTOCOL.equals(invokerUrls.get(0).getProtocol())) { this.forbidden = true; // Forbid to access this.invokers = Collections.emptyList(); routerChain.setInvokers(this.invokers); destroyAllInvokers(); // Close all invokers } else { this.forbidden = false; // Allow to access Map<String, Invoker> oldUrlInvokerMap = this.urlInvokerMap; // local reference if (invokerUrls == Collections.emptyList()) { invokerUrls = new ArrayList(); } if (invokerUrls.isEmpty() && this.cachedInvokerUrls != null) { invokerUrls.addAll(this.cachedInvokerUrls); } else { this.cachedInvokerUrls = new HashSet(); this.cachedInvokerUrls.addAll(invokerUrls);//Cached invoker urls, convenient for comparison } if (invokerUrls.isEmpty()) { return; } // 这里会先按Protocol进行过滤，并且调用DubboProtocol.refer方法得到DubboInvoker Map<String, Invoker> newUrlInvokerMap = toInvokers(invokerUrls);// Translate url list to Invoker map /** * If the calculation is wrong, it is not processed. * * 1. The protocol configured by the client is inconsistent with the protocol of the server. * eg: consumer protocol = dubbo, provider only has other protocol services(rest). * 2. The registration center is not robust and pushes illegal specification data. * */ if (CollectionUtils.isEmptyMap(newUrlInvokerMap)) { logger.error(...); return; } List<Invoker> newInvokers = Collections.unmodifiableList(new ArrayList(newUrlInvokerMap.values())); // pre-route and build cache, notice that route cache should build on original Invoker list. // toMergeMethodInvokerMap() will wrap some invokers having different groups, those wrapped invokers not should be routed. // 得到了所引入的服务Invoker之后，把它们设置到路由链中去，在调用时使用，并且会调用TagRouter的notify方法 routerChain.setInvokers(newInvokers); this.invokers = multiGroup ? toMergeInvokerList(newInvokers) : newInvokers; this.urlInvokerMap = newUrlInvokerMap; try { destroyUnusedInvokers(oldUrlInvokerMap, newUrlInvokerMap); // Close the unused Invoker } catch (Exception e) { logger.warn("destroyUnusedInvokers error. ", e); } }}private Map<String, Invoker> toInvokers(List urls) { Map<String, Invoker> newUrlInvokerMap = new HashMap(); if (urls == null || urls.isEmpty()) { return newUrlInvokerMap; } Set keys = new HashSet(); String queryProtocols = this.queryMap.get(PROTOCOL_KEY); // 遍历当前服务所有的服务提供者URL for (URL providerUrl : urls) { // If protocol is configured at the reference side, only the matching protocol is selected if (queryProtocols != null && queryProtocols.length() > 0) { boolean accept = false; String[] acceptProtocols = queryProtocols.split(","); // 当前消费者如果手动配置了Protocol，那么则进行匹配 for (String acceptProtocol : acceptProtocols) { if (providerUrl.getProtocol().equals(acceptProtocol)) { accept = true; break; } } if (!accept) { continue; } } if (EMPTY_PROTOCOL.equals(providerUrl.getProtocol())) { continue; } // 当前Protocol是否在应用中存在对应的扩展点 if (!ExtensionLoader.getExtensionLoader(Protocol.class).hasExtension(providerUrl.getProtocol())) { logger.error(...); continue; } URL url = mergeUrl(providerUrl); String key = url.toFullString(); // The parameter urls are sorted if (keys.contains(key)) { // Repeated url continue; } keys.add(key); // Cache key is url that does not merge with consumer side parameters, regardless of how the consumer combines parameters, if the server url changes, then refer again Map<String, Invoker> localUrlInvokerMap = this.urlInvokerMap; // local reference Invoker invoker = localUrlInvokerMap == null ? null : localUrlInvokerMap.get(key); // 如果当前服务提供者URL没有生产过Invoker if (invoker == null) { // Not in the cache, refer again try { boolean enabled = true; if (url.hasParameter(DISABLED_KEY)) { enabled = !url.getParameter(DISABLED_KEY, false); } else { enabled = url.getParameter(ENABLED_KEY, true); } if (enabled) { // 调用Protocol的refer方法得到一个Invoker invoker = new InvokerDelegate(protocol.refer(serviceType, url), url, providerUrl); } } catch (Throwable t) { logger.error(...); } if (invoker != null) { // Put new invoker in cache newUrlInvokerMap.put(key, invoker); } } else { newUrlInvokerMap.put(key, invoker); } } keys.clear(); return newUrlInvokerMap;}