源码分析Dubbo服务消费端启动流程

经过前面文章详解,咱们知道Dubbo服务消费者标签dubbo:reference最终会在Spring容器中建立一个对应的ReferenceBean实例,而ReferenceBean实现了Spring生命周期接口:InitializingBean,接下来应该看一下其afterPropertiesSet方法的实现。spring

一、源码分析ReferenceBean#afterPropertiesSet

ReferenceBean#afterPropertiesSet缓存

if (getConsumer() == null) {
            Map<string, consumerconfig> consumerConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, 
                 ConsumerConfig.class, false, false);
            if (consumerConfigMap != null &amp;&amp; consumerConfigMap.size() &gt; 0) {
                ConsumerConfig consumerConfig = null;
                for (ConsumerConfig config : consumerConfigMap.values()) {
                    if (config.isDefault() == null || config.isDefault().booleanValue()) {
                        if (consumerConfig != null) {
                            throw new IllegalStateException("Duplicate consumer configs: " + consumerConfig + " and " + config);
                        }
                        consumerConfig = config;
                    }
                }
                if (consumerConfig != null) {
                    setConsumer(consumerConfig);
                }
            }
        }

Step1:若是consumer为空,说明dubbo:reference标签未设置consumer属性,若是一个dubbo:consumer标签,则取该实例,若是存在多个dubbo:consumer 配置,则consumer必须设置,不然会抛出异常:"Duplicate consumer configs"。架构

Step2:若是application为空,则尝试从BeanFactory中查询dubbo:application实例,若是存在多个dubbo:application配置,则抛出异常:"Duplicate application configs"。并发

Step3:若是ServiceBean的module为空,则尝试从BeanFactory中查询dubbo:module实例,若是存在多个dubbo:module,则抛出异常:"Duplicate module configs: "。app

Step4:尝试从BeanFactory中加载全部的注册中心,注意ServiceBean的List< RegistryConfig&gt; registries属性,为注册中心集合。jvm

Step5:尝试从BeanFacotry中加载一个监控中心,填充ServiceBean的MonitorConfig monitor属性,若是存在多个dubbo:monitor配置,则抛出"Duplicate monitor configs: "。分布式

ReferenceBean#afterPropertiesSetide

Boolean b = isInit();
if (b == null &amp;&amp; getConsumer() != null) {
      b = getConsumer().isInit();
}
if (b != null &amp;&amp; b.booleanValue()) {
      getObject();
}

Step6:判断是否初始化,若是为初始化,则调用getObject()方法,该方法也是FactoryBean定义的方法,ReferenceBean是dubbo:reference所真实引用的类(interface)的实例工程,getObject发返回的是interface的实例,而不是ReferenceBean实例。高并发

1.1 源码分析getObject()

public Object getObject() throws Exception {
        return get();
}

ReferenceBean#getObject()方法直接调用其父类的get方法,get方法内部调用init()方法进行初始化源码分析

1.2 源码分析ReferenceConfig#init方法

ReferenceConfig#init

if (initialized) {
     return;
 }
initialized = true;
if (interfaceName == null || interfaceName.length() == 0) {
      throw new IllegalStateException("<dubbo:reference interface="\&quot;\&quot;" /> interface not allow null!");
}

Step1:若是已经初始化,直接返回,若是interfaceName为空,则抛出异常。

ReferenceConfig#init调用ReferenceConfig#checkDefault

private void checkDefault() {
        if (consumer == null) {
            consumer = new ConsumerConfig();
        }
        appendProperties(consumer);
    }

Step2:若是dubbo:reference标签也就是ReferenceBean的consumer属性为空,调用appendProperties方法,填充默认属性,其具体加载顺序:

  1. 从系统属性加载对应参数值,参数键:dubbo.consumer.属性名,从系统属性中获取属性值的方法为:System.getProperty(key)。
  2. 加载属性配置文件的值。属性配置文件,可经过系统属性:dubbo.properties.file,若是该值未配置,则默认取dubbo.properties属性配置文件。 ReferenceConfig#init
appendProperties(this);

Step3:调用appendProperties方法,填充ReferenceBean的属性,属性值来源与step2同样,固然只填充ReferenceBean中属性为空的属性。

ReferenceConfig#init

if (getGeneric() == null &amp;&amp; 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(e.getMessage(), e);
      }
      checkInterfaceAndMethods(interfaceClass, methods);
}

