Dubbo源码学习--服务引用

痛定思痛。 2022-05-27 01:22 314阅读 0赞

服务引用是服务的消费者向注册中心订阅服务提供者的消息，根据java的接口编程获取代理接口远程调用服务提供者。

  1. <!-- consumer's application name, used for tracing dependency relationship (not a matching criterion),
  2.     don't set it same as provider -->
  3.     <dubbo:application name="demo-consumer"/>
  5.     <!-- use multicast registry center to discover service -->
  6.     <dubbo:registry address="zookeeper://127.0.0.1:2181"/>
  8.     <!-- generate proxy for the remote service, then demoService can be used in the same way as the
  9.     local regular interface -->
  10.     <dubbo:reference id="demoService" check="false" interface="com.alibaba.dubbo.demo.DemoService"/>

服务消费者配置如上配置，服务消费者初始化的Bean实体为ReferenceBean

70 

  1. ReferenceBean除了继承了配置的抽象类来处理配置信息外，它还实现spring容器的一些接口，这里我们分析下FactoryBean, 首先它对于spring来说是个bean，参与Bean创建的所有生命周期，关键在于spring的bean工厂beanFactory.getBean(“demoService”)获取的bean的时候会判断下是不是FactoryBean的实例，如果是调factoryBean.getObject()返回，否则返回bean。  
  4. 我们对于远程调用获取的demoService其实并不是想要ReferenceBean这个对象实例本身，我们是想获取对远程调用的代理，能够通过这个代理服务调用远程服务。这里就是通过factoryBean.getObject()来创建引用返回基于DemoService接口的代理给引用，对用户透明dubbo封装了复杂实现。

接下来我们对消费者的初始化流程通过代码来了解一下：

