Dubbo源码分析（二）：Dubbo之消费端（Consumer）

    通观全部Dubbo代码，有两个很重要的对象就是Invoker和Exporter，Dubbo会根据用户配置的协议调用不同协议的Invoker，再通过ReferenceFonfig将Invoker的引用关联到Reference的ref属性上提供给消费端调用。当用户调用一个Service接口的一个方法后由于Dubbo使用javassist动态代理，会调用Invoker的Invoke方法从而初始化一个RPC调用访问请求访问服务端的Service返回结果。下面我们就从Comsumer端开始逐步解析这个框架。

    Dubbo首先使用com.alibaba.dubbo.config.spring.schema.NamespaceHandler注册解析器，当spring解析xml配置文件时就会调用这些解析器生成对应的BeanDefinition交给spring管理：

[java] view plain copy

 

 

1. public class DubboNamespaceHandler extends NamespaceHandlerSupport {  
2.   
3.     static {  
4.         Version.checkDuplicate(DubboNamespaceHandler.class);  
5.     }  
6.   
7.     public void init() {  
8.         //配置<dubbo:application>标签解析器  
9.         registerBeanDefinitionParser("application", new DubboBeanDefinitionParser(ApplicationConfig.class, true));  
10.         //配置<dubbo:module>标签解析器  
11.         registerBeanDefinitionParser("module", new DubboBeanDefinitionParser(ModuleConfig.class, true));  
12.         //配置<dubbo:registry>标签解析器  
13.         registerBeanDefinitionParser("registry", new DubboBeanDefinitionParser(RegistryConfig.class, true));  
14.         //配置<dubbo:monitor>标签解析器  
15.         registerBeanDefinitionParser("monitor", new DubboBeanDefinitionParser(MonitorConfig.class, true));  
16.         //配置<dubbo:provider>标签解析器  
17.         registerBeanDefinitionParser("provider", new DubboBeanDefinitionParser(ProviderConfig.class, true));  
18.         //配置<dubbo:consumer>标签解析器  
19.         registerBeanDefinitionParser("consumer", new DubboBeanDefinitionParser(ConsumerConfig.class, true));  
20.         //配置<dubbo:protocol>标签解析器  
21.         registerBeanDefinitionParser("protocol", new DubboBeanDefinitionParser(ProtocolConfig.class, true));  
22.         //配置<dubbo:service>标签解析器  
23.         registerBeanDefinitionParser("service", new DubboBeanDefinitionParser(ServiceBean.class, true));  
24.         //配置<dubbo:refenrence>标签解析器  
25.         registerBeanDefinitionParser("reference", new DubboBeanDefinitionParser(ReferenceBean.class, false));  
26.         //配置<dubbo:annotation>标签解析器  
27.         registerBeanDefinitionParser("annotation", new DubboBeanDefinitionParser(AnnotationBean.class, true));  
28.     }  
29. }  

    Spring在初始化IOC容器时会利用这里注册的BeanDefinitionParser的parse方法获取对应的ReferenceBean的BeanDefinition实例，由于ReferenceBean实现了InitializingBean接口，在设置了bean的所有属性后会调用afterPropertiesSet方法：

[java] view plain copy

 

 