Step4:若是使用返回引用,将interface值替换为GenericService全路径名,若是不是,则加载interfacename,并检验dubbo:reference子标签dubbo:method引用的方法是否在interface指定的接口中存在。

ReferenceConfig#init

String resolve = System.getProperty(interfaceName);      // @1
String resolveFile = null;
if (resolve == null || resolve.length() == 0) {                       // @2
     resolveFile = System.getProperty("dubbo.resolve.file");    // @3 start
     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();
          }
     }    // @3 end
     if (resolveFile != null &amp;&amp; resolveFile.length() &gt; 0) {    // @4
          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 &amp;&amp; resolve.length() &gt; 0) {  // @5
      url = resolve;
      if (logger.isWarnEnabled()) {
         if (resolveFile != null &amp;&amp; resolveFile.length() &gt; 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.");
        }
    }
}

Step5:处理dubbo服务消费端resolve机制,也就是说消息消费者只连服务提供者,绕过注册中心。

代码@1:从系统属性中获取该接口的直连服务提供者,若是存在 -Dinterface=dubbo://127.0.0.1:20880,其中interface为dubbo:reference interface属性的值。

代码@2:若是未指定-D属性,尝试从resolve配置文件中查找,从这里看出-D的优先级更高。

代码@3:首先尝试获取resolve配置文件的路径,其来源能够经过-Ddubbo.resolve.file=文件路径名来指定,若是未配置该系统参数,则默认从${user.home}/dubbo-resolve.properties,若是过文件存在,则设置resolveFile的值,不然resolveFile为null。

代码@4:若是resolveFile不为空,则加载resolveFile文件中内容,而后经过interface获取其配置的直连服务提供者URL。

代码@5:若是resolve不为空,则填充ReferenceBean的url属性为resolve(点对点服务提供者URL),打印日志,点对点URL的来源(系统属性、resolve配置文件)。

ReferenceConfig#init

checkApplication();
checkStubAndMock(interfaceClass);

Step6:校验ReferenceBean的application是否为空,若是为空,new 一个application,并尝试从系统属性(优先)、资源文件中填充其属性;同时校验stub、mock实现类与interface的兼容性。系统属性、资源文件属性的配置以下: application dubbo.application.属性名,例如 dubbo.application.name

ReferenceConfig#init

Map<string, string> map = new HashMap<string, string>();
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() &gt; 0) {
     map.put(Constants.PID_KEY, String.valueOf(ConfigUtils.getPid()));
}

Step7:构建Map,封装服务消费者引用服务提供者URL的属性,这里主要填充side:consume(消费端)、dubbo:2.0.0(版本)、timestamp、pid:进程ID。

ReferenceConfig#init

if (!isGeneric()) {
    String revision = Version.getVersion(interfaceClass, version);
    if (revision != null &amp;&amp; revision.length() &gt; 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)), ","));
     }

}

Step8:若是不是泛化引用,增长methods:interface的全部方法名,多个用逗号隔开。 ReferenceConfig#init

map.put(Constants.INTERFACE_KEY, interfaceName);
appendParameters(map, application);
appendParameters(map, module);
appendParameters(map, consumer, Constants.DEFAULT_KEY);
appendParameters(map, this);

Step9:用Map存储application配置、module配置、默认消费者参数(ConsumerConfig)、服务消费者dubbo:reference的属性。 ReferenceConfig#init

String prefix = StringUtils.getServiceKey(map);
if (methods != null &amp;&amp; !methods.isEmpty()) {
    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, prefix + "." + method.getName());
         checkAndConvertImplicitConfig(method, map, attributes);
   }
}

Step10:获取服务键值 /{group}/interface:版本,若是group为空,则为interface:版本,其值存为prifex,而后将dubbo:method的属性名称也填入map中,键前缀为dubbo.method.methodname.属性名。dubbo:method的子标签dubbo:argument标签的属性也追加到attributes map中,键为 prifex + methodname.属性名。

ReferenceConfig#init

String hostToRegistry = ConfigUtils.getSystemProperty(Constants.DUBBO_IP_TO_REGISTRY);
if (hostToRegistry == null || hostToRegistry.length() == 0) {
      hostToRegistry = NetUtils.getLocalHost();
} else if (isInvalidLocalHost(hostToRegistry)) {
      throw new IllegalArgumentException("Specified invalid registry ip from property:" + Constants.DUBBO_IP_TO_REGISTRY + ", value:" + 
           hostToRegistry);
}
map.put(Constants.REGISTER_IP_KEY, hostToRegistry);