ReferenceBean在初始化的为bean的时候会调用afterPropertiesSet函数，在afterPropertiesSet中完成了消费者的一些配置的初始化操作 

  1. @SuppressWarnings({"unchecked"})
  2.     public void afterPropertiesSet() throws Exception {
  3.         //获取consumer标签相关配置
  4.         if (getConsumer() == null) {
  5.             Map<String, ConsumerConfig> consumerConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ConsumerConfig.class, false, false);
  6.             if (consumerConfigMap != null && consumerConfigMap.size() > 0) {
  7.                 ConsumerConfig consumerConfig = null;
  8.                 for (ConsumerConfig config : consumerConfigMap.values()) {
  9.                     if (config.isDefault() == null || config.isDefault().booleanValue()) {
  10.                         if (consumerConfig != null) {
  11.                             throw new IllegalStateException("Duplicate consumer configs: " + consumerConfig + " and " + config);
  12.                         }
  13.                         consumerConfig = config;
  14.                     }
  15.                 }
  16.                 if (consumerConfig != null) {
  17.                     setConsumer(consumerConfig);
  18.                 }
  19.             }
  20.         }
  21.         //获取application标签相关配置
  22.         if (getApplication() == null
  23.                 && (getConsumer() == null || getConsumer().getApplication() == null)) {
  24.             Map<String, ApplicationConfig> applicationConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ApplicationConfig.class, false, false);
  25.             if (applicationConfigMap != null && applicationConfigMap.size() > 0) {
  26.                 ApplicationConfig applicationConfig = null;
  27.                 for (ApplicationConfig config : applicationConfigMap.values()) {
  28.                     if (config.isDefault() == null || config.isDefault().booleanValue()) {
  29.                         if (applicationConfig != null) {
  30.                             throw new IllegalStateException("Duplicate application configs: " + applicationConfig + " and " + config);
  31.                         }
  32.                         applicationConfig = config;
  33.                     }
  34.                 }
  35.                 if (applicationConfig != null) {
  36.                     setApplication(applicationConfig);
  37.                 }
  38.             }
  39.         }
  40.         //获取module标签相关配置
  41.         if (getModule() == null
  42.                 && (getConsumer() == null || getConsumer().getModule() == null)) {
  43.             Map<String, ModuleConfig> moduleConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ModuleConfig.class, false, false);
  44.             if (moduleConfigMap != null && moduleConfigMap.size() > 0) {
  45.                 ModuleConfig moduleConfig = null;
  46.                 for (ModuleConfig config : moduleConfigMap.values()) {
  47.                     if (config.isDefault() == null || config.isDefault().booleanValue()) {
  48.                         if (moduleConfig != null) {
  49.                             throw new IllegalStateException("Duplicate module configs: " + moduleConfig + " and " + config);
  50.                         }
  51.                         moduleConfig = config;
  52.                     }
  53.                 }
  54.                 if (moduleConfig != null) {
  55.                     setModule(moduleConfig);
  56.                 }
  57.             }
  58.         }
  59.         //获取registry相关的标签配置
  60.         if ((getRegistries() == null || getRegistries().isEmpty())
  61.                 && (getConsumer() == null || getConsumer().getRegistries() == null || getConsumer().getRegistries().isEmpty())
  62.                 && (getApplication() == null || getApplication().getRegistries() == null || getApplication().getRegistries().isEmpty())) {
  63.             Map<String, RegistryConfig> registryConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, RegistryConfig.class, false, false);
  64.             if (registryConfigMap != null && registryConfigMap.size() > 0) {
  65.                 List<RegistryConfig> registryConfigs = new ArrayList<RegistryConfig>();
  66.                 for (RegistryConfig config : registryConfigMap.values()) {
  67.                     if (config.isDefault() == null || config.isDefault().booleanValue()) {
  68.                         registryConfigs.add(config);
  69.                     }
  70.                 }
  71.                 if (registryConfigs != null && !registryConfigs.isEmpty()) {
  72.                     super.setRegistries(registryConfigs);
  73.                 }
  74.             }
  75.         }
  76.         //获取monitor相关的标签配置
  77.         if (getMonitor() == null
  78.                 && (getConsumer() == null || getConsumer().getMonitor() == null)
  79.                 && (getApplication() == null || getApplication().getMonitor() == null)) {
  80.             Map<String, MonitorConfig> monitorConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, MonitorConfig.class, false, false);
  81.             if (monitorConfigMap != null && monitorConfigMap.size() > 0) {
  82.                 MonitorConfig monitorConfig = null;
  83.                 for (MonitorConfig config : monitorConfigMap.values()) {
  84.                     if (config.isDefault() == null || config.isDefault().booleanValue()) {
  85.                         if (monitorConfig != null) {
  86.                             throw new IllegalStateException("Duplicate monitor configs: " + monitorConfig + " and " + config);
  87.                         }
  88.                         monitorConfig = config;
  89.                     }
  90.                 }
  91.                 if (monitorConfig != null) {
  92.                     setMonitor(monitorConfig);
  93.                 }
  94.             }
  95.         }
  96.         //如果配置了初始化，在ReferenceBean构造完成之后就初始化消费者
  97.         Boolean b = isInit();
  98.         if (b == null && getConsumer() != null) {
  99.             b = getConsumer().isInit();
  100.         }
  101.         if (b != null && b.booleanValue()) {
  102.             getObject();
  103.         }
  104.     }

在getObject中获取代理类

  1. //获取代理类
  2.   public Object getObject() throws Exception {
  3.         return get();
  4.     }

在get方法中会调用init函数，完成消费者代理类的初始化操作

  1. public synchronized T get() {
  2.         if (destroyed) {
  3.             throw new IllegalStateException("Already destroyed!");
  4.         }
  5.         //初始化消费者，
  6.         if (ref == null) {
  7.             init();
  8.         }
  9.         return ref;
  10.     }

