Dubbo源码—1.服务发布的主要流程

大纲
1.Dubbo 2.7和3.x版本的区别
2.Dubbo服务的基本流程和启动入口
3.Dubbo服务发布的主流程
4.服务发布时执行相关组件的初始化
5.服务发布时执行的服务实例刷新操作
6.服务发布时执行的服务实例初始化操作
7.服务发布时执行的服务实例发布操作
8.执行服务实例发布操作时的主流程
9.服务发布过程中ProxyFactory生成Invoker
10.服务发布过程中Protocol组件发布Invoker
11.服务发布过程中NettyServer的构造流程
12.服务发布过程中RegistryProtocol的服务注册
13.Dubbo服务发布的完整流程总结
 
1.Dubbo 2.7和3.x版本的区别
区别一：后者引入了ModuleDeployer组件专门做服务启动时的初始化工作。将原来的注册中心拆分为三大中心：注册中心、配置中心、元数据中心。
 
区别二：后者很多地方使用了Double Check来代替前者对方法加Synchronized锁，大量采用了Double Check + Volatile + Static来实现单例模式。
 
区别三：后者引入了MigrationRuleListener、MigrationRuleHandler、MigrationInvoker，引入DynamicDirectory代替RegistryDirectory。
 
2.Dubbo服务的基本流程和启动入口
(1)Dubbo服务的基本流程
(2)Provider启动入口
(3)Consumer启动入口
 
(1)Dubbo服务的基本流程

(2)Provider启动入口

1. public class Application {
2.     public static void main(String[] args) throws Exception {
3.         //Service和ServiceConfig是什么
4.         //Service是一个服务，每个服务可能会包含多个接口
5.         //ServiceConfig便是针对这个服务的一些配置
6.         //下面传入的泛型DemoServiceImpl便是服务接口的实现
7.         ServiceConfig<DemoServiceImpl> service = new ServiceConfig<>();
9.         //设置服务暴露出去的接口
10.         service.setInterface(DemoService.class);
12.         //设置暴露出去的接口的实现类
13.         service.setRef(new DemoServiceImpl());
15.         //服务名称，可以在服务框架里进行定位
16.         service.setApplication(new ApplicationConfig("dubbo-demo-api-provider"));
18.         //所有的RPC框架，必须要和注册中心配合使用，服务启动后必须向注册中心进行注册
19.         //注册中心可以知道每个服务有几个实例，每个实例在哪台服务器上
20.         //进行服务调用时，要先找注册中心咨询要调用的服务有几个实例，分别都在什么机器上
21.         //下面便是设置ZooKeeper作为注册中心的地址
22.         service.setRegistry(new RegistryConfig("zookeeper://127.0.0.1:2181"));
23.         
24.         //设置元数据上报的地方
25.         //Dubbo服务实例启动后，会有自己的元数据，需要上报到一个地方进行管理，比如zookeeper
26.         service.setMetadataReportConfig(new MetadataReportConfig("zookeeper://127.0.0.1:2181"));
28.         //配置完毕后，调用ServiceConfig的export()方法启动网络监听程序
29.         //当接收到调用请求时，该网络监听程序会建立网络连接进行通信
30.         //接收按照协议封装的请求数据，该网络监听程序会执行RPC调用
31.         //此外，ServiceConfig的export()方法还会把自己作为一个服务实例注册到zk里
32.         service.export();
34.         System.out.println("dubbo service started");
35.         new CountDownLatch(1).await();
36.     }
37. }

复制代码

(3)Consumer启动入口

1. public class Application {
2.     public static void main(String[] args) throws Exception {
3.         //Reference和ReferenceConfig是什么
4.         //Reference是一个引用，是对Provider端的一个服务实例的引用
5.         //ReferenceConfig这个服务实例的引用的一些配置
6.         //通过泛型传递了这个服务实例对外暴露的接口
7.         ReferenceConfig<DemoService> reference = new ReferenceConfig<>();
8.         
9.         //设置应用名称
10.         reference.setApplication(new ApplicationConfig("dubbo-demo-api-consumer"));
12.         //设置注册中心的地址，默认是ZooKeeper
13.         reference.setRegistry(new RegistryConfig("zookeeper://127.0.0.1:2181"));
15.         //设置元数据上报地址
16.         reference.setMetadataReportConfig(new MetadataReportConfig("zookeeper://127.0.0.1:2181"));
18.         //设置要调用的服务的接口
19.         reference.setInterface(DemoService.class);
21.         //直接通过ReferenceConfig的get()方法来拿到一个DemoService接口
22.         //它是一个动态代理接口，只要被调用，便会通过底层调用Provider服务实例的对应接口
23.         DemoService service = reference.get();
24.         String message = service.sayHello("dubbo");
25.         System.out.println(message);
26.         Thread.sleep(10000000L);
27.     }
28. }

复制代码

 
3.Dubbo服务发布的主流程
ServiceConfig的export()方法在进行服务发布时，首先会初始化相关组件，然后刷新服务实例，接着初始化服务实例，最后发布服务实例。

1. public class ServiceConfig<T> extends ServiceConfigBase<T> {
2.     ...
3.     @Override
4.     public void export() {
5.         if (this.exported) {
6.             return;
7.         }
9.         //prepare for export
10.         //对比Dubbo 2.6.x和2.7.x源码，Dubbo 3.0的变动还是有点大的，比如这里使用了ModuleDeployer组件
11.         //Dubbo服务实例内部会有很多的代码组件，通过ModuleDeployer便可以避免零零散散的去调用和初始化
13.         //1.执行相关组件的初始化
14.         //通过获取到的ModuleDeployer来对相关组件进行初始化
15.         //比如会对MetadataReport元数据上报组件进行构建和初始化，以及启动(建立跟zk的连接)
16.         getScopeModel().getDeployer().start();
18.         synchronized (this) {
19.             if (this.exported) {
20.                 return;
21.             }
23.             //2.执行服务实例的刷新操作
24.             //也就是刷新ProviderConfig->MethodConfig->ArgumentConfig
25.             if (!this.isRefreshed()) {
26.                 this.refresh();
27.             }
28.             if (this.shouldExport()) {
29.                 //3.执行服务实例的初始化
30.                 //也就是会把Metadata元数据给准备好，后续可以进行元数据上报
31.                 this.init();
33.                 //这是Dubbo服务实例的延迟发布的特性
34.                 //如果设置了Dubbo服务实例是延迟发布的，当调用了export()方法后，就会进入这里
35.                 //在延迟指定的时间后，再去进行服务的发布
36.                 if (shouldDelay()) {
37.                     //延迟发布
38.                     doDelayExport();
39.                 } else {
40.                     //立即发布
41.                     //4.执行服务实例的发布
42.                     //这里可以作为服务发布的直接入口
43.                     doExport();
44.                 }
45.             }
46.         }
47.     }
48.     ...
49. }

