dubbo源碼分析（3）
繼上一章研究provider的加載過程以後，同理consumer的加載過程基本上和provider過程如出一轍。java
一樣也是先讀取consumer.xml文件spring
<?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:dubbo="http://code.alibabatech.com/schema/dubbo" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://code.alibabatech.com/schema/dubbo http://code.alibabatech.com/schema/dubbo/dubbo.xsd"> <dubbo:application name="zookeeper_privider" /> <dubbo:consumer timeout="100000" filter="DataSourceKeyFilter"/> <dubbo:registry protocol="zookeeper" address="127.0.0.1:2181" /> <dubbo:reference id="xxxService" interface="com.xxx.xx.service.xxxService" group="xyx"></dubbo:reference> </beans>
Dubbo首先使用com.alibaba.dubbo.config.spring.schema.NamespaceHandler註冊解析器，當spring解析xml配置文件時就會調用這些解析器生成對應的BeanDefinition交給spring管理：app
/**
* DubboNamespaceHandler
*
* @author william.liangf
* @export
*/
public class DubboNamespaceHandler extends NamespaceHandlerSupport {
static {
Version.checkDuplicate(DubboNamespaceHandler.class);
}
public void init() {
registerBeanDefinitionParser("application", new DubboBeanDefinitionParser(ApplicationConfig.class, true));
registerBeanDefinitionParser("module", new DubboBeanDefinitionParser(ModuleConfig.class, true));
registerBeanDefinitionParser("registry", new DubboBeanDefinitionParser(RegistryConfig.class, true));
registerBeanDefinitionParser("monitor", new DubboBeanDefinitionParser(MonitorConfig.class, true));
registerBeanDefinitionParser("provider", new DubboBeanDefinitionParser(ProviderConfig.class, true));
registerBeanDefinitionParser("consumer", new DubboBeanDefinitionParser(ConsumerConfig.class, true));
registerBeanDefinitionParser("protocol", new DubboBeanDefinitionParser(ProtocolConfig.class, true));
registerBeanDefinitionParser("service", new DubboBeanDefinitionParser(ServiceBean.class, true));
registerBeanDefinitionParser("reference", new DubboBeanDefinitionParser(ReferenceBean.class, false));
registerBeanDefinitionParser("annotation", new DubboBeanDefinitionParser(AnnotationBean.class, true));
}
Spring在初始化IOC容器時會利用這裏註冊的BeanDefinitionParser的parse方法獲取對應的ReferenceBean的BeanDefinition實例，因爲ReferenceBean實現了InitializingBean接口，在設置了bean的全部屬性後會調用afterPropertiesSet方法：jvm
public void afterPropertiesSet() throws Exception {
//若是Consumer還未註冊
if (getConsumer() == null) {
//獲取applicationContext這個IOC容器實例中的全部ConsumerConfig
Map<String, ConsumerConfig> consumerConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ConsumerConfig.class, false, false);
//若是IOC容器中存在這樣的ConsumerConfig
if (consumerConfigMap != null && consumerConfigMap.size() > 0) {
ConsumerConfig consumerConfig = null;
//遍歷這些ConsumerConfig
for (ConsumerConfig config : consumerConfigMap.values()) {
//若是用戶沒配置Consumer系統會生成一個默認Consumer，且它的isDefault返回ture
//這裏是說要麼是Consumer是默認的要麼是用戶配置的Consumer而且沒設置isDefault屬性
if (config.isDefault() == null || config.isDefault().booleanValue()) {
//防止存在兩個默認Consumer
if (consumerConfig != null) {
throw new IllegalStateException("Duplicate consumer configs: " + consumerConfig + " and " + config);
}
//獲取默認Consumer
consumerConfig = config;
}
}
if (consumerConfig != null) {
//設置默認Consumer
setConsumer(consumerConfig);
}
}
}
//若是reference未綁定application且（reference未綁定consumer或referenc綁定的consumer沒綁定application
if (getApplication() == null
&& (getConsumer() == null || getConsumer().getApplication() == null)) {
//獲取IOC中全部application的實例
Map<String, ApplicationConfig> applicationConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ApplicationConfig.class, false, false);
if (applicationConfigMap != null && applicationConfigMap.size() > 0) {
//若是IOC中存在application
ApplicationConfig applicationConfig = null;
//遍歷這些application
for (ApplicationConfig config : applicationConfigMap.values()) {
//若是application是默認建立或者被指定成默認
if (config.isDefault() == null || config.isDefault().booleanValue()) {
if (applicationConfig != null) {
throw new IllegalStateException("Duplicate application configs: " + applicationConfig + " and " + config);
}
//獲取application
applicationConfig = config;
}
}
if (applicationConfig != null) {
//關聯到reference
setApplication(applicationConfig);
}
}
}
//若是reference未綁定module且（reference未綁定consumer或referenc綁定的consumer沒綁定module
if (getModule() == null
&& (getConsumer() == null || getConsumer().getModule() == null)) {
//獲取IOC中全部module的實例
Map<String, ModuleConfig> moduleConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ModuleConfig.class, false, false);
if (moduleConfigMap != null && moduleConfigMap.size() > 0) {
ModuleConfig moduleConfig = null;
//遍歷這些module
for (ModuleConfig config : moduleConfigMap.values()) {
//若是module是默認建立或者被指定成默認
if (config.isDefault() == null || config.isDefault().booleanValue()) {
if (moduleConfig != null) {
throw new IllegalStateException("Duplicate module configs: " + moduleConfig + " and " + config);
}
//獲取module
moduleConfig = config;
}
}
if (moduleConfig != null) {
//關聯到reference
setModule(moduleConfig);
}
}
}
//若是reference未綁定註冊中心（Register）且（reference未綁定consumer或referenc綁定的consumer沒綁定註冊中心（Register）
if ((getRegistries() == null || getRegistries().size() == 0)
&& (getConsumer() == null || getConsumer().getRegistries() == null || getConsumer().getRegistries().size() == 0)
&& (getApplication() == null || getApplication().getRegistries() == null || getApplication().getRegistries().size() == 0)) {
//獲取IOC中全部的註冊中心（Register）實例
Map<String, RegistryConfig> registryConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, RegistryConfig.class, false, false);
if (registryConfigMap != null && registryConfigMap.size() > 0) {
List<RegistryConfig> registryConfigs = new ArrayList<RegistryConfig>();
//遍歷這些registry
for (RegistryConfig config : registryConfigMap.values()) {
//若是registry是默認建立或者被指定成默認
if (config.isDefault() == null || config.isDefault().booleanValue()) {
registryConfigs.add(config);
}
}
if (registryConfigs != null && registryConfigs.size() > 0) {
//關聯到reference，此處能夠看出一個consumer能夠綁定多個registry（註冊中心）
super.setRegistries(registryConfigs);
}
}
}
//若是reference未綁定監控中心（Monitor）且（reference未綁定consumer或reference綁定的consumer沒綁定監控中心（Monitor）
if (getMonitor() == null
&& (getConsumer() == null || getConsumer().getMonitor() == null)
&& (getApplication() == null || getApplication().getMonitor() == null)) {
//獲取IOC中全部的監控中心（Monitor）實例
Map<String, MonitorConfig> monitorConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, MonitorConfig.class, false, false);
if (monitorConfigMap != null && monitorConfigMap.size() > 0) {
MonitorConfig monitorConfig = null;
//遍歷這些監控中心（Monitor）
for (MonitorConfig config : monitorConfigMap.values()) {
//若是monitor是默認建立或者被指定成默認
if (config.isDefault() == null || config.isDefault().booleanValue()) {
if (monitorConfig != null) {
throw new IllegalStateException("Duplicate monitor configs: " + monitorConfig + " and " + config);
}
monitorConfig = config;
}
}
if (monitorConfig != null) {
//關聯到reference,一個consumer綁定到一個監控中心（monitor）
setMonitor(monitorConfig);
}
}
}
Boolean b = isInit();
if (b == null && getConsumer() != null) {
b = getConsumer().isInit();
}
if (b != null && b.booleanValue()) {
//若是consumer已經被關聯則組裝Reference
getObject();
}
}
}
這步實際上是Reference確認生成Invoker所須要的組件是否已經準備好，都準備好後咱們進入生成Invoker的部分。這裏的getObject會調用父類ReferenceConfig的init方法完成組裝：ide
首先ReferenceConfig類的init方法調用Protocol的refer方法生成Invoker實例(如上圖中的紅色部分)，這是服務消費的關鍵。接下來把Invoker轉換爲客戶端須要的接口。詳細代碼以下：this
private void init() {
//避免重複初始化
if (initialized) {
return;
}
//置爲已經初始化
initialized = true;
//若是interfaceName不存在
if (interfaceName == null || interfaceName.length() == 0) {
throw new IllegalStateException("<dubbo:reference interface=\"\" /> interface not allow null!");
}
// 獲取消費者
checkDefault();
appendProperties(this);
//若是未使用泛接口而且consumer已經準備好的狀況下，reference使用和consumer同樣的泛接口
if (getGeneric() == null && getConsumer() != null) {
setGeneric(getConsumer().getGeneric());
}
//若是是泛接口那麼interface的類型是GenericService
if (ProtocolUtils.isGeneric(getGeneric())) {
interfaceClass = GenericService.class;
} else {
//若是不是泛接口使用interfaceName指定的泛接口
try {
interfaceClass = Class.forName(interfaceName, true, Thread.currentThread()
.getContextClassLoader());
} catch (ClassNotFoundException e) {
throw new IllegalStateException(e.getMessage(), e);
}
//檢查接口以及接口中的方法都是否配置齊全
checkInterfaceAndMethods(interfaceClass, methods);
}
//若是服務比較多能夠指定dubbo-resolve.properties文件配置service（service集中配置文件）
String resolve = System.getProperty(interfaceName);
String resolveFile = null;
if (resolve == null || resolve.length() == 0) {
resolveFile = System.getProperty("dubbo.resolve.file");
if (resolveFile == null || resolveFile.length() == 0) {
File userResolveFile = new File(new File(System.getProperty("user.home")), "dubbo-resolve.properties");
if (userResolveFile.exists()) {
resolveFile = userResolveFile.getAbsolutePath();
}
}
if (resolveFile != null && resolveFile.length() > 0) {
Properties properties = new Properties();
FileInputStream fis = null;
try {
fis = new FileInputStream(new File(resolveFile));
properties.load(fis);
} catch (IOException e) {
throw new IllegalStateException("Unload " + resolveFile + ", cause: " + e.getMessage(), e);
} finally {
try {
if(null != fis) fis.close();
} catch (IOException e) {
logger.warn(e.getMessage(), e);
}
}
resolve = properties.getProperty(interfaceName);
}
}
if (resolve != null && resolve.length() > 0) {
url = resolve;
if (logger.isWarnEnabled()) {
if (resolveFile != null && resolveFile.length() > 0) {
logger.warn("Using default dubbo resolve file " + resolveFile + " replace " + interfaceName + "" + resolve + " to p2p invoke remote service.");
} else {
logger.warn("Using -D" + interfaceName + "=" + resolve + " to p2p invoke remote service.");
}
}
}
//若是application、module、registries、monitor未配置則使用consumer的
if (consumer != null) {
if (application == null) {
application = consumer.getApplication();
}
if (module == null) {
module = consumer.getModule();
}
if (registries == null) {
registries = consumer.getRegistries();
}
if (monitor == null) {
monitor = consumer.getMonitor();
}
}
//若是module已關聯則關聯module的registries和monitor
if (module != null) {
if (registries == null) {
registries = module.getRegistries();
}
if (monitor == null) {
monitor = module.getMonitor();
}
}
//若是application已關聯則關聯application的registries和monitor
if (application != null) {
if (registries == null) {
registries = application.getRegistries();
}
if (monitor == null) {
monitor = application.getMonitor();
}
}
//檢查application
checkApplication();
//檢查遠端和本地服務接口真實存在（是否可load）
checkStubAndMock(interfaceClass);
Map<String, String> map = new HashMap<String, String>();
//配置dubbo的端屬性（是consumer仍是provider）、版本屬性、建立時間、進程號
Map<Object, Object> attributes = new HashMap<Object, Object>();
map.put(Constants.SIDE_KEY, Constants.CONSUMER_SIDE);
map.put(Constants.DUBBO_VERSION_KEY, Version.getVersion());
map.put(Constants.TIMESTAMP_KEY, String.valueOf(System.currentTimeMillis()));
if (ConfigUtils.getPid() > 0) {
map.put(Constants.PID_KEY, String.valueOf(ConfigUtils.getPid()));
}
if (! isGeneric()) {
String revision = Version.getVersion(interfaceClass, version);
if (revision != null && revision.length() > 0) {
map.put("revision", revision);
}
String[] methods = Wrapper.getWrapper(interfaceClass).getMethodNames();
if(methods.length == 0) {
logger.warn("NO method found in service interface " + interfaceClass.getName());
map.put("methods", Constants.ANY_VALUE);
}
else {
map.put("methods", StringUtils.join(new HashSet<String>(Arrays.asList(methods)), ","));
}
}
map.put(Constants.INTERFACE_KEY, interfaceName);
//調用application、module、consumer的get方法將屬性設置到map中
appendParameters(map, application);
appendParameters(map, module);
appendParameters(map, consumer, Constants.DEFAULT_KEY);
appendParameters(map, this);
String prifix = StringUtils.getServiceKey(map);
if (methods != null && methods.size() > 0) {
for (MethodConfig method : methods) {
appendParameters(map, method, method.getName());
String retryKey = method.getName() + ".retry";
if (map.containsKey(retryKey)) {
String retryValue = map.remove(retryKey);
if ("false".equals(retryValue)) {
map.put(method.getName() + ".retries", "0");
}
}
appendAttributes(attributes, method, prifix + "." + method.getName());
checkAndConvertImplicitConfig(method, map, attributes);
}
}
//attributes經過系統context進行存儲.
StaticContext.getSystemContext().putAll(attributes);
//在map裝載了application、module、consumer、reference的全部屬性信息後建立代理
ref = createProxy(map);
}
private T createProxy(Map<String, String> map) {
URL tmpUrl = new URL("temp", "localhost", 0, map);
final boolean isJvmRefer;
if (isInjvm() == null) {
if (url != null && url.length() > 0) { //指定URL的狀況下，不作本地引用
isJvmRefer = false;
} else if (InjvmProtocol.getInjvmProtocol().isInjvmRefer(tmpUrl)) {
//默認狀況下若是本地有服務暴露，則引用本地服務.
isJvmRefer = true;
} else {
isJvmRefer = false;
}
} else {
isJvmRefer = isInjvm().booleanValue();
}
if (isJvmRefer) {
URL url = new URL(Constants.LOCAL_PROTOCOL, NetUtils.LOCALHOST, 0, interfaceClass.getName()).addParameters(map);
invoker = refprotocol.refer(interfaceClass, url);
if (logger.isInfoEnabled()) {
logger.info("Using injvm service " + interfaceClass.getName());
}
} else {
if (url != null && url.length() > 0) { // 用戶指定URL，指定的URL多是對點對直連地址，也多是註冊中心URL
String[] us = Constants.SEMICOLON_SPLIT_PATTERN.split(url);
if (us != null && us.length > 0) {
for (String u : us) {
URL url = URL.valueOf(u);
if (url.getPath() == null || url.getPath().length() == 0) {
url = url.setPath(interfaceName);
}
if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {
urls.add(url.addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map)));
} else {
urls.add(ClusterUtils.mergeUrl(url, map));
}
}
}
} else { // 經過註冊中心配置拼裝URL
List<URL> us = loadRegistries(false);
if (us != null && us.size() > 0) {
for (URL u : us) {
URL monitorUrl = loadMonitor(u);
if (monitorUrl != null) {
map.put(Constants.MONITOR_KEY, URL.encode(monitorUrl.toFullString()));
}
urls.add(u.addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map)));
}
}
if (urls == null || urls.size() == 0) {
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.");
}
}
if (urls.size() == 1) {
//此處舉例說明若是是Dubbo協議則調用DubboProtocol的refer方法生成invoker，當用戶調用service接口實際調用的是invoker的invoke方法
invoker = refprotocol.refer(interfaceClass, urls.get(0));
} else {
//多個service生成多個invoker
List<Invoker<?>> invokers = new ArrayList<Invoker<?>>();
URL registryURL = null;
for (URL url : urls) {
invokers.add(refprotocol.refer(interfaceClass, url));
if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {
registryURL = url; // 用了最後一個registry url
}
}
if (registryURL != null) { // 有 註冊中心協議的URL
// 對有註冊中心的Cluster 只用 AvailableCluster
URL u = registryURL.addParameter(Constants.CLUSTER_KEY, AvailableCluster.NAME);
invoker = cluster.join(new StaticDirectory(u, invokers));
} else { // 不是 註冊中心的URL
invoker = cluster.join(new StaticDirectory(invokers));
}
}
}
Boolean c = check;
if (c == null && consumer != null) {
c = consumer.isCheck();
}
if (c == null) {
c = true; // default true
}
if (c && ! invoker.isAvailable()) {
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());
}
if (logger.isInfoEnabled()) {
logger.info("Refer dubbo service " + interfaceClass.getName() + " from url " + invoker.getUrl());
}
// 建立服務代理
return (T) proxyFactory.getProxy(invoker);
}
至此Reference在關聯了全部application、module、consumer、registry、monitor、service、protocol後調用對應Protocol類的refer方法生成InvokerProxy。當用戶調用service時dubbo會經過InvokerProxy調用Invoker的invoke的方法向服務端發起請求。客戶端就這樣完成了本身的初始化。url
通觀所有dubbo代碼，有兩個很重要的對象就是Invoker和Exporter，Dubbo會根據用戶配置的協議調用不一樣協議的Invoker，再經過ReferenceFonfig將Invoker的引用關聯到Reference的ref屬性上提供給消費端調用。當用戶調用一個Service接口的一個方法後因爲dubbo使用javassist動態代理，會調用Invoker的Invoke方法從而初始化一個RPC調用訪問請求訪問服務端的Service返回結果。spa
相關文章
- 1. Dubbo源碼分析——Dubbo SPI
- 2. Dubbo 源碼分析
- 3. dubbo源碼分析
- 4. dubbo源碼分析系列——dubbo的SPI機制源碼分析
- 5. dubbo源碼分析系列——dubbo-cluster模塊源碼分析
- 6. dubbo源碼分析系列——dubbo-rpc-default模塊源碼分析
- 7. dubbo源碼分析系列——dubbo-rpc-api模塊源碼分析
- 8. dubbo源碼分析系列——dubbo-register-api模塊源碼分析
- 9. dubbo源碼分析--dubbo spi解析
- 10. Dubbo實現源碼分析
- 更多相關文章...
- • Docker 資源彙總 - Docker教程
- • Java操作Neo4j數據庫（附帶源碼） - NoSQL教程
- • 互聯網組織的未來：剖析GitHub員工的任性之源
- • Java Agent入門實戰（二）-Instrumentation源碼概述
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
歡迎關注本站公眾號,獲取更多信息