init方法中会初始化一些消费者相关的配置信息，在createProxy中完成消费者注册和订阅服务提供者，返回代理类

  1. private void init() {
  2.         if (initialized) {
  3.             return;
  4.         }
  5.         initialized = true;
  6.         if (interfaceName == null || interfaceName.length() == 0) {
  7.             throw new IllegalStateException("<dubbo:reference interface=\"\" /> interface not allow null!");
  8.         }
  9.         // get consumer's global configuration
  10.         checkDefault();
  11.         appendProperties(this);
  12.         if (getGeneric() == null && getConsumer() != null) {
  13.             setGeneric(getConsumer().getGeneric());
  14.         }
  15.         if (ProtocolUtils.isGeneric(getGeneric())) {
  16.             interfaceClass = GenericService.class;
  17.         } else {
  18.             try {
  19.                 interfaceClass = Class.forName(interfaceName, true, Thread.currentThread()
  20.                         .getContextClassLoader());
  21.             } catch (ClassNotFoundException e) {
  22.                 throw new IllegalStateException(e.getMessage(), e);
  23.             }
  24.             checkInterfaceAndMethods(interfaceClass, methods);
  25.         }
  26.         String resolve = System.getProperty(interfaceName);
  27.         String resolveFile = null;
  28.         if (resolve == null || resolve.length() == 0) {
  29.             resolveFile = System.getProperty("dubbo.resolve.file");
  30.             if (resolveFile == null || resolveFile.length() == 0) {
  31.                 File userResolveFile = new File(new File(System.getProperty("user.home")), "dubbo-resolve.properties");
  32.                 if (userResolveFile.exists()) {
  33.                     resolveFile = userResolveFile.getAbsolutePath();
  34.                 }
  35.             }
  36.             if (resolveFile != null && resolveFile.length() > 0) {
  37.                 Properties properties = new Properties();
  38.                 FileInputStream fis = null;
  39.                 try {
  40.                     fis = new FileInputStream(new File(resolveFile));
  41.                     properties.load(fis);
  42.                 } catch (IOException e) {
  43.                     throw new IllegalStateException("Unload " + resolveFile + ", cause: " + e.getMessage(), e);
  44.                 } finally {
  45.                     try {
  46.                         if (null != fis) fis.close();
  47.                     } catch (IOException e) {
  48.                         logger.warn(e.getMessage(), e);
  49.                     }
  50.                 }
  51.                 resolve = properties.getProperty(interfaceName);
  52.             }
  53.         }
  54.         if (resolve != null && resolve.length() > 0) {
  55.             url = resolve;
  56.             if (logger.isWarnEnabled()) {
  57.                 if (resolveFile != null && resolveFile.length() > 0) {
  58.                     logger.warn("Using default dubbo resolve file " + resolveFile + " replace " + interfaceName + "" + resolve + " to p2p invoke remote service.");
  59.                 } else {
  60.                     logger.warn("Using -D" + interfaceName + "=" + resolve + " to p2p invoke remote service.");
  61.                 }
  62.             }
  63.         }
  64.         //消费者注册时的一些信息
  65.         if (consumer != null) {
  66.             if (application == null) {
  67.                 application = consumer.getApplication();
  68.             }
  69.             if (module == null) {
  70.                 module = consumer.getModule();
  71.             }
  72.             if (registries == null) {
  73.                 registries = consumer.getRegistries();
  74.             }
  75.             if (monitor == null) {
  76.                 monitor = consumer.getMonitor();
  77.             }
  78.         }
  79.         if (module != null) {
  80.             if (registries == null) {
  81.                 registries = module.getRegistries();
  82.             }
  83.             if (monitor == null) {
  84.                 monitor = module.getMonitor();
  85.             }
  86.         }
  87.         if (application != null) {
  88.             if (registries == null) {
  89.                 registries = application.getRegistries();
  90.             }
  91.             if (monitor == null) {
  92.                 monitor = application.getMonitor();
  93.             }
  94.         }
  95.         checkApplication();
  96.         checkStubAndMock(interfaceClass);
  97.         //将消费者的一些配置信息添加到map中
  98.         Map<String, String> map = new HashMap<String, String>();
  99.         Map<Object, Object> attributes = new HashMap<Object, Object>();
  100.         map.put(Constants.SIDE_KEY, Constants.CONSUMER_SIDE);
  101.         map.put(Constants.DUBBO_VERSION_KEY, Version.getVersion());
  102.         map.put(Constants.TIMESTAMP_KEY, String.valueOf(System.currentTimeMillis()));
  103.         if (ConfigUtils.getPid() > 0) {
  104.             map.put(Constants.PID_KEY, String.valueOf(ConfigUtils.getPid()));
  105.         }
  106.         if (!isGeneric()) {
  107.             String revision = Version.getVersion(interfaceClass, version);
  108.             if (revision != null && revision.length() > 0) {
  109.                 map.put("revision", revision);
  110.             }
  112.             String[] methods = Wrapper.getWrapper(interfaceClass).getMethodNames();
  113.             if (methods.length == 0) {
  114.                 logger.warn("NO method found in service interface " + interfaceClass.getName());
  115.                 map.put("methods", Constants.ANY_VALUE);
  116.             } else {
  117.                 map.put("methods", StringUtils.join(new HashSet<String>(Arrays.asList(methods)), ","));
  118.             }
  119.         }
  120.         map.put(Constants.INTERFACE_KEY, interfaceName);
  121.         appendParameters(map, application);
  122.         appendParameters(map, module);
  123.         appendParameters(map, consumer, Constants.DEFAULT_KEY);
  124.         appendParameters(map, this);
  125.         String prefix = StringUtils.getServiceKey(map);
  126.         if (methods != null && !methods.isEmpty()) {
  127.             for (MethodConfig method : methods) {
  128.                 appendParameters(map, method, method.getName());
  129.                 String retryKey = method.getName() + ".retry";
  130.                 if (map.containsKey(retryKey)) {
  131.                     String retryValue = map.remove(retryKey);
  132.                     if ("false".equals(retryValue)) {
  133.                         map.put(method.getName() + ".retries", "0");
  134.                     }
  135.                 }
  136.                 appendAttributes(attributes, method, prefix + "." + method.getName());
  137.                 checkAndConvertImplicitConfig(method, map, attributes);
  138.             }
  139.         }
  141.         String hostToRegistry = ConfigUtils.getSystemProperty(Constants.DUBBO_IP_TO_REGISTRY);
  142.         if (hostToRegistry == null || hostToRegistry.length() == 0) {
  143.             hostToRegistry = NetUtils.getLocalHost();
  144.         } else if (isInvalidLocalHost(hostToRegistry)) {
  145.             throw new IllegalArgumentException("Specified invalid registry ip from property:" + Constants.DUBBO_IP_TO_REGISTRY + ", value:" + hostToRegistry);
  146.         }
  147.         map.put(Constants.REGISTER_IP_KEY, hostToRegistry);
  149.         //attributes are stored by system context.
  150.         StaticContext.getSystemContext().putAll(attributes);
  151.         //根据消费者配置信息完成消费者代理类的初始化操作
  152.         ref = createProxy(map);
  153.         ConsumerModel consumerModel = new ConsumerModel(getUniqueServiceName(), this, ref, interfaceClass.getMethods());
  154.         ApplicationModel.initConsumerModel(getUniqueServiceName(), consumerModel);
  155.     }