复制代码

 
4.服务发布时执行相关组件的初始化

1. public class DefaultModuleDeployer extends AbstractDeployer<ModuleModel> implements ModuleDeployer {
2.     //已经完成发布的服务实例集合
3.     private List<ServiceConfigBase<?>> exportedServices = new ArrayList<>();
5.     //下面这些组件，本身都是跟model组件体系强关联的
6.     private ModuleModel moduleModel;
8.     //父级ApplicationDeployer组件
9.     private ApplicationDeployer applicationDeployer;
10.     ...
12.     @Override
13.     public Future start() throws IllegalStateException {
14.         //initialize，maybe deadlock applicationDeployer lock & moduleDeployer lock
15.         //调用DefaultApplicationDeployer的initialize()方法
16.         applicationDeployer.initialize();
17.         return startSync();
18.     }
20.     @Override
21.     public void prepare() {
22.         //module这个层级是application层级的下层
23.         //application层级是framework层级的下层
24.         applicationDeployer.initialize();
25.         this.initialize();
26.     }
27.     ...
28. }
30. public class DefaultApplicationDeployer extends AbstractDeployer implements ApplicationDeployer {
31.     private final DubboShutdownHook dubboShutdownHook;
32.     ...
34.     @Override
35.     public void initialize() {
36.         if (initialized) {
37.             return;
38.         }
39.         //ApplicationDeployer组件，可能会被多线程并发访问
40.         //Ensure that the initialization is completed when concurrent calls
41.         synchronized (startLock) {
42.             if (initialized) {
43.                 return;
44.             }
45.             //注册退出时需要进行资源销毁的ShutdownHook
46.             //register shutdown hook
47.             registerShutdownHook();
49.             //启动配置中心ConfigCenter
50.             //动配置中心是专门存放配置信息的
51.             startConfigCenter();
53.             //加载应用配置
54.             loadApplicationConfigs();
56.             //初始化ModuleDeployer
57.             initModuleDeployers();
59.             //启动元数据中心MetadataCenter
60.             //元数据中心是专门存放发布d的服务实例信息的
61.             startMetadataCenter();
63.             initialized = true;
65.             if (logger.isInfoEnabled()) {
66.                 logger.info(getIdentifier() + " has been initialized!");
67.             }
68.         }
70.         //老的Dubbo版本只有注册中心的概念
71.         //后来随着版本的迭代和演进，出现了配置中心、元数据中心等这些用于解耦的概念
72.         //虽然Dubbo的服务实例信息、配置信息、元数据信息，都可以放在zk里
73.         //但如果把一个服务实例的各种数据和信息都存储在zk里，那么这些数据和信息会严重耦合在一起
74.         //下面从扩展性和可用性两个层面来分析，如果把各种数据都耦合放在一个zk里可能会出现的问题
75.         //(1)扩展性
76.         //在一个大规模的微服务系统里，服务本身可能都有上百个，服务实例可能有几百上千个，所以服务实例数据可能就很多
77.         //但每个服务实例关联的配置数据，可能不是太多，尤其是元数据可能也不是太多
78.         //而这里的扩展性不是指功能上的扩展，而是指数据上的扩展
79.         //随着服务实例数据在膨胀，配置数据和元数据虽然可能也在增加，但是增加的速度可能比不上服务实例数据
80.         //此时就需要对注册中心进行扩容或者更换一个系统去存储
81.         //而在这个过程之中，由于所有数据耦合在一起了，牵一发而动全身，不好扩展了
82.         //这就是多种不同类型的数据耦合在一起时会出现的痛点，也就是数据耦合导致数据扩展性差的问题
83.         //(2)可用性
84.         //由于注册数据、配置数据、元数据都放在一个地方比如zk
85.         //那么一旦zk出现了故障，这三种数据就一起没了，这就是可用性问题
86.         //因此，Dubbo3的架构设计会对不同类型的数据进行分离，形成了三种数据：注册数据、元数据、配置数据
87.         //于是就有了三大中心：注册中心、配置中心、元数据中心，这样就可以把三种不同类型的数据，放到不同的地方去
88.         //在扩展性方面，当服务实例数据太多时要进行扩容或者切换存储技术，此时对另外两种数据是没有直接影响的
89.         //在可用性方面，一旦作为注册中心的zk突然挂了，此时配置中心可能是Nacos，对它来说也没有直接影响的
90.     }
92.     private void registerShutdownHook() {
93.         dubboShutdownHook.register();
94.     }
95.     ...
96. }

复制代码

 
6.服务发布时执行的服务实例初始化操作