1.     public void afterPropertiesSet() throws Exception {  
2.         //如果Consumer还未注册  
3.         if (getConsumer() == null) {  
4.             //获取applicationContext这个IOC容器实例中的所有ConsumerConfig  
5.             Map<String, ConsumerConfig> consumerConfigMap = applicationContext == null ? null  : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ConsumerConfig.class, false, false);  
6.             //如果IOC容器中存在这样的ConsumerConfig  
7.             if (consumerConfigMap != null && consumerConfigMap.size() > 0) {  
8.                 ConsumerConfig consumerConfig = null;  
9.                 //遍历这些ConsumerConfig  
10.                 for (ConsumerConfig config : consumerConfigMap.values()) {  
11.                     //如果用户没配置Consumer系统会生成一个默认Consumer，且它的isDefault返回ture  
12.                     //这里是说要么是Consumer是默认的要么是用户配置的Consumer并且没设置isDefault属性  
13.                     if (config.isDefault() == null || config.isDefault().booleanValue()) {  
14.                         //防止存在两个默认Consumer  
15.                         if (consumerConfig != null) {  
16.                             throw new IllegalStateException("Duplicate consumer configs: " + consumerConfig + " and " + config);  
17.                         }  
18.                         //获取默认Consumer  
19.                         consumerConfig = config;  
20.                     }  
21.                 }  
22.                 if (consumerConfig != null) {  
23.                     //设置默认Consumer  
24.                     setConsumer(consumerConfig);  
25.                 }  
26.             }  
27.         }  
28.         //如果reference未绑定application且（reference未绑定consumer或referenc绑定的consumer没绑定application  
29.         if (getApplication() == null  
30.                 && (getConsumer() == null || getConsumer().getApplication() == null)) {  
31.             //获取IOC中所有application的实例  
32.             Map<String, ApplicationConfig> applicationConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ApplicationConfig.class, false, false);  
33.             if (applicationConfigMap != null && applicationConfigMap.size() > 0) {  
34.                 //如果IOC中存在application  
35.                 ApplicationConfig applicationConfig = null;  
36.                 //遍历这些application  
37.                 for (ApplicationConfig config : applicationConfigMap.values()) {  
38.                     //如果application是默认创建或者被指定成默认  
39.                     if (config.isDefault() == null || config.isDefault().booleanValue()) {  
40.                         if (applicationConfig != null) {  
41.                             throw new IllegalStateException("Duplicate application configs: " + applicationConfig + " and " + config);  
42.                         }  
43.                         //获取application  
44.                         applicationConfig = config;  
45.                     }  
46.                 }  
47.                 if (applicationConfig != null) {  
48.                     //关联到reference  
49.                     setApplication(applicationConfig);  
50.                 }  
51.             }  
52.         }  
53.         //如果reference未绑定module且（reference未绑定consumer或referenc绑定的consumer没绑定module  
54.         if (getModule() == null  
55.                 && (getConsumer() == null || getConsumer().getModule() == null)) {  
56.             //获取IOC中所有module的实例  
57.             Map<String, ModuleConfig> moduleConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ModuleConfig.class, false, false);  
58.             if (moduleConfigMap != null && moduleConfigMap.size() > 0) {  
59.                 ModuleConfig moduleConfig = null;  
60.               //遍历这些module  
61.                 for (ModuleConfig config : moduleConfigMap.values()) {  
62.                     //如果module是默认创建或者被指定成默认  
63.                     if (config.isDefault() == null || config.isDefault().booleanValue()) {  
64.                         if (moduleConfig != null) {  
65.                             throw new IllegalStateException("Duplicate module configs: " + moduleConfig + " and " + config);  
66.                         }  
67.                       //获取module  
68.                         moduleConfig = config;  
69.                     }  
70.                 }  
71.                 if (moduleConfig != null) {  
72.                     //关联到reference  
73.                     setModule(moduleConfig);  
74.                 }  
75.             }  
76.         }  
77.         //如果reference未绑定注册中心（Register）且（reference未绑定consumer或referenc绑定的consumer没绑定注册中心（Register）  
78.         if ((getRegistries() == null || getRegistries().size() == 0)  
79.                 && (getConsumer() == null || getConsumer().getRegistries() == null || getConsumer().getRegistries().size() == 0)  
80.                 && (getApplication() == null || getApplication().getRegistries() == null || getApplication().getRegistries().size() == 0)) {  
81.             //获取IOC中所有的注册中心（Register）实例  
82.             Map<String, RegistryConfig> registryConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, RegistryConfig.class, false, false);  
83.             if (registryConfigMap != null && registryConfigMap.size() > 0) {  
84.                 List<RegistryConfig> registryConfigs = new ArrayList<RegistryConfig>();  
85.                 //遍历这些registry  
86.                 for (RegistryConfig config : registryConfigMap.values()) {  
87.                     //如果registry是默认创建或者被指定成默认  
88.                     if (config.isDefault() == null || config.isDefault().booleanValue()) {  
89.                         registryConfigs.add(config);  
90.                     }  
91.                 }  
92.                   
93.                 if (registryConfigs != null && registryConfigs.size() > 0) {  
94.                     //关联到reference，此处可以看出一个consumer可以绑定多个registry（注册中心）  
95.                     super.setRegistries(registryConfigs);  
96.                 }  
97.             }  
98.         }  
99.       //如果reference未绑定监控中心（Monitor）且（reference未绑定consumer或reference绑定的consumer没绑定监控中心（Monitor）  
100.         if (getMonitor() == null  
101.                 && (getConsumer() == null || getConsumer().getMonitor() == null)  
102.                 && (getApplication() == null || getApplication().getMonitor() == null)) {  
103.             //获取IOC中所有的监控中心（Monitor）实例  
104.             Map<String, MonitorConfig> monitorConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, MonitorConfig.class, false, false);  
105.             if (monitorConfigMap != null && monitorConfigMap.size() > 0) {  
106.                 MonitorConfig monitorConfig = null;  
107.                 //遍历这些监控中心（Monitor）  
108.                 for (MonitorConfig config : monitorConfigMap.values()) {  
109.                     //如果monitor是默认创建或者被指定成默认  
110.                     if (config.isDefault() == null || config.isDefault().booleanValue()) {  
111.                         if (monitorConfig != null) {  
112.                             throw new IllegalStateException("Duplicate monitor configs: " + monitorConfig + " and " + config);  
113.                         }  
114.                         monitorConfig = config;  
115.                     }  
116.                 }  
117.                 if (monitorConfig != null) {  
118.                     //关联到reference,一个consumer绑定到一个监控中心（monitor）  
119.                     setMonitor(monitorConfig);  
120.                 }  
121.             }  
122.         }  
123.         Boolean b = isInit();  
124.         if (b == null && getConsumer() != null) {  
125.             b = getConsumer().isInit();  
126.         }  
127.         if (b != null && b.booleanValue()) {  
128.             //如果consumer已经被关联则组装Reference  
129.             getObject();  
130.         }  
131.     }  
132. }  

      这步其实是Reference确认生成Invoker所需要的组件是否已经准备好，都准备好后我们进入生成Invoker的部分。这里的getObject会调用父类ReferenceConfig的init方法完成组装：