在createProxy中首先根据消费者的相关配置将消费者添加到注册中心，并订阅相关通知，最终返回invoker调用者，将invoker包装成代理类返回到spring容器中

  1. 在createProxy中首先根据消费者的相关配置将消费者添加到注册中心，并订阅相关通知，最终返回invoker调用者，将invoker包装成代理类返回到spring容器中
  3.  private T createProxy(Map<String, String> map) {
  4.         //获取消费者url信息
  5.         URL tmpUrl = new URL("temp", "localhost", 0, map);
  6.         final boolean isJvmRefer;
  7.         //判断是否是同一个jvm，如果是则可以直接本地调用
  8.         if (isInjvm() == null) {
  9.             if (url != null && url.length() > 0) { // if a url is specified, don't do local reference
  10.                 isJvmRefer = false;
  11.             } else if (InjvmProtocol.getInjvmProtocol().isInjvmRefer(tmpUrl)) {
  12.                 // by default, reference local service if there is
  13.                 isJvmRefer = true;
  14.             } else {
  15.                 isJvmRefer = false;
  16.             }
  17.         } else {
  18.             isJvmRefer = isInjvm().booleanValue();
  19.         }
  21.         if (isJvmRefer) {
  22.             URL url = new URL(Constants.LOCAL_PROTOCOL, NetUtils.LOCALHOST, 0, interfaceClass.getName()).addParameters(map);
  23.             invoker = refprotocol.refer(interfaceClass, url);
  24.             if (logger.isInfoEnabled()) {
  25.                 logger.info("Using injvm service " + interfaceClass.getName());
  26.             }
  27.         } else {
  28.             if (url != null && url.length() > 0) { // user specified URL, could be peer-to-peer address, or register center's address.
  29.                 String[] us = Constants.SEMICOLON_SPLIT_PATTERN.split(url);
  30.                 if (us != null && us.length > 0) {
  31.                     for (String u : us) {
  32.                         URL url = URL.valueOf(u);
  33.                         if (url.getPath() == null || url.getPath().length() == 0) {
  34.                             url = url.setPath(interfaceName);
  35.                         }
  36.                         if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {
  37.                             urls.add(url.addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map)));
  38.                         } else {
  39.                             urls.add(ClusterUtils.mergeUrl(url, map));
  40.                         }
  41.                     }
  42.                 }
  43.             } else { // assemble URL from register center's configuration
  44.                 List<URL> us = loadRegistries(false);
  45.                 if (us != null && !us.isEmpty()) {
  46.                     for (URL u : us) {
  47.                         URL monitorUrl = loadMonitor(u);
  48.                         if (monitorUrl != null) {
  49.                             map.put(Constants.MONITOR_KEY, URL.encode(monitorUrl.toFullString()));
  50.                         }
  51.                         urls.add(u.addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map)));
  52.                     }
  53.                 }
  54.                 if (urls == null || urls.isEmpty()) {
  55.                     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.");
  56.                 }
  57.             }
  58.             //将消费者信息注册到注册中心，并订阅注册中心相关接口，最终获取调用器
  59.             if (urls.size() == 1) {
  60.                 invoker = refprotocol.refer(interfaceClass, urls.get(0));
  61.             } else {
  62.                 List<Invoker<?>> invokers = new ArrayList<Invoker<?>>();
  63.                 URL registryURL = null;
  64.                 for (URL url : urls) {
  65.                     invokers.add(refprotocol.refer(interfaceClass, url));
  66.                     if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {
  67.                         registryURL = url; // use last registry url
  68.                     }
  69.                 }
  70.                 if (registryURL != null) { // registry url is available
  71.                     // use AvailableCluster only when register's cluster is available
  72.                     URL u = registryURL.addParameter(Constants.CLUSTER_KEY, AvailableCluster.NAME);
  73.                     invoker = cluster.join(new StaticDirectory(u, invokers));
  74.                 } else { // not a registry url
  75.                     invoker = cluster.join(new StaticDirectory(invokers));
  76.                 }
  77.             }
  78.         }
  80.         Boolean c = check;
  81.         if (c == null && consumer != null) {
  82.             c = consumer.isCheck();
  83.         }
  84.         if (c == null) {
  85.             c = true; // default true
  86.         }
  87.         if (c && !invoker.isAvailable()) {
  88.             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());
  89.         }
  90.         if (logger.isInfoEnabled()) {
  91.             logger.info("Refer dubbo service " + interfaceClass.getName() + " from url " + invoker.getUrl());
  92.         }
  93.         // create service proxy
  94.         //根据调用器创建代理类
  95.         return (T) proxyFactory.getProxy(invoker);
  96.     }