1. public class DefaultApplicationDeployer extends AbstractDeployer implements ApplicationDeployer {
2.     ...
3.     //启动元数据中心
4.     private void startMetadataCenter() {
5.         //先分析元数据中心是否需要用注册中心来做元数据中心
6.         useRegistryAsMetadataCenterIfNecessary();
7.         ApplicationConfig applicationConfig = getApplication();
9.         //获取元数据类型metadataType
10.         String metadataType = applicationConfig.getMetadataType();
12.         //进行元数据的配置，里面包含了MetadataReport的配置
13.         Collection<MetadataReportConfig> metadataReportConfigs = configManager.getMetadataConfigs();
14.         if (CollectionUtils.isEmpty(metadataReportConfigs)) {
15.             return;
16.         }
18.         //从applicationModel中获取一个BeanFactory
19.         //再根据BeanFactory获取一个元数据上报组件的实例MetadataReportInstance
20.         MetadataReportInstance metadataReportInstance = applicationModel.getBeanFactory().getBean(MetadataReportInstance.class);
21.         List<MetadataReportConfig> validMetadataReportConfigs = new ArrayList<>(metadataReportConfigs.size());
22.         for (MetadataReportConfig metadataReportConfig : metadataReportConfigs) {
23.             ConfigValidationUtils.validateMetadataConfig(metadataReportConfig);
24.             validMetadataReportConfigs.add(metadataReportConfig);
25.         }
27.         //对于唯一的一个MetadataReport，会在这里进行初始化
28.         //把配置的metadataReport地址和config传递进init()方法进行初始化
29.         metadataReportInstance.init(validMetadataReportConfigs);
30.         if (!metadataReportInstance.inited()) {
31.             throw new IllegalStateException(String.format("%s MetadataConfigs found, but none of them is valid.", metadataReportConfigs.size()));
32.         }
33.         //所以MetadataReport的启动
34.         //其实就是根据配置去获取对应的BeanFactory，然后通过BeanFactory创建出对应的MetadataReport实例
35.         //最后根据配置对元数据上报组件的实例MetadataReportInstance进行初始化
36.     }
37.     ...
38. }
40. public class MetadataReportInstance implements Disposable {
41.     ...
42.     public void init(List<MetadataReportConfig> metadataReportConfigs) {
43.         if (!init.compareAndSet(false, true)) {
44.             return;
45.         }
47.         this.metadataType = applicationModel.getApplicationConfigManager().getApplicationOrElseThrow().getMetadataType();
48.         if (metadataType == null) {
49.             this.metadataType = DEFAULT_METADATA_STORAGE_TYPE;
50.         }
52.         //在这里会通过SPI机制的adaptive自适应，生成一个代理类
53.         //底层会通过自适应的机制，根据url里的参数去拿到对应的实现类，来调用它的方法
54.         //如果使用zk作为元数据中心，那么拿到的应该是一个ZooKeeperMetadataReportFactory
55.         MetadataReportFactory metadataReportFactory = applicationModel.getExtensionLoader(MetadataReportFactory.class).getAdaptiveExtension();
56.         for (MetadataReportConfig metadataReportConfig : metadataReportConfigs) {
57.             init(metadataReportConfig, metadataReportFactory);
58.         }
59.     }
61.     private void init(MetadataReportConfig config, MetadataReportFactory metadataReportFactory) {
62.         ...
63.         //这种url一般来说是针对zk的url地址
64.         MetadataReport metadataReport = metadataReportFactory.getMetadataReport(url);
65.         if (metadataReport != null) {
66.             metadataReports.put(relatedRegistryId, metadataReport);
67.         }
68.     }
69.     ...
70. }
72. public abstract class AbstractMetadataReportFactory implements MetadataReportFactory {
73.     ...
74.     @Override
75.     public MetadataReport getMetadataReport(URL url) {
76.         ...
77.         metadataReport = createMetadataReport(url);
78.         ...
79.     }
81.     protected abstract MetadataReport createMetadataReport(URL url);
82.     ...
83. }
85. public class ZookeeperMetadataReportFactory extends AbstractMetadataReportFactory {
86.     ...
87.     @Override
88.     public MetadataReport createMetadataReport(URL url) {
89.         return new ZookeeperMetadataReport(url, zookeeperTransporter);
90.     }
91.     ...
92. }
94. public class ZookeeperMetadataReport extends AbstractMetadataReport {
95.     private final String root;
96.     ZookeeperClient zkClient;
97.     ...
99.     public ZookeeperMetadataReport(URL url, ZookeeperTransporter zookeeperTransporter) {
100.         super(url);
101.         ...
102.         zkClient = zookeeperTransporter.connect(url);
103.     }
104.     ...
105. }

复制代码

 
7.服务发布时执行的服务实例发布操作
首先调用ServiceConfig的doExportUrls()方法发布服务，然后再调用其exported()方法进行服务发布后的处理，比如打印日志和回调监听器。

