Dubbo源码解读-Consumer消费端@Reference服务端引用流程

上篇我们介绍了Provider端监听注册中心动态配置原理，地址如下

Dubbo源码解析-Provider端监听注册中心动态配置原理-CSDN博客

        本文主要针Dubbo消费端@Reference服务端引用流程原理，从dubbo源码角度进行解析。

        大家可以好好仔细读一下本文。有疑问欢迎留言。

        接着说明，读Dubbo源码最好是先对Spring源码有一定的了解。如果大家需要，我也可以针对Spring框架做一系列源码的解读专栏。

         不过不用担心，如果需要Spring的源码知识，文章中也会进行Spring源码铺垫介绍的。

        如果内容中有没描述清楚的，或者大家在阅读源代码有疑问的，欢迎留言，看到就会及时回复。

消费端@Reference依赖注入服务端流程

        Dubbo消费端引用依赖注入服务端，整个细节过程是比较复杂的。因为好多同学为了应付面试或者觉得细节流程太繁琐或者复杂，只想弄清楚相对简练的过程。所以对细节流程做了一个相对笼统的总结，方便大家理解大概过程，以及应对面试。

流程概述：

        其实就是注册节点到consumers节点下，然后订阅了providers/configurators/routers节点事件，创建Invoker代理。

    

1. ReferenceAnnotationBeanPostProcessor。负责@Reference属性的收集和依赖注入
2. postProcessPropertyValues()负责对收集好的@refer注解进行依赖注入。
3. AnnotatedFieldElement和.AnnotatedMethodElement。inject()负责方法和属性上的依赖注入。下面以字段依赖注入为例
4. 接下来创建依赖bean。会使用双重代理。
   • JDK代理bean。JDK->ReferenceBeanInvocationHandler->bean
   • javassis创建invoker代理，复制给bean。Javassis->InvokerInvocationHandler->MockClusterInvoker
5. 创建Invoke代理
   • 封装配置属性到map
   • 获取所有注册协议，进行遍历urls=loadRegistries(false);registry协议
   • 将map变为String绑定到注册协议的URL上,key为“refer”。：regiestry协议
   • 调用Protocol包装链.referr()
   • QosProtocol  启动QosServer
   • RegistyProtocol.referr()
     • 创建RegistryDirectory服务列表集合：该类非常重要。用来存储服务列表
     • 实现NotifyListener监听zk节点变化。
     • 注册节点到zookeeper中的consumers节:./dubbo/com.*.AsyncService/consumers
     • 注册节点监听事件。：providers，configurators，routers节点的变化代码和@Servcie流程一样。
     • 返回MockClusterInvoker对象。持有RegistryDirectory和FailbackClusterInvoker
   • 返回mockCluserInvoker代理。javassisProxy创建mockCluserInvoker代理
6. JDK创建代理bean代理
   •  创建InvokerHandler：ReferenceBeanInvocationHandler，持有Bean就是5中生成的invoke代理。

总结：

1. 依赖注入的对象创建了两层代理
2. 核心RegistryProtocol.doRefer()
   1. 构建服务列表缓存类RegistryDirectory。实现NotifyListener,监听节点事件。
   2. 注册consumers节点
   3. 注册监听节点事件。节点：providers，configurators，routers
   4. 返回MockClusterInvoker。持有RegistryDirectory和FailbackClusterInvoker.

详细流程：

1. ReferenceAnnotationBeanPostProcessor.postProcessPropertyValues()负责对收集好的@refer注解进行依赖注入
2. 遍历收集的注解集合
3. 属性注入.方法上注解和字段上注解
   1. AnnotationInjectedBeanPostProcessor.AnnotatedFieldElement.inject()
   2. AnnotationInjectedBeanPostProcessor.AnnotatedMethodElement.inject()