创建代理的过程：

  1. 获取消费者配置

  2. 获取配置的注册中心，通过配置中心配置拼装URL，线上应该是个配置中心集群

  3. 遍历注册中心List集合

加载监控中心URL，如果配置了监控中心在注册中心url加上MONITOR_KEY

根据配置的引用服务参数给注册中URL上加上REFER_KEY

  1. 遍历注册中心List集合，这里注册中心url包含了monitorUrl和referUrl

protocol.refer(interface,url)调用protocol引用服务返回invoker可执行对象（这个invoker并不是简单的DubboInvoker, 而是由RegistryProtocol构建基于目录服务的集群策略Invoker, 这个invoker可以通过目录服务list出真正可调用的远程服务invoker）

对于注册中心Url设置集群策略为AvailableCluster， 由AvailableCluster将所有对象注册中调用的invoker伪装成一个invoker

  1. 通过代理工厂创建远程服务代理返回给使用着proxyFactory.getProxy(invoker);

procotol.refer(interface, url) 引用服务的过程

  1. 经过ProtocolListenerWrapper， ProtocolFilterWrapper由于是注册中心url调用RegistryProtocol.refer

  2. 获取注册中心协议zookeeper, redis, 还是dubbo， 并根据注册中心协议通过注册器工厂RegistryFactory.getRegistry(url)获取注册器Registry用来跟注册中心交互

  3. 根据配置的group分组

  4. 创建注册服务目录RegistryDirectory并设置注册器

  5. 构建订阅服务的subscribeUrl

  6. 通过注册器registry向注册中心注册subscribeUrl消费端url

  7. 目录服务registryDirectory.subscribe(subscribeUrl)订阅服务（这里我们以开源版本zookeeper为注册中心为例来讲解， dubbo协议的注册中心有点不一样）