1. public class ServiceConfig<T> extends ServiceConfigBase<T> {
2.     ...
3.     @Override
4.     public void export() {
5.         ...
6.         //1.执行相关组件的初始化
7.         //通过获取到的ModuleDeployer来对相关组件进行初始化
8.         //比如会对MetadataReport元数据上报组件进行构建和初始化，以及启动(建立跟zk的连接)
9.         getScopeModel().getDeployer().start();
10.         synchronized (this) {
11.             if (this.exported) {
12.                 return;
13.             }
14.             //2.执行服务实例的刷新操作
15.             //也就是刷新ProviderConfig->MethodConfig->ArgumentConfig
16.             if (!this.isRefreshed()) {
17.                 //执行AbstractConfig的refresh()方法
18.                 this.refresh();
19.             }
20.             ...
21.         }
22.     }
23.     ...
24. }
26. public abstract class AbstractConfig implements Serializable {
27.     ...
28.     public void refresh() {
29.         //check and init before do refresh
30.         //调用AbstractConfig的子类ServiceConfigBase的preProcessRefresh()方法
31.         //初始化一个ProviderConfig，也就是Provider服务实例
32.         preProcessRefresh();
34.         //Model组件体系对Dubbo的运行很关键，可以认为它是SPI机制使用的入口
35.         //而ScopeModel是Model组件体系的一个基础，ScopeModel类型是可以转换为ModuleModel、ApplicationModel
36.         //比如像ModuleServiceRepository、ModelEnvironment、BeanFactory等很多通用的组件都可以通过ScopeModel去获取
37.         Environment environment = getScopeModel().getModelEnvironment();//获取Environment对象
38.         List<Map<String, String>> configurationMaps = environment.getConfigurationMaps();
40.         //Search props starts with PREFIX in order
41.         //接下来要获取和拼接preferredPrefix
42.         String preferredPrefix = null;
43.         List<String> prefixes = getPrefixes();
44.         for (String prefix : prefixes) {
45.             if (ConfigurationUtils.hasSubProperties(configurationMaps, prefix)) {
46.                 preferredPrefix = prefix;
47.                 break;
48.             }
49.         }
50.         if (preferredPrefix == null) {
51.             preferredPrefix = prefixes.get(0);
52.         }
54.         ...
55.         //使用反射注入需要的方法
56.         assignProperties(this, environment, subProperties, subPropsConfiguration);
58.         //process extra refresh of subclass, e.g. refresh method configs
59.         //调用AbstractInterfaceConfig的processExtraRefresh()方法
60.         //该方法中preferredPrefix是关键，它的值可能是：dubbo.service.org.apache.dubbo.demo.DemoService
61.         //其中dubbo.service代表dubbo服务名称的一个固定前缀，属于固定拼接的
62.         //而中间的org.apache.dubbo.demo，则是从服务接口所在包名里截取出来的，并且最后会加上服务接口的接口名
63.         //所以preferredPrefix会作为当前dubbo服务的全限定的名字
64.         //而这段refresh的代码的作用，就是处理这个preferredPrefix以及其他相关的配置信息
65.         processExtraRefresh(preferredPrefix, subPropsConfiguration);
67.         postProcessRefresh();
68.         refreshed.set(true);
69.     }
71.     protected void preProcessRefresh() {
72.         // pre-process refresh
73.     }
75.     protected void processExtraRefresh(String preferredPrefix, InmemoryConfiguration subPropsConfiguration) {
76.         // process extra refresh
77.     }
78.       
79.     protected void postProcessRefresh() {
80.         // post-process refresh
81.         checkDefault();
82.     }
83.     ...
84. }
86. public abstract class AbstractInterfaceConfig extends AbstractMethodConfig {
87.     private List<MethodConfig> methods;
88.     ...
90.     //该方法会通过反射技术，对需要发布的服务的接口方法和参数封装成MethodConfig、ArgumentConfig
91.     @Override
92.     protected void processExtraRefresh(String preferredPrefix, InmemoryConfiguration subPropsConfiguration) {
93.         if (StringUtils.hasText(interfaceName)) {
94.             //通过反射技术获取需要发布的服务的接口
95.             Class<?> interfaceClass;
96.             interfaceClass = ClassUtils.forName(interfaceName);
97.             ...
99.             //Auto create MethodConfig/ArgumentConfig according to config props
100.             Map<String, String> configProperties = subPropsConfiguration.getProperties();
101.             //获取需要发布的服务的接口的所有方法
102.             Method[] methods;
103.             methods = interfaceClass.getMethods();
105.             //接下来对需要发布的服务的接口方法进行处理
106.             //整理出MethodConfig对象及其对应的ArgumentConfig对象
107.             //接口里的每个方法都要创建一个MethodConfig
108.             //方法里的每一个参数都要创建一个ArgumentConfig
109.             for (Method method : methods) {
110.                 //因为服务端每次处理客户端调用时，不可能都通过反射来获取method和args的情况
111.                 //所以在刚开始启动时就需要对接口进行解析，将所有的method和args整理到methods属性中
112.                 if (ConfigurationUtils.hasSubProperties(configProperties, method.getName())) {
113.                     MethodConfig methodConfig = getMethodByName(method.getName());
114.                     //Add method config if not found
115.                     if (methodConfig == null) {
116.                         //需要发布的的服务的每个方法，都创建一个MethodConfig对象
117.                         methodConfig = new MethodConfig();
118.                         methodConfig.setName(method.getName());
119.                         //将MethodConfig对象添加到methods属性中
120.                         this.addMethod(methodConfig);
121.                     }
122.                     //Add argument config
123.                     //dubbo.service.{interfaceName}.{methodName}.{arg-index}.xxx=xxx
124.                     java.lang.reflect.Parameter[] arguments = method.getParameters();
125.                     for (int i = 0; i < arguments.length; i++) {
126.                         if (getArgumentByIndex(methodConfig, i) == null && hasArgumentConfigProps(configProperties, methodConfig.getName(), i)) {
127.                             //方法里的每个args参数，都创建一个ArgumentConfig对象
128.                             ArgumentConfig argumentConfig = new ArgumentConfig();
129.                             argumentConfig.setIndex(i);
130.                             //将ArgumentConfig对象添加到MethodConfig对象中
131.                             methodConfig.addArgument(argumentConfig);
132.                         }
133.                     }
134.                 }
135.             }
137.             //refresh MethodConfigs，刷新刚才解析出来的MethodConfig对象
138.             List<MethodConfig> methodConfigs = this.getMethods();
139.             if (methodConfigs != null && methodConfigs.size() > 0) {
140.                 //whether ignore invalid method config
141.                 Object ignoreInvalidMethodConfigVal = getEnvironment().getConfiguration()
142.                     .getProperty(ConfigKeys.DUBBO_CONFIG_IGNORE_INVALID_METHOD_CONFIG, "false");
143.                 boolean ignoreInvalidMethodConfig = Boolean.parseBoolean(ignoreInvalidMethodConfigVal.toString());
144.                 Class<?> finalInterfaceClass = interfaceClass;
145.                 List<MethodConfig> validMethodConfigs = methodConfigs.stream().filter(methodConfig -> {
146.                     methodConfig.setParentPrefix(preferredPrefix);
147.                     //关联Model组件
148.                     methodConfig.setScopeModel(getScopeModel());
149.                     methodConfig.refresh();
150.                     //verify method config
151.                     return verifyMethodConfig(methodConfig, finalInterfaceClass, ignoreInvalidMethodConfig);
152.                 }).collect(Collectors.toList());
153.                 this.setMethods(validMethodConfigs);
154.             }
155.         }
156.     }
158.     public void addMethod(MethodConfig methodConfig) {
159.         if (this.methods == null) {
160.             this.methods = new ArrayList<>();
161.         }
162.         this.methods.add(methodConfig);
163.     }
164.     ...
165. }

复制代码

 
8.执行服务实例发布操作时的主流程
首先通过ScopeModel组件体系获取服务数据存储组件，然后将要发布的服务注册到服务数据存储组件里，接着把相关信息封装成一个服务提供者，并将该服务提供者也注册到服务数据存储组件中，然后生成注册的URL，最后根据协议和生成的注册的URL来发布服务。