4. 获取依赖bean:getInjectedObject()获取需要依赖注入的实例。创建了两层代理
   1. 创建对象:doGetInjectedBean(annotation, bean, beanName, injectedType, injectedElement);
      1. 创建referencedbean的名称:ServiceBean:+接口名称
      2. 创建ReferenceBean:buildReferenceBeanIfAbsent()。
      3. 每个@Reference注解都对应一个ReferenceBean对象
         1. 创建对象
         2. 解析注解中的属性，配置到ReferenceBean对象
      4. 创建需要依赖注入的代理实例:buildProxy(referencedBeanName, referenceBean, injectedType)JDK方式创建代理。
         1. 创建InvokerHandler：ReferenceBeanInvocationHandler,持有ReferenceBean和bean对象，通过handler.init初始化bean,构建mockCluserInvoker代理对象，复给bean.
            1. 生成bean对象：handler.init()->referenceBean.get()->referenceConfig.init();初始化远程接口
               1.  以下几步和@Servcie过程一样
               2. checkDefault():从系统属性和dubbo.properties中拿属性设置到ConsumerConfig中
               3. appendProperties(this):系统属性和dubbo.properties中拿属性设置到ReferenceConfig中
               4. 初始化各种配置：application，module,registries,monitor等
               5. 校验各种配置
               6. 生成map,把配置属性专为map.
               7. 创建ref代理： createProxy(Map<String, String> map)
                  1. 判断是否本地injvm（以下远程为主来介绍）
                  2. 获取所有注册协议，进行遍历urls=loadRegistries(false)。registry协议。
                     1. registry://localhost:2181/com.alibaba.dubbo.registry.RegistryService?application=dubbo_consumer&dubbo=2.0.2&owner=world&pid=52033&registry=zookeeper&timestamp=1710587235093
                  3. 将map变为String绑定到注册协议的URL上,key为“refer”，并复制给全局urls集合。registry协议。
                     1. urls.add(u.addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map)));
                     2. .StringUtils.toQueryString(map
                        1. .application=dubbo_consumer&check=false&default.check=false&dubbo=2.0.2&interface=com.xiangxue.jack.service.UserService&methods=doKill,queryUser&owner=world&pid=52033&register.ip=192.168.42.25&revision=0.0.1-SNAPSHOT&side=consumer&timestamp=1710587116529
                     3. registry://localhost:2181/com.alibaba.dubbo.registry.RegistryService?application=dubbo_consumer&dubbo=2.0.2&owner=world&pid=52033&refer=application%3Ddubbo_consumer%26check%3Dfalse%26default.check%3Dfalse%26dubbo%3D2.0.2%26interface%3Dcom.xiangxue.jack.service.UserService%26methods%3DdoKill%2CqueryUser%26owner%3Dworld%26pid%3D52033%26register.ip%3D192.168.42.25%26revision%3D0.0.1-SNAPSHOT%26side%3Dconsumer%26timestamp%3D1710587116529&registry=zookeeper&timestamp=1710587134398
                  4. 创建invoker():refprotocol.refer(interfaceClass, urls.get(0))开始处理regsitry协议Ur
                     1. .QosProtocolWrapper.refer()：启动QosServer
                     2. RegistryProtocol.refer(class,url):
                        1. 修改url协议为：zookeeper
                        2. doRefer()
                        3. RegistryDirectory服务列表集合：该类非常重要。用来存储服务列表directory = new RegistryDirectory<T>(type, url)
                           1. RegistryDirectory实现了NotifyListener。监听zookeeper的Notify事件，来刷新本地服务列表
                           2. methodInvokerMap：缓存本地服务列表，建立method和invokers的映射关系
                           3. 持有type,url(zookeeper协议)，zookeeperRegistr
                        4. 生成订阅url:subscribeUrl，consumer协议正好与生产端协议相反（provider://）
                           1. consumer://192.168.42.25/com.xiangxue.jack.service.UserService?application=dubbo_consumer&check=false&default.check=false&dubbo=2.0.2&interface=com.xiangxue.jack.service.UserService&methods=doKill,queryUser&owner=world&pid=52033&revision=0.0.1-SNAPSHOT&side=consumer&timestamp=1710587116529
                        5. 生成注册url: registeredConsumerUrl，多了个目录属性category
                           1. consumer://192.168.42.25/com.xiangxue.jack.service.UserService?application=dubbo_consumer&category=consumers&check=false&default.check=false&dubbo=2.0.2&interface=com.xiangxue.jack.service.UserService&methods=doKill,queryUser&owner=world&pid=52033&revision=0.0.1-SNAPSHOT&side=consumer&timestamp=1710587116529
                        6. 把消费端注册到zookeeper中的consumers节点
                           1. /dubbo/com.*.AsyncService/consumers
                        7. 将registeredConsumerUrl绑定到RegistryDirectory属性中
                        8. 监控 providers，configurators，routers节点的变化
                           1. directory.subscribe（url,this）。
                           2. 监控了3个节点，代码和@Servcie流程一样
                              1. 创建节点
                              2. 订阅节点事件
                              3. 发布empty协议消息。如果节点下无子节点，则处罚empty协议，有子节点则正常触发（服务端和消费端启动时都会触发），如服务端启动是，没有provider，则触发empty协议，有provider则将provider协议加载到服务列表
                           3. RegistryDirectory继承了notifylistenner。通知形式和ServiceBean export流程一样。
                        9. 返回MockClusterInvoker对象。持有RegistryDirectory和FailbackClusterInvoker。
                  5. 得到MockClusterInvoker：支持mock降级和failover集群容错
                  6. 返回mockCluserInvoker代理。javassisProxy创建mockCluserInvoker代理。
               8. 赋给ReferenceBeanInvocationHandler的bean属性
         2. 创建代理：Proxy.newProxyInstance
5. 反射注入bean

源码分析：

ReferenceAnnotionBeanPostProcessor.构建依赖注入对象 

protected Object doGetInjectedBean(Reference reference, Object bean, String beanName, Class<?> injectedType,
                                       InjectionMetadata.InjectedElement injectedElement) throws Exception {

        //创建referencedbean的名称
        String referencedBeanName = buildReferencedBeanName(reference, injectedType);

        //创建ReferenceBean
        ReferenceBean referenceBean = buildReferenceBeanIfAbsent(referencedBeanName, reference, injectedType, getClassLoader());
        //field和method的injectedElement，和ReferenceBean建立缓存
        cacheInjectedReferenceBean(referenceBean, injectedElement);
        //创建需要依赖注入的代理实例
        Object proxy = buildProxy(referencedBeanName, referenceBean, injectedType);

        return proxy;
    }

 ReferenceAnnotionBeanPostProcessor.JDK创建第1层动态代理

private Object buildProxy(String referencedBeanName, ReferenceBean referenceBean, Class<?> injectedType) {
        //创建代理的advice
        InvocationHandler handler = buildInvocationHandler(referencedBeanName, referenceBean);
        //对依赖注入的属性接口创建jdk的代理
        Object proxy = Proxy.newProxyInstance(getClassLoader(), new Class[]{injectedType}, handler);
        return proxy;
    }

  ReferenceAnnotionBeanPostProcessor. handler.init远程应用接口初始化，返回MockClusterInvoker。Javassis创建代理

private InvocationHandler buildInvocationHandler(String referencedBeanName, ReferenceBean referenceBean) {

        ReferenceBeanInvocationHandler handler = localReferenceBeanInvocationHandlerCache.get(referencedBeanName);

        if (handler == null) {
            //创建advice
            handler = new ReferenceBeanInvocationHandler(referenceBean);
        }

        //这里其实就是判断容器中是否有referencedBeanName的实例，如果是@Service注解的类就会在spring容器中
        if (applicationContext.containsBean(referencedBeanName)) { // Is local @Service Bean or not ?
            // ReferenceBeanInvocationHandler's initialization has to wait for current local @Service Bean has been exported.
            localReferenceBeanInvocationHandlerCache.put(referencedBeanName, handler);
        } else {
            // Remote Reference Bean should initialize immediately
            //如果是远程应用的接口，则在这里初始化
            handler.init();
        }

        return handler;
    }

RegistryProtocol.doRefer

private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) {
        RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url);
        directory.setRegistry(registry);
        directory.setProtocol(protocol);
        // all attributes of REFER_KEY
        Map<String, String> parameters = new HashMap<String, String>(directory.getUrl().getParameters());
        URL subscribeUrl = new URL(Constants.CONSUMER_PROTOCOL, parameters.remove(Constants.REGISTER_IP_KEY), 0, type.getName(), parameters);
        if (!Constants.ANY_VALUE.equals(url.getServiceInterface())
                && url.getParameter(Constants.REGISTER_KEY, true)) {
            URL registeredConsumerUrl = getRegisteredConsumerUrl(subscribeUrl, url);
            //把消费端注册到zookeeper中的consumers节点
            registry.register(registeredConsumerUrl);
            directory.setRegisteredConsumerUrl(registeredConsumerUrl);
        }
        //消费端 监控 providers，configurators，routers节点的变化
        directory.subscribe(subscribeUrl.addParameter(Constants.CATEGORY_KEY,
                Constants.PROVIDERS_CATEGORY
                        + "," + Constants.CONFIGURATORS_CATEGORY
                        + "," + Constants.ROUTERS_CATEGORY));

        //这里获取到MocklusterWrapper
        Invoker invoker = cluster.join(directory);
        //QOS统计
        ProviderConsumerRegTable.registerConsumer(invoker, url, subscribeUrl, directory);
        return invoker;
    }

总结：上面内容中，每个从业务流程和源码角度进行了详细分析，如果大家有疑问或者对文章排版任何方面有建议都可以留言评论，看到都会及时回复大家。

知识总结，分享不易，全文手敲，欢迎大家关注点赞评论收藏。