[java] view plain copy

 

 

1. private void init() {  
2.         //避免重复初始化  
3.         if (initialized) {  
4.             return;  
5.         }  
6.         //置为已经初始化  
7.         initialized = true;  
8.         //如果interfaceName不存在  
9.         if (interfaceName == null || interfaceName.length() == 0) {  
10.             throw new IllegalStateException("<dubbo:reference interface=\"\" /> interface not allow null!");  
11.         }  
12.         // 获取消费者  
13.         checkDefault();  
14.         appendProperties(this);  
15.         //如果未使用泛接口并且consumer已经准备好的情况下，reference使用和consumer一样的泛接口  
16.         if (getGeneric() == null && getConsumer() != null) {  
17.             setGeneric(getConsumer().getGeneric());  
18.         }  
19.         //如果是泛接口那么interface的类型是GenericService  
20.         if (ProtocolUtils.isGeneric(getGeneric())) {  
21.             interfaceClass = GenericService.class;  
22.         } else {  
23.             //如果不是泛接口使用interfaceName指定的泛接口  
24.             try {  
25.                 interfaceClass = Class.forName(interfaceName, true, Thread.currentThread()  
26.                         .getContextClassLoader());  
27.             } catch (ClassNotFoundException e) {  
28.                 throw new IllegalStateException(e.getMessage(), e);  
29.             }  
30.             //检查接口以及接口中的方法都是否配置齐全  
31.             checkInterfaceAndMethods(interfaceClass, methods);  
32.         }  
33.         //如果服务比较多可以指定dubbo-resolve.properties文件配置service（service集中配置文件）  
34.         String resolve = System.getProperty(interfaceName);  
35.         String resolveFile = null;  
36.         if (resolve == null || resolve.length() == 0) {  
37.             resolveFile = System.getProperty("dubbo.resolve.file");  
38.             if (resolveFile == null || resolveFile.length() == 0) {  
39.                 File userResolveFile = new File(new File(System.getProperty("user.home")), "dubbo-resolve.properties");  
40.                 if (userResolveFile.exists()) {  
41.                     resolveFile = userResolveFile.getAbsolutePath();  
42.                 }  
43.             }  
44.             if (resolveFile != null && resolveFile.length() > 0) {  
45.                 Properties properties = new Properties();  
46.                 FileInputStream fis = null;  
47.                 try {  
48.                     fis = new FileInputStream(new File(resolveFile));  
49.                     properties.load(fis);  
50.                 } catch (IOException e) {  
51.                     throw new IllegalStateException("Unload " + resolveFile + ", cause: " + e.getMessage(), e);  
52.                 } finally {  
53.                     try {  
54.                         if(null != fis) fis.close();  
55.                     } catch (IOException e) {  
56.                         logger.warn(e.getMessage(), e);  
57.                     }  
58.                 }  
59.                 resolve = properties.getProperty(interfaceName);  
60.             }  
61.         }  
62.         if (resolve != null && resolve.length() > 0) {  
63.             url = resolve;  
64.             if (logger.isWarnEnabled()) {  
65.                 if (resolveFile != null && resolveFile.length() > 0) {  
66.                     logger.warn("Using default dubbo resolve file " + resolveFile + " replace " + interfaceName + "" + resolve + " to p2p invoke remote service.");  
67.                 } else {  
68.                     logger.warn("Using -D" + interfaceName + "=" + resolve + " to p2p invoke remote service.");  
69.                 }  
70.             }  
71.         }  
72.         //如果application、module、registries、monitor未配置则使用consumer的  
73.         if (consumer != null) {  
74.             if (application == null) {  
75.                 application = consumer.getApplication();  
76.             }  
77.             if (module == null) {  
78.                 module = consumer.getModule();  
79.             }  
80.             if (registries == null) {  
81.                 registries = consumer.getRegistries();  
82.             }  
83.             if (monitor == null) {  
84.                 monitor = consumer.getMonitor();  
85.             }  
86.         }  
87.         //如果module已关联则关联module的registries和monitor  
88.         if (module != null) {  
89.             if (registries == null) {  
90.                 registries = module.getRegistries();  
91.             }  
92.             if (monitor == null) {  
93.                 monitor = module.getMonitor();  
94.             }  
95.         }  
96.       //如果application已关联则关联application的registries和monitor  
97.         if (application != null) {  
98.             if (registries == null) {  
99.                 registries = application.getRegistries();  
100.             }  
101.             if (monitor == null) {  
102.                 monitor = application.getMonitor();  
103.             }  
104.         }  
105.         //检查application  
106.         checkApplication();  
107.         //检查远端和本地服务接口真实存在（是否可load）  
108.         checkStubAndMock(interfaceClass);  
109.         Map<String, String> map = new HashMap<String, String>();  
110.         //配置dubbo的端属性（是consumer还是provider）、版本属性、创建时间、进程号  
111.         Map<Object, Object> attributes = new HashMap<Object, Object>();  
112.         map.put(Constants.SIDE_KEY, Constants.CONSUMER_SIDE);  
113.         map.put(Constants.DUBBO_VERSION_KEY, Version.getVersion());  
114.         map.put(Constants.TIMESTAMP_KEY, String.valueOf(System.currentTimeMillis()));  
115.         if (ConfigUtils.getPid() > 0) {  
116.             map.put(Constants.PID_KEY, String.valueOf(ConfigUtils.getPid()));  
117.         }  
118.         if (! isGeneric()) {  
119.             String revision = Version.getVersion(interfaceClass, version);  
120.             if (revision != null && revision.length() > 0) {  
121.                 map.put("revision", revision);  
122.             }  
123.   
124.             String[] methods = Wrapper.getWrapper(interfaceClass).getMethodNames();  
125.             if(methods.length == 0) {  
126.                 logger.warn("NO method found in service interface " + interfaceClass.getName());  
127.                 map.put("methods", Constants.ANY_VALUE);  
128.             }  
129.             else {  
130.                 map.put("methods", StringUtils.join(new HashSet<String>(Arrays.asList(methods)), ","));  
131.             }  
132.         }  
133.         map.put(Constants.INTERFACE_KEY, interfaceName);  
134.         //调用application、module、consumer的get方法将属性设置到map中  
135.         appendParameters(map, application);  
136.         appendParameters(map, module);  
137.         appendParameters(map, consumer, Constants.DEFAULT_KEY);  
138.         appendParameters(map, this);  
139.         String prifix = StringUtils.getServiceKey(map);  
140.         if (methods != null && methods.size() > 0) {  
141.             for (MethodConfig method : methods) {  
142.                 appendParameters(map, method, method.getName());  
143.                 String retryKey = method.getName() + ".retry";  
144.                 if (map.containsKey(retryKey)) {  
145.                     String retryValue = map.remove(retryKey);  
146.                     if ("false".equals(retryValue)) {  
147.                         map.put(method.getName() + ".retries", "0");  
148.                     }  
149.                 }  
150.                 appendAttributes(attributes, method, prifix + "." + method.getName());  
151.                 checkAndConvertImplicitConfig(method, map, attributes);  
152.             }  
153.         }  
154.         //attributes通过系统context进行存储.  
155.         StaticContext.getSystemContext().putAll(attributes);  
156.         //在map装载了application、module、consumer、reference的所有属性信息后创建代理  
157.         ref = createProxy(map);  
158.     }  