1. public class ServiceConfig<T> extends ServiceConfigBase<T> {
2.     ...
3.     @Override
4.     public void export() {
5.         ...
6.         synchronized (this) {
7.             ...
8.             if (this.shouldExport()) {
9.                 //3.执行服务实例的初始化
10.                 //也就是会把Metadata元数据给准备好，后续可以进行元数据上报
11.                 this.init();
12.                 ...
13.             }
14.         }
15.     }
17.     public void init() {
18.         //通过SPI机制获取ServiceListener扩展点的所有实现类实例
19.         //然后添加到ServiceConfig的serviceListeners字段里
20.         if (this.initialized.compareAndSet(false, true)) {
21.             //load ServiceListeners from extension
22.             ExtensionLoader<ServiceListener> extensionLoader = this.getExtensionLoader(ServiceListener.class);
23.             this.serviceListeners.addAll(extensionLoader.getSupportedExtensionInstances());
24.         }
26.         //初始化ServiceMetadata，也就是服务元数据
27.         //这需要与前面设置的MetadataCenter元数据中心配合起来看
28.         //ServiceMetadata作为服务实例的元数据，会对服务实例做一些描述，比如版本号、实现类等
29.         initServiceMetadata(provider);
30.         serviceMetadata.setServiceType(getInterfaceClass());
31.         serviceMetadata.setTarget(getRef());
32.         serviceMetadata.generateServiceKey();
33.     }
34.     ...
35. }

复制代码

 
9.服务发布过程中ProxyFactory生成Invoker

1. public class ServiceConfig<T> extends ServiceConfigBase<T> {
2.     ...
3.     @Override
4.     public void export() {
5.         ...
6.         //4.执行服务实例的发布
7.         //进行debug时，可以通过控制台打印的日志去分析运行流程
8.         //比如通过log日志可以发现服务发布的流程可能涉及：
9.         //一.export Dubbo Service，发布dubbo服务实例
10.         //二.register Dubbo Service，往zk进行注册
11.         //三.启动NettyServer，监听端口和请求处理
12.         //四.服务发现注册
13.         //五.MetadataReport：服务实例上报
14.         //六.关闭JVM时的逆向处理过程
15.         //这里可以作为一个服务发布的直接入口
16.         doExport();
17.     }
19.     protected synchronized void doExport() {
20.         ...
21.         //发布服务
22.         doExportUrls();
23.         //服务发布完成后的处理，比如打印日志和回调监听器
24.         exported();
25.     }
26.     ...
27. }

复制代码

 
10.服务发布过程中Protocol组件发布Invoker
(1)Protocol协议接口
(2)Protocol组件发布Invoker
 
(1)Protocol协议接口