Step11:填充register.ip属性,该属性是消息消费者链接注册中心的IP,并非注册中心自身的IP。 ReferenceConfig#init

ref = createProxy(map);

Step12:调用createProxy方法建立消息消费者代理,下面详细分析其实现细节。 ReferenceConfig#init

ConsumerModel consumerModel = new ConsumerModel(getUniqueServiceName(), this, ref, interfaceClass.getMethods());
ApplicationModel.initConsumerModel(getUniqueServiceName(), consumerModel);

Step13:将消息消费者缓存在ApplicationModel中。

1.2.1 源码分析ReferenceConfig#createProxy方法

ReferenceConfig#createProxy

URL tmpUrl = new URL("temp", "localhost", 0, map);
final boolean isJvmRefer;
if (isInjvm() == null) {
    if (url != null &amp;&amp; url.length() &gt; 0) { // if a url is specified, don't do local reference
          isJvmRefer = false;
    } else if (InjvmProtocol.getInjvmProtocol().isInjvmRefer(tmpUrl)) {
         // by default, reference local service if there is
         isJvmRefer = true;
    } else {
         isJvmRefer = false;
    }
} else {
    isJvmRefer = isInjvm().booleanValue();
}

Step1:判断该消费者是不是引用本(JVM)内提供的服务。 若是dubbo:reference标签的injvm(已过时,被local属性替换)若是不为空,则直接取该值,若是该值未配置,则判断ReferenceConfig的url属性是否为空,若是不为空,则isJvmRefer =false,代表该服务消费者将直连该URL的服务提供者;若是url属性为空,则判断该协议是不是isInjvm,其实现逻辑:获取dubbo:reference的scop属性,根据其值判断:

  • 若是为空,isJvmRefer为false。
  • 若是协议为injvm,就是表示为本地协议,既然提供了本地协议的实现,则无需配置isJvmRefer该标签为true,故,isJvmRerfer=false。
  • 若是scope=local或injvm=true,isJvmRefer=true。
  • 若是scope=remote,isJvmRefer设置为false。
  • 若是是泛化引用,isJvmRefer设置为false。
  • 其余默认状况,isJvmRefer设置为true。

ReferenceConfig#createProxy

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());
   }
}

Step2:若是消费者引用本地JVM中的服务,则利用InjvmProtocol建立Invoker,dubbo中的invoker主要负责服务调用的功能,是其核心实现,后续会在专门的章节中详细分析,在这里咱们须要知道,会建立于协议相关的Invoker便可。

ReferenceConfig#createProxy

if (url != null &amp;&amp; url.length() &gt; 0) { // user specified URL, could be peer-to-peer address, or register center's address.
     String[] us = Constants.SEMICOLON_SPLIT_PATTERN.split(url);   // @1
     if (us != null &amp;&amp; us.length &gt; 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())) {   // @2
                      urls.add(url.addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map)));
                  } else {
                      urls.add(ClusterUtils.mergeUrl(url, map));  // @3
                 }
             }
       }
}

Step3:处理直连状况,与step2互斥。

代码@1:对直连URL进行分割,多个直连URL用分号隔开,若是URL中不包含path属性,则为URL设置path属性为interfaceName。

代码@2:若是直连提供者的协议为registry,则对url增长refer属性,其值为消息消费者全部的属性。(表示从注册中心发现服务提供者)

代码@3:若是是其余协议提供者,则合并服务提供者与消息消费者的属性,并移除服务提供者默认属性。以default开头的属性。

ReferenceConfig#createProxy

List<url> us = loadRegistries(false);   // @1
if (us != null &amp;&amp; !us.isEmpty()) {
     for (URL u : us) {
           URL monitorUrl = loadMonitor(u);   // @2
           if (monitorUrl != null) {
                 map.put(Constants.MONITOR_KEY, URL.encode(monitorUrl.toFullString()));   // @3
            }
            urls.add(u.addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map)));  // @4
       }
}
if (urls == null || urls.isEmpty()) {
       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="\&quot;...\&quot;" /> to your spring config.");
}

Step4:普通消息消费者,从注册中心订阅服务。

代码@1:获取全部注册中心URL,其中参数false表示消费端,须要排除dubbo:registry subscribe=false的注册中心,其值为false表示不接受订阅。

代码@2:根据注册中心URL,构建监控中心URL。

代码@3:若是监控中心不为空,在注册中心URL后增长属性monitor。