[java] view plain copy

 

 

1. private T createProxy(Map<String, String> map) {  
2.         URL tmpUrl = new URL("temp", "localhost", 0, map);  
3.         final boolean isJvmRefer;  
4.         if (isInjvm() == null) {  
5.             if (url != null && url.length() > 0) { //指定URL的情况下，不做本地引用  
6.                 isJvmRefer = false;  
7.             } else if (InjvmProtocol.getInjvmProtocol().isInjvmRefer(tmpUrl)) {  
8.                 //默认情况下如果本地有服务暴露，则引用本地服务.  
9.                 isJvmRefer = true;  
10.             } else {  
11.                 isJvmRefer = false;  
12.             }  
13.         } else {  
14.             isJvmRefer = isInjvm().booleanValue();  
15.         }  
16.           
17.         if (isJvmRefer) {  
18.             URL url = new URL(Constants.LOCAL_PROTOCOL, NetUtils.LOCALHOST, 0, interfaceClass.getName()).addParameters(map);  
19.             invoker = refprotocol.refer(interfaceClass, url);  
20.             if (logger.isInfoEnabled()) {  
21.                 logger.info("Using injvm service " + interfaceClass.getName());  
22.             }  
23.         } else {  
24.             if (url != null && url.length() > 0) { // 用户指定URL，指定的URL可能是对点对直连地址，也可能是注册中心URL  
25.                 String[] us = Constants.SEMICOLON_SPLIT_PATTERN.split(url);  
26.                 if (us != null && us.length > 0) {  
27.                     for (String u : us) {  
28.                         URL url = URL.valueOf(u);  
29.                         if (url.getPath() == null || url.getPath().length() == 0) {  
30.                             url = url.setPath(interfaceName);  
31.                         }  
32.                         if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {  
33.                             urls.add(url.addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map)));  
34.                         } else {  
35.                             urls.add(ClusterUtils.mergeUrl(url, map));  
36.                         }  
37.                     }  
38.                 }  
39.             } else { // 通过注册中心配置拼装URL  
40.                 List<URL> us = loadRegistries(false);  
41.                 if (us != null && us.size() > 0) {  
42.                     for (URL u : us) {  
43.                         URL monitorUrl = loadMonitor(u);  
44.                         if (monitorUrl != null) {  
45.                             map.put(Constants.MONITOR_KEY, URL.encode(monitorUrl.toFullString()));  
46.                         }  
47.                         urls.add(u.addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map)));  
48.                     }  
49.                 }  
50.                 if (urls == null || urls.size() == 0) {  
51.                     throw new IllegalStateException("No such any registry to reference " + interfaceName  + " on the consumer " + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion() + ", please config <dubbo:registry address=\"...\" /> to your spring config.");  
52.                 }  
53.             }  
54.   
55.             if (urls.size() == 1) {  
56.                 //此处举例说明如果是Dubbo协议则调用DubboProtocol的refer方法生成invoker，当用户调用service接口实际调用的是invoker的invoke方法  
57.                 invoker = refprotocol.refer(interfaceClass, urls.get(0));  
58.             } else {  
59.                 //多个service生成多个invoker  
60.                 List<Invoker<?>> invokers = new ArrayList<Invoker<?>>();  
61.                 URL registryURL = null;  
62.                 for (URL url : urls) {  
63.                     invokers.add(refprotocol.refer(interfaceClass, url));  
64.                     if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {  
65.                         registryURL = url; // 用了最后一个registry url  
66.                     }  
67.                 }  
68.                 if (registryURL != null) { // 有 注册中心协议的URL  
69.                     // 对有注册中心的Cluster 只用 AvailableCluster  
70.                     URL u = registryURL.addParameter(Constants.CLUSTER_KEY, AvailableCluster.NAME);   
71.                     invoker = cluster.join(new StaticDirectory(u, invokers));  
72.                 }  else { // 不是 注册中心的URL  
73.                     invoker = cluster.join(new StaticDirectory(invokers));  
74.                 }  
75.             }  
76.         }  
77.   
78.         Boolean c = check;  
79.         if (c == null && consumer != null) {  
80.             c = consumer.isCheck();  
81.         }  
82.         if (c == null) {  
83.             c = true; // default true  
84.         }  
85.         if (c && ! invoker.isAvailable()) {  
86.             throw new IllegalStateException("Failed to check the status of the service " + interfaceName + ". No provider available for the service " + (group == null ? "" : group + "/") + interfaceName + (version == null ? "" : ":" + version) + " from the url " + invoker.getUrl() + " to the consumer " + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion());  
87.         }  
88.         if (logger.isInfoEnabled()) {  
89.             logger.info("Refer dubbo service " + interfaceClass.getName() + " from url " + invoker.getUrl());  
90.         }  
91.         // 创建服务代理  
92.         return (T) proxyFactory.getProxy(invoker);  
93.     }  

      至此Reference在关联了所有application、module、consumer、registry、monitor、service、protocol后调用对应Protocol类的refer方法生成InvokerProxy。当用户调用service时dubbo会通过InvokerProxy调用Invoker的invoke的方法向服务端发起请求。客户端就这样完成了自己的初始化。