1. public class ServiceConfig<T> extends ServiceConfigBase<T> {
2.     ...  
3.     @SuppressWarnings({"unchecked", "rawtypes"})
4.     private void doExportUrls() {
5.         //所谓的ScopeModel，真实的类型是ModuleModel
6.         //下面这行代码会通过AbstractMethodConfig的getScopeModel()方法获取ScopeModel
7.         //接着会通过ModuleModel的getServiceRepository()方法去获取ServiceRepository
8.         //事实上，Dubbo会把它的各个组件都集中在ScopeModel(ModuleModel)里，而ScopeModel就类似于设计模式里的门面模式
9.         //ScopeModel、ModuleModel、ApplicationModel、FrameworkModel等多个Model会组成一个Model组件体系
11.         //1.通过ScopeModel组件体系获取服务数据存储组件ModuleServiceRepository
12.         ModuleServiceRepository repository = getScopeModel().getServiceRepository();
14.         //ServiceRepository是Dubbo服务的数据存储组件
15.         //一个系统可以发布多个Dubbo服务
16.         //每个Dubbo服务的核心就是一个接口和一个实现类
18.         //2.把当前要发布的服务注册到Dubbo的服务数据存储组件中
19.         ServiceDescriptor serviceDescriptor;
20.         final boolean serverService = ref instanceof ServerService;
21.         if (serverService) {
22.             serviceDescriptor = ((ServerService) ref).getServiceDescriptor();
23.             repository.registerService(serviceDescriptor);
24.         } else {
25.             serviceDescriptor = repository.registerService(getInterfaceClass());
26.         }
28.         //ProviderModel也就是服务提供者，由于这里是暴露服务出去的，所以属于Provider
29.         //3.把所有相关的信息封装成一个服务提供者ProviderModel
30.         providerModel = new ProviderModel(
31.             serviceMetadata.getServiceKey(),
32.             ref,//ref代表的是实际实现的类，通过泛型传入
33.             serviceDescriptor,//表示服务实例相关的信息
34.             getScopeModel(),
35.             serviceMetadata,
36.             interfaceClassLoader
37.         );
38.         providerModel.setConfig(this);
39.         providerModel.setDestroyRunner(getDestroyRunner());
40.         //3.将服务提供者ProviderModel注册到服务数据存储组件中
41.         repository.registerProvider(providerModel);
43.         //4.生成注册的URL：包含2181的端口号、注册到zk中        
44.         //service-discovery-registry://127.0.0.1:2181/org.apache.dubbo.registry.RegistryService?application=dubbo-demo-api-provider&dubbo=2.0.2&pid=989®istry=zookeeper×tamp=1724302222103        //registry://127.0.0.1:2181/org.apache.dubbo.registry.RegistryService?application=dubbo-demo-api-provider&dubbo=2.0.2&pid=989®istry=zookeeper×tamp=1724302222103
45.         List<URL> registryURLs = ConfigValidationUtils.loadRegistries(this, true);
47.         for (ProtocolConfig protocolConfig : protocols) {
48.             String pathKey = URL.buildKey(
49.                 getContextPath(protocolConfig).map(p -> p + "/" + path).orElse(path),
50.                 group,
51.                 version
52.             );
53.             //stub service will use generated service name
54.             if (!serverService) {
55.                 //In case user specified path, register service one more time to map it to path.
56.                 //将接口注册到服务数据存储组件中
57.                 repository.registerService(pathKey, interfaceClass);
58.             }
59.             //5.调用doExportUrlsFor1Protocol()方法，根据协议和注册的URL来发布服务
60.             doExportUrlsFor1Protocol(protocolConfig, registryURLs);
61.         }
62.         providerModel.setServiceUrls(urls);
63.     }
65.     private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List<URL> registryURLs) {
66.         Map<String, String> map = buildAttributes(protocolConfig);
67.         //remove null key and null value
68.         map.keySet().removeIf(key -> StringUtils.isEmpty(key) || StringUtils.isEmpty(map.get(key)));
69.         //init serviceMetadata attachments
70.         //将map数据放入serviceMetadata中，这与元数据相关
71.         serviceMetadata.getAttachments().putAll(map);
72.         //根据ProtocolConfig构建URL
73.         URL url = buildUrl(protocolConfig, map);
74.         //发布服务
75.         exportUrl(url, registryURLs);
76.     }
77.     ...
78. }
80. //ScopeModel、ModuleModel、ApplicationModel、FrameworkModel等多个Model会组成一个Model组件体系
81. public class ModuleModel extends ScopeModel {
82.     public static final String NAME = "ModuleModel";
83.     //ApplicationModel内部封装了其他很多组件，在这里是一个引用关系，通过构造方法传入进来
84.     private final ApplicationModel applicationModel;
85.     //包含了ServiceModule环境相关的数据，里面封装的都是各种各样的配置信息
86.     private ModuleEnvironment moduleEnvironment;
87.     //serviceRepository是一个服务仓储组件，存储了一些服务相关的数据
88.     private ModuleServiceRepository serviceRepository;
89.     //这是module配置管理器，用于存放一些服务相关的配置数据
90.     private ModuleConfigManager moduleConfigManager;
91.     //这是ModuleDeployer组件，用于管理其他的一些组件和模块的生命周期
92.     private ModuleDeployer deployer;
93.     ...
94. }
96. public class ApplicationModel extends ScopeModel {
97.     ...
98.     //包含了多个ModuleModel
99.     private final List<ModuleModel> moduleModels = new CopyOnWriteArrayList<>();
100.     private final List<ModuleModel> pubModuleModels = new CopyOnWriteArrayList<>();
101.     //环境变量、配置信息
102.     private Environment environment;
103.     //服务配置相关的一些信息
104.     private ConfigManager configManager;
105.     //服务数据相关的一些存储
106.     private ServiceRepository serviceRepository;
107.     //属于application层级的一些组件的生命周期管理
108.     private ApplicationDeployer deployer;
109.     //父级组件
110.     private final FrameworkModel frameworkModel;
111.     //内部的一个ModuleModel组件
112.     private ModuleModel internalModule;
113.     //默认的一个ModuleModel组件
114.     private volatile ModuleModel defaultModule;
115.     //internal module index is 0, default module index is 1
116.     private AtomicInteger moduleIndex = new AtomicInteger(0);
117.     //是一个锁
118.     private Object moduleLock = new Object();
119.     ...
120. }
122. public class FrameworkModel extends ScopeModel {
123.     ...
124.     //它没有父层级了，所以只能通过static静态变量，类级别去引用自己的FrameworkModel集合
125.     private static List<FrameworkModel> allInstances = new CopyOnWriteArrayList<>();
126.     //包含了多个ApplicationModel
127.     private List applicationModels = new CopyOnWriteArrayList<>();
128.     //通过Framework、Application、Module各个层级都可以获取到service相关的配置和数据
129.     private FrameworkServiceRepository serviceRepository;
130.     ...
131. }
133. public class ModuleServiceRepository {
134.     private final ModuleModel moduleModel;
135.     //services，代表服务相关的数据，StubServiceDescriptor
136.     private final ConcurrentMap<String, List<ServiceDescriptor>> services = new ConcurrentHashMap<>();
137.     //consumers(key - group/interface:version, value - consumerModel list)
138.     //代表服务的调用方(consumer即消费方/调用方)
139.     private final ConcurrentMap<String, List<ConsumerModel>> consumers = new ConcurrentHashMap<>();
140.     //providers，代表服务提供方
141.     private final ConcurrentMap<String, ProviderModel> providers = new ConcurrentHashMap<>();
142.     //FrameworkServiceRepository存储的也是一些服务相关的数据
143.     private final FrameworkServiceRepository frameworkServiceRepository;
144.     ...
145. }
147. public class ModuleServiceRepository {
148.     //services，代表服务相关的数据，StubServiceDescriptor
149.     private final ConcurrentMap<String, List<ServiceDescriptor>> services = new ConcurrentHashMap<>();
150.     ...
152.     public ServiceDescriptor registerService(ServiceDescriptor serviceDescriptor) {
153.         return registerService(serviceDescriptor.getServiceInterfaceClass(), serviceDescriptor);
154.     }
156.     public ServiceDescriptor registerService(Class<?> interfaceClazz) {
157.         ServiceDescriptor serviceDescriptor = new ReflectionServiceDescriptor(interfaceClazz);
158.         return registerService(interfaceClazz, serviceDescriptor);
159.     }
161.     public ServiceDescriptor registerService(Class<?> interfaceClazz, ServiceDescriptor serviceDescriptor) {
162.         List<ServiceDescriptor> serviceDescriptors = services.computeIfAbsent(interfaceClazz.getName(), k -> new CopyOnWriteArrayList<>());
163.         synchronized (serviceDescriptors) {
164.             Optional<ServiceDescriptor> previous = serviceDescriptors.stream()
165.                 .filter(s -> s.getServiceInterfaceClass().equals(interfaceClazz)).findFirst();
166.             if (previous.isPresent()) {
167.                 return previous.get();
168.             } else {
169.                 serviceDescriptors.add(serviceDescriptor);
170.                 return serviceDescriptor;
171.             }
172.         }
173.     }
174.     ...
175. }

复制代码

(2)Protocol组件发布Invoker
本地发布时使用InjvmProtocol + InjvmExporter，远程发布时使用RegistryProtocol + DestroyableExporter。
 
RegistryProtocol的export()方法被远程发布调用的时候，会调用到DubboProtocol的export()方法，并最终调用到HeaderExchanger的bind()方法执行NettyTransporter的bind()方法构建Netty服务器。