其实内部也是通过注册器registry.subscribe(url,this) 这里this就是registryDirectory它实现了NotifyListener。

服务提供者在向zookeeper注册服务/dubbo/com.alibaba.dubbo.demo.DemoService/providers/节点下写下自己的URL地址

服务消费者向zookeepr注册服务/dubbo/com.alibaba.dubbo.demo.DemoService/consumers/节点下写下自己的URL地址

服务消费者向zookeeper 订阅服务/dubbo/com.alibaba.dubbo.demo.DemoService/ providers /节点下所有服务提供者URL地址

Zookeeper通过watcher机制实现对节点的监听，节点数据变化通过节点上的watcher回调客户端， 重新生成对服务的refer

在订阅的过程中通过获取/dubbo/com.alibaba.dubbo.demo.DemoService/providers /下的所有服务提供者的urls(类似dubbo://10.33. 37.8:20880/com.alibaba.dubbo.demo.DemoService?anyhost=true&application=demo-consumer&check=false&dubbo=2.0.0&generic=false&interface=com.alibaba.dubbo.demo.DemoService&methods=sayHello&owner=william&pid=7356&side=consumer&timestamp=1416971340626)，主动回调NotifyListener来根据urls生成对服务提供者的引用生成可执行invokers,供目录服务持有

  1. 通过cluster.join(directory) 合并invoker并提供集群调用策略

DubboProtocol.refer过程:

  1. 经过ProtocolListenerWrapper， ProtocolFilterWrapper构建监听器链和过滤器链。

  2. DubboProtocol根据url获取ExchangeClient对象，如果是share存在就返回不存在创建新对象不是share直接创建。ExchangeClient是底层通信的客户端，对于通信层的创建功能不在这里讲解。

  3. 创建DubboInvoker, 这个invoker对象包含对远程服务提供者的长链接，是真正执行远程服务调用的可执行对象

  4. 将创建的invoker返回给目录服务

官方文档的应用服务的序列图

Center

发布服务活动图：

Center 1

引用： https://blog.csdn.net/quhongwei\_zhanqiu/article/details/41651487

发表评论

表情：
评论列表 （有 0 条评论，314人围观）

还没有评论，来说两句吧...

相关阅读