代码@4:在注册中心URL中,追加属性refer,其值为消费端的全部配置组成的URL。

ReferenceConfig#createProxy

if (urls.size() == 1) {
    invoker = refprotocol.refer(interfaceClass, urls.get(0));     // @1
} else {
    List<invoker<?>&gt; invokers = new ArrayList<invoker<?>&gt;();    // @2,多个服务提供者URL,集群模式
    URL registryURL = null;
    for (URL url : urls) {
         invokers.add(refprotocol.refer(interfaceClass, url));    // @2
         if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {
               registryURL = url; // use last registry url
          }
     }
     if (registryURL != null) { // registry url is available
          // use AvailableCluster only when register's cluster is available
          URL u = registryURL.addParameter(Constants.CLUSTER_KEY, AvailableCluster.NAME);
          invoker = cluster.join(new StaticDirectory(u, invokers));    // @3
     } else { // not a registry url
          invoker = cluster.join(new StaticDirectory(invokers));
     }
 }

Step5:根据URL获取对应协议的Invoker。 代码@1:若是只有一个服务提供者URL,则直接根据协议构建Invoker,具体有以下协议: 这里写图片描述

代码@2:若是有多个服务提供者,则众多服务提供者构成一个集群。 首先根据协议构建服务Invoker,默认Dubbo基于服务注册于发现,在服务消费端不会指定url属性,从注册中心获取服务提供者列表,此时的URL:registry://开头,url中会包含register属性,其值为注册中心的类型,例如zookeeper,将使用RedisProtocol构建Invoker,该方法将自动发现注册在注册中心的服务提供者,后续文章将会zookeeper注册中心为例,详细分析其实现原理。 代码@3:返回集群模式实现的Invoker,Dubbo中的Invoker类继承体系以下: 这里写图片描述

集群模式的Invoker和单个协议Invoker同样实现Invoker接口,而后在集群Invoker中利用Directory保证一个一个协议的调用器,十分的巧妙,在后续章节中将重点分析Dubbo Invoker实现原理,包含集群实现机制。

ReferenceConfig#createProxy

Boolean c = check;
if (c == null &amp;&amp; consumer != null) {
      c = consumer.isCheck();
}
if (c == null) {
      c = true; // default true
}
if (c &amp;&amp; !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());
}

代码@4:若是dubbo:referecnce的check=true或默认为空,则须要判断服务提供者是否存在。

ReferenceConfig#createProxy

return (T) proxyFactory.getProxy(invoker);
AbstractProxyFactory#getProxy
public <t> T getProxy(Invoker<t> invoker) throws RpcException {
        Class<!--?-->[] interfaces = null;
        String config = invoker.getUrl().getParameter("interfaces");     // @1
        if (config != null &amp;&amp; config.length() &gt; 0) {
            String[] types = Constants.COMMA_SPLIT_PATTERN.split(config);
            if (types != null &amp;&amp; types.length &gt; 0) {
                interfaces = new Class<!--?-->[types.length + 2];
                interfaces[0] = invoker.getInterface();
                interfaces[1] = EchoService.class;        // @2
                for (int i = 0; i &lt; types.length; i++) {
                    interfaces[i + 1] = ReflectUtils.forName(types[i]);
                }
            }
        }
        if (interfaces == null) {
            interfaces = new Class<!--?-->[]{invoker.getInterface(), EchoService.class};
        }
        return getProxy(invoker, interfaces);    // @3
    }

根据invoker获取代理类,其实现逻辑以下:

代码@1:从消费者URL中获取interfaces的值,用,分隔出单个服务应用接口。

代码@2:增长默认接口EchoService接口。

代码@3:根据须要实现的接口,使用jdk或Javassist建立代理类。 最后给出消息消费者启动时序图: 这里写图片描述

本节关于Dubbo服务消费者(服务调用者)的启动流程就梳理到这里,下一篇将重点关注Invoker(服务调用相关的实现细节)。


做者介绍:丁威,《RocketMQ技术内幕》做者,RocketMQ 社区布道师,公众号:中间件兴趣圈 维护者,目前已陆续发表源码分析Java集合、Java 并发包(JUC)、Netty、Mycat、Dubbo、RocketMQ、Mybatis等源码专栏。能够点击连接:中间件知识星球,一块儿探讨高并发、分布式服务架构,交流源码。

</t></t></invoker<?></invoker<?></url></string></object,></object,></string,></string,></string,>