1. public class ServiceConfig<T> extends ServiceConfigBase<T> {
2.     ...
3.     private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List<URL> registryURLs) {
4.         Map<String, String> map = buildAttributes(protocolConfig);
5.         //remove null key and null value
6.         map.keySet().removeIf(key -> StringUtils.isEmpty(key) || StringUtils.isEmpty(map.get(key)));
7.         //init serviceMetadata attachments
8.         //将map数据放入serviceMetadata中，这与元数据相关
9.         serviceMetadata.getAttachments().putAll(map);
10.         //根据ProtocolConfig构建URL
11.         URL url = buildUrl(protocolConfig, map);
12.         //发布服务
13.         exportUrl(url, registryURLs);
14.     }
16.     private URL buildUrl(ProtocolConfig protocolConfig, Map<String, String> params) {
17.         //获取协议名称
18.         String name = protocolConfig.getName();
19.         if (StringUtils.isEmpty(name)) {
20.             //默认使用Dubbo协议
21.             name = DUBBO;
22.         }
23.         //获取host值
24.         String host = findConfiguredHosts(protocolConfig, provider, params);
25.         //获取port值
26.         Integer port = findConfiguredPort(protocolConfig, provider, this.getExtensionLoader(Protocol.class), name, params);
27.         //根据上面获取的host、port以及前文获取的map集合组装URL
28.         URL url = new ServiceConfigURL(name, null, null, host, port, getContextPath(protocolConfig).map(p -> p + "/" + path).orElse(path), params);
29.         //通过Configurator覆盖或添加新的参数
30.         if (this.getExtensionLoader(ConfiguratorFactory.class).hasExtension(url.getProtocol())) {
31.             url = this.getExtensionLoader(ConfiguratorFactory.class).getExtension(url.getProtocol()).getConfigurator(url).configure(url);
32.         }
33.         url = url.setScopeModel(getScopeModel());
34.         url = url.setServiceModel(providerModel);
35.         return url;
36.     }
38.     private void exportUrl(URL url, List<URL> registryURLs) {
39.         //从URL中获取scope参数，其中可选值有none、remote、local三个，分别代表不发布、发布到本地以及发布到远端
40.         String scope = url.getParameter(SCOPE_KEY);
41.         //scope不为none，才进行发布
42.         if (!SCOPE_NONE.equalsIgnoreCase(scope)) {
43.             //scope为local，只发布到本地
44.             if (!SCOPE_REMOTE.equalsIgnoreCase(scope)) {
45.                 exportLocal(url);
46.             }
47.             //export to remote if the config is not local (export to local only when config is local)
48.             //scope为remote，发布到远端的注册中心
49.             if (!SCOPE_LOCAL.equalsIgnoreCase(scope)) {
50.                 //进行远程发布
51.                 url = exportRemote(url, registryURLs);
52.                 if (!isGeneric(generic) && !getScopeModel().isInternal()) {
53.                     //通过MetadataUtils推送这个服务实例的元数据到元数据中心
54.                     //元数据中心是一个动态的配置中心，可以从里面获取参数，也可以添加监听器监听配置项的变更
55.                     MetadataUtils.publishServiceDefinition(url, providerModel.getServiceModel(), getApplicationModel());
56.                 }
57.             }
58.         }
59.         this.urls.add(url);
60.     }
62.     private void exportLocal(URL url) {
63.         //创建新URL
64.         URL local = URLBuilder.from(url)
65.             .setProtocol(LOCAL_PROTOCOL)
66.             .setHost(LOCALHOST_VALUE)
67.             .setPort(0)
68.             .build();
69.         local = local.setScopeModel(getScopeModel()).setServiceModel(providerModel);
70.         //本地发布
71.         doExportUrl(local, false);
72.         //exportLocal，指的是发布到本地
73.         //具体就是在jvm内部完成了组件之间的一些交互关系和发布
74.         logger.info("Export dubbo service " + interfaceClass.getName() + " to local registry url : " + local);
75.     }
77.     private URL exportRemote(URL url, List<URL> registryURLs) {
78.         //如果当前配置了至少一个注册中心
79.         if (CollectionUtils.isNotEmpty(registryURLs)) {
80.             //URL里有很多的信息，比如协议、各种参数等
81.             //URL可以在后续代码运行过程中提供配置和信息
82.             //接下来会向每个注册中心发布服务
83.             for (URL registryURL : registryURLs) {
84.                 //registryURL.getProtocol()会获取协议
85.                 if (SERVICE_REGISTRY_PROTOCOL.equals(registryURL.getProtocol())) {
86.                     url = url.addParameterIfAbsent(SERVICE_NAME_MAPPING_KEY, "true");
87.                 }
88.                 //injvm协议只在exportLocal()中有用，不会将服务发布到注册中心，所以这里忽略injvm协议
89.                 if (LOCAL_PROTOCOL.equalsIgnoreCase(url.getProtocol())) {
90.                     continue;
91.                 }
92.                 //设置服务URL的dynamic参数
93.                 url = url.addParameterIfAbsent(DYNAMIC_KEY, registryURL.getParameter(DYNAMIC_KEY));
94.                 //创建monitorUrl，并作为monitor参数添加到服务URL中
95.                 URL monitorUrl = ConfigValidationUtils.loadMonitor(this, registryURL);
96.                 if (monitorUrl != null) {
97.                     url = url.putAttribute(MONITOR_KEY, monitorUrl);
98.                 }
99.                 //For providers, this is used to enable custom proxy to generate invoker
100.                 //设置服务URL的proxy参数，即生成动态代理方式(jdk或是javassist)，作为参数添加到RegistryURL中
101.                 String proxy = url.getParameter(PROXY_KEY);
102.                 if (StringUtils.isNotEmpty(proxy)) {
103.                     registryURL = registryURL.addParameter(PROXY_KEY, proxy);
104.                 }
105.                 doExportUrl(registryURL.putAttribute(EXPORT_KEY, url), true);
106.             }
107.         } else {
108.             //不存在注册中心，仅发布服务，不会将服务信息发布到注册中心
109.             doExportUrl(url, true);
110.         }
111.         return url;
112.     }
114.     private void doExportUrl(URL url, boolean withMetaData) {
115.         //动态代理技术有很多种，比如cglib，jdk
116.         //而动态代理就是：面向一个接口，动态生成该接口的一个实现类，然后根据这个实现类再动态生成对应的对象
117.         //这个对象就是动态代理的对象，所以该对象会代理自己背后的一个实现类
118.         //当这个对象被调用时，背后的实现类也会被调用
119.         //ProxyFactory，Proxy就是动态代理
120.         //下面传入的ref指的是实现类
121.         //下面传入的interfaceClass指的是接口
122.         //下面传入的url就是服务实例对外暴露出去的一些核心信息
123.         //proxyFactory.getInvoker()获取到的是Invoker调用组件
124.         //当Dubbo的NettyServer监听到网络连接进行请求处理时，需要有一个调用组件去根据请求进行调用
125.         //Invoker调用组件可以认为是ProxyFactory基于DemoService接口生成的动态代理
126.         //当需要根据请求调用接口时，底层就会回调自己写的实现类DemoServiceImpl
127.         //proxyFactory.getInvoker()会封装一个AbstractProxyInvoker，对本地实现类进行代理
128.         //默认情况下，会通过Javassist技术生成Wrapper，该Wrapper会将本地实现类包装进去
129.         //调用AbstractProxyInvoker的invoke方法时，最终就会基于Javassist动态生成的Wrapper进行调用
130.         //下面这一行代码是为服务实现类的对象创建相应的Invoker
131.         //其中传入的服务url会作为export参数添加到RegistryURL中
132.         //这里的proxyFactory就是ProxyFactory接口的适配器，会通过SPI机制进行初始化
133.         //比如下面就会调用JavassistProxyFactory.getInvoker()方法
134.         //proxyFactory.getInvoker()会获取到Invoker调用组件(生成Invoker动态代理)
135.         //所以下面这一行代码会为本地实现类的对象创建相应的Invoker(封装着AbstractProxyInvoker)
136.         Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, url);
137.         if (withMetaData) {
138.             //DelegateProviderMetaDataInvoker是个装饰类
139.             //它会将当前ServiceConfig和Invoker关联起来
140.             invoker = new DelegateProviderMetaDataInvoker(invoker, this);
141.         }
143.         //调用Protocol的实现进行发布，protocolSPI是Protocol接口的适配器
144.         //进行本地发布时，使用的是InjvmProtocol + InjvmExporter
145.         //进行远程发布时，使用的是RegistryProtocol，它会对DubboProtocol进行包装和装饰
146.         //RegistryProtocol会先执行来处理服务注册的一些事情
147.         //DubboProtocol会后执行来启动NettyServer网络服务器
148.         Exporter<?> exporter = protocolSPI.export(invoker);
149.         exporters.add(exporter);
150.     }
151.     ...
152. }
154. public class JavassistProxyFactory extends AbstractProxyFactory {
155.     ...
156.     @Override
157.     public <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) throws RpcException {
158.         //下面会通过Wrapper创建一个包装类对象
159.         //该对象是动态构建出来的，它属于Wrapper的一个子类，里面会拼接一个关键的方法invokeMethod()，拼接代码由javassist动态生成
160.         final Wrapper wrapper = Wrapper.getWrapper(proxy.getClass().getName().indexOf('$') < 0 ? proxy.getClass() : type);
162.         //下面会创建一个实现了AbstractProxyInvoker的匿名内部类
163.         //其doInvoker()方法会直接委托给Wrapper对象的invokeMethod()方法
164.         return new AbstractProxyInvoker<T>(proxy, type, url) {
165.             @Override
166.             protected Object doInvoke(T proxy, String methodName, Class<?>[] parameterTypes, Object[] arguments) throws Throwable {
167.                 //当AbstractProxyInvoker.invoke()方法被调用时，便会执行到这里
168.                 //这里会通过类似于JDK反射的技术，调用本地实现类如DemoServiceImpl.sayHello()
169.                 //这个wrapper对象是由javassist技术动态生成的，已经对本地实现类进行包装
170.                 //这个动态生成的wrapper对象会通过javassist技术自己特有的方法
171.                 //在invokerMethod()方法被调用时执行本地实现类的目标方法
172.                 return wrapper.invokeMethod(proxy, methodName, parameterTypes, arguments);
173.             }
174.         };
175.     }
176.     ...
177. }
179. public abstract class AbstractProxyInvoker<T> implements Invoker<T> {
180.     ...
181.     //当Netty Server接受到了请求后，经过解析就会知道是要调用什么
182.     //然后会把解析出来的数据放入Invocation中，通过AbstractProxyInvoker的invoke()方法来进行调用
183.     @Override
184.     public Result invoke(Invocation invocation) throws RpcException {
185.         ...
186.         //执行doInvoke()方法，调用业务实现
187.         Object value = doInvoke(proxy, invocation.getMethodName(), invocation.getParameterTypes(), invocation.getArguments());
188.         ...
189.         //将value值封装成CompletableFuture对象
190.         CompletableFuture<Object> future = wrapWithFuture(value, invocation);
191.         //再次转换，转换为CompletableFuture类型
192.         CompletableFuture appResponseFuture = future.handle((obj, t) -> {
193.         AppResponse result = new AppResponse(invocation);
194.         ...
195.         //将CompletableFuture封装成AsyncRpcResult返回
196.         return new AsyncRpcResult(appResponseFuture, invocation);
197.     }
199.     protected abstract Object doInvoke(T proxy, String methodName, Class<?>[] parameterTypes, Object[] arguments) throws Throwable;
200.     ...
201. }

复制代码

[code]//-> Protocol$Adaptive.export()//-> ProtocolSerializationWrapper.export()//-> ProtocolFilterWrapper.export()//-> ProtocolListenerWrapper.export()//-> InjvmProtocol.export()public class InjvmProtocol extends AbstractProtocol implements Protocol {    ...    @Override    public  Exporter export(Invoker invoker) throws RpcException {        return new InjvmExporter(invoker, invoker.getUrl().getServiceKey(), exporterMap);    }    ...}public class InjvmExporter extends AbstractExporter {    private final String key;    //这就是在JVM里存放了    private final Map> exporterMap;    public DubboExporter(Invoker invoker, String key, Map