Davids原理探究：Dubbo服务消费原理

快来打我* 2023-02-20 07:40 49阅读 0赞

### 文章目录 ###

*   *  Dubbo服务消费原理
     *   *  服务消费核心类ReferenceConfig
         *  从注册中心消费
         *  创建具体远程Invoker
         *  多注册中心消费原理
         *  直连服务消费原理

## Dubbo服务消费原理 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0Rhdmlkc18_size_16_color_FFFFFF_t_70_pic_center]

> **关注可以查看更多粉丝专享blog~**

前面已经讲过[Dubbo服务暴露的原理][Dubbo]，今天分析一下Dubbo服务消费原理，相比于服务暴露原理，服务消费就有点像把暴露原理倒过来，可以看一下消费原理图。跟服务暴露的流程很相似，只是服务暴露的核心类是ServiceConfig，服务消费的核心类是ReferenceConfig，从名字上也可以看出来。

服务暴露流程：`ServiceConfig --> ProxyFactory --> Invoker --> Protocol --> Exporter`  
服务消费流程：`ReferenceConfig --> Protocol --> Invoker --> ProxyFactory --> ref`

一个是把本地的推倒远端，一个是把远端的代理到本地，像本地调用一样的进行远程调用，具体是否调用本地或者远端，对于我们来说是透明的，Dubbo已经帮我们封装好了。

![Dubbo服务消费机制][Dubbo 1]

整体上看，Dubbo框架实现服务消费也分为两大部分：

1.  通过持有远程对象实例生成Invoker，这个Invoker在客户端是核心的远程代理对象。
2.  把Invoker通过动态代理转换成实现用户接口的动态代理引用。

这里Invoker承载了网络连接、服务调用和重试等功能，在客户端，它可能是一个远程的实现，也可能是一个集群的实现。  
框架真正进行服务引用的入口点在ReferenceBean\#getObject，不管是XML还是注解，都会转换成ReferenceBean，它继承自ReferenceConfig。

### 服务消费核心类ReferenceConfig ###

1.  优先判断是否处于同一个JVM里面，默认场景下Dubbo会找出内存中的injvm协议的服务（服务暴露时会注册一份到injvm，将服务实例放到内存map中）直接获取实例调用。
2.  在注册中心追加消费者元数据信息，应用启动时订阅注册中心、服务提供者参数等合并时会用到这部分信息。
3.  处理只有一个注册中心的场景，这种场景在客户端中是最常见的，客户端启动拉取服务元数据，订阅provider、路由和配置变更。
4.  处理多注册中心的场景。逐个获取注册中心的服务，并添加到invokers列表中，后面通过Cluster将多个Invoker转换成一个Invoker。

// ReferenceConfig#createProxy
    private T createProxy(Map<String, String> map) {
        
    	// 检查是否是同一个JVM的内部引用
        if (shouldJvmRefer(map)) {
        
        	// 构造injvm url
            URL url = new URL(LOCAL_PROTOCOL, LOCALHOST_VALUE, 0, interfaceClass.getName()).addParameters(map);
            // 直接使用injvm协议从内存中获取实例
            invoker = REF_PROTOCOL.refer(interfaceClass, url);
            if (logger.isInfoEnabled()) {
        
                logger.info("Using injvm service " + interfaceClass.getName());
            }
        } else {
        
            urls.clear();
            // 用户指定的URL，可以是点对点地址，或注册中心地址
            if (url != null && url.length() > 0) {
        
                String[] us = SEMICOLON_SPLIT_PATTERN.split(url);
                if (us != null && us.length > 0) {
        
                    for (String u : us) {
        
                        URL url = URL.valueOf(u);
                        if (StringUtils.isEmpty(url.getPath())) {
        
                            url = url.setPath(interfaceName);
                        }
                        if (UrlUtils.isRegistry(url)) {
        
                        	// 注册中心地址后面添加refer存储服务消费元数据信息
                            urls.add(url.addParameterAndEncoded(REFER_KEY, StringUtils.toQueryString(map)));
                        } else {
        
                        	// 直连某台服务提供者
                            urls.add(ClusterUtils.mergeUrl(url, map));
                        }
                    }
                }
            } else {
        
                // 如果协议是不injvm，则从注册中心的配置组装URL
                if (!LOCAL_PROTOCOL.equalsIgnoreCase(getProtocol())) {
        
                    checkRegistry();
                    List<URL> us = ConfigValidationUtils.loadRegistries(this, false);
                    if (CollectionUtils.isNotEmpty(us)) {
        
                        for (URL u : us) {
        
                            URL monitorUrl = ConfigValidationUtils.loadMonitor(this, u);
                            if (monitorUrl != null) {
        
                                map.put(MONITOR_KEY, URL.encode(monitorUrl.toFullString()));
                            }
                            // 注册中心地址后面添加refer存储服务消费元数据信息
                            urls.add(u.addParameterAndEncoded(REFER_KEY, StringUtils.toQueryString(map)));
                        }
                    }
                    // 如果urls为空则报错，既不是injvm，注册中也未获取到对应的服务
                    if (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=\"...\" /> to your spring config.");
                    }
                }
            }
            // 单注册中心消费
            if (urls.size() == 1) {
        
            	// 使用注册中心对应协议从内存中获取实例
                invoker = REF_PROTOCOL.refer(interfaceClass, urls.get(0));
            } else {
        
            	// 多注册中心消费
                List<Invoker<?>> invokers = new ArrayList<Invoker<?>>();
                URL registryURL = null;
                for (URL url : urls) {
        
                	// 遍历注册中心列表，诸葛获取注册中心服务，并添加到invokers列表
                    invokers.add(REF_PROTOCOL.refer(interfaceClass, url));
                    if (UrlUtils.isRegistry(url)) {
        
                        // 使用最后的注册地址
                        registryURL = url;
                    }
                }
                // registryURL != null，标识注册表url可用
                if (registryURL != null) {
        
                    // 多注册中心场景, 使用'zone-aware'策略
                    URL u = registryURL.addParameterIfAbsent(CLUSTER_KEY, ZoneAwareCluster.NAME);
                    // 使用Cluster将多个Invoker转换成一个Invoker
                    invoker = CLUSTER.join(new StaticDirectory(u, invokers));
                } else {
        
                	// 非注册中心url则直接调用
                    invoker = CLUSTER.join(new StaticDirectory(invokers));
                }
            }
        }
    	// 如果开启了状态检查，则检查Invoker是否可用
        if (shouldCheck() && !invoker.isAvailable()) {
        
            invoker.destroy();
            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());
        }
        // 把Invoker转换成代理接口
        return (T) PROXY_FACTORY.getProxy(invoker, ProtocolUtils.isGeneric(generic));
    }

### 从注册中心消费 ###

当经过注册中心线消费时，主要通过RegistryProtocol\#refer触发数据拉取、订阅和服务Invoker转换等操作，其中最核心的数据结构是RegistryDirectory。这段逻辑主要完成了注册中心实例的创建，元数据注册到注册中心及订阅的功能。

1.  根据用户指定的注册中心进行协议替换，具体注册中心协议会在启动时用registry储存对应值。
2.  创建注册中心实例，这里的URL其实是注册中心地址，真实消费方的元数据信息是放在refer属性中储存的。
3.  提取消费方refer中的元数据信息，如果包含多个分组值则会把调用结果进行合并。
4.  触发真正的服务订阅和Invoker转换。
5.  RegistryDirectory实现了NotifyListerer接口，服务变更会触发这个类回调notify方法，用于重新引用服务。
6.  把消费者元数据信息注册到注册中心（如：消费方应用名、IP和端口号等）。
7.  处理provider、路由和动态配置的订阅。
8.  通过Cluster合并invokers

// RegistryProtocol#refer
    @Override
    @SuppressWarnings("unchecked")
    public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
        
    	// 设置具体的注册中心协议
        url = URLBuilder.from(url)
                    .setProtocol(url.getParameter(REGISTRY_KEY, DEFAULT_REGISTRY))
                    .removeParameter(REGISTRY_KEY)
                    .build();
        // 创建具体注册中心实例
        Registry registry = registryFactory.getRegistry(url);
        // 如果引用服务的接口类是RegistryService则直接使用动态代理创建Invoker代理对象
        if (RegistryService.class.equals(type)) {
        
            return proxyFactory.getInvoker((T) registry, type, url);
        }
    
        // group="a,b" or group="*"
        // 根据配置处理多分组结果聚合
        Map<String, String> qs = StringUtils.parseQueryString(url.getParameterAndDecoded(REFER_KEY));
        String group = qs.get(GROUP_KEY);
        if (group != null && group.length() > 0) {
        
        	// 当一个接口有多种实现时，使用MergeableCluster对结果集进行合并处理
            if ((COMMA_SPLIT_PATTERN.split(group)).length > 1 || "*".equals(group)) {
        
                return doRefer(getMergeableCluster(), registry, type, url);
            }
        }
        // 处理订阅数据并通过Cluster合并多个Invoker
        return doRefer(cluster, registry, type, url);
    }
    
    private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) {
        
    	// 消费核心关键，持有实际Invoker和接收订阅通知
    	// RegistryDirectory实现了NotifyListerer接口，服务变更会触发这个类回调notify方法，用于重新引用服务
        RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url);
        directory.setRegistry(registry);
        directory.setProtocol(protocol);
        // REFER_KEY的所有属性
        Map<String, String> parameters = new HashMap<String, String>(directory.getConsumerUrl().getParameters());
        URL subscribeUrl = new URL(CONSUMER_PROTOCOL, parameters.remove(REGISTER_IP_KEY), 0, type.getName(), parameters);
        // isShouldRegister = !ANY_VALUE.equals(url.getServiceInterface()) && url.getParameter(REGISTER_KEY, true);
        // 如果directory需要注册，则将消费者注册到注册中心
        if (directory.isShouldRegister()) {
        
            directory.setRegisteredConsumerUrl(subscribeUrl);
            registry.register(directory.getRegisteredConsumerUrl());
        }
        // 构造路由规则链
        directory.buildRouterChain(subscribeUrl);
        // 向注册中心发起订阅（服务提供者、路由和动态配置）
        directory.subscribe(toSubscribeUrl(subscribeUrl));
    	// 通过Cluster合并Invokers
        Invoker<T> invoker = cluster.join(directory);
        // 如果没有相关订阅者则直接返回
        List<RegistryProtocolListener> listeners = findRegistryProtocolListeners(url);
        if (CollectionUtils.isEmpty(listeners)) {
        
            return invoker;
        }
    	// 如果有订阅者则通知订阅者，该consumer注册事件
        RegistryInvokerWrapper<T> registryInvokerWrapper = new RegistryInvokerWrapper<>(directory, cluster, invoker, subscribeUrl);
        for (RegistryProtocolListener listener : listeners) {
        
            listener.onRefer(this, registryInvokerWrapper);
        }
        return registryInvokerWrapper;
    }

### 创建具体远程Invoker ###

当第一次发起订阅时会进行一次数据拉取操作，同时触发RegistryDirectory\#notify方法，这里的通知数据是某一个类别的全量数据，比如provider和router类别数据。当通知provider数据时，在RegistryDirectory\#toInvokers方法内完成Invoker转换。

// RegistryDirectory#toInvokers
    private Map<String, Invoker<T>> toInvokers(List<URL> urls) {
        
        Map<String, Invoker<T>> newUrlInvokerMap = new HashMap<>();
        if (urls == null || urls.isEmpty()) {
        
            return newUrlInvokerMap;
        }
        Set<String> keys = new HashSet<>();
        String queryProtocols = this.queryMap.get(PROTOCOL_KEY);
        for (URL providerUrl : urls) {
        
            // 如果协议在reference端配置，则只选择匹配的协议
            if (queryProtocols != null && queryProtocols.length() > 0) {
        
                boolean accept = false;
                String[] acceptProtocols = queryProtocols.split(",");
                // 根据消费方protocol配置过滤不匹配协议
                for (String acceptProtocol : acceptProtocols) {
        
                    if (providerUrl.getProtocol().equals(acceptProtocol)) {
        
                        accept = true;
                        break;
                    }
                }
                if (!accept) {
        
                    continue;
                }
            }
            // empty协议直接跳过
            if (EMPTY_PROTOCOL.equals(providerUrl.getProtocol())) {
        
                continue;
            }
            // 没有对应的协议扩展打印日志并跳过
            if (!ExtensionLoader.getExtensionLoader(Protocol.class).hasExtension(providerUrl.getProtocol())) {
        
                logger.error(new IllegalStateException("Unsupported protocol " + providerUrl.getProtocol() +
                        " in notified url: " + providerUrl + " from registry " + getUrl().getAddress() +
                        " to consumer " + NetUtils.getLocalHost() + ", supported protocol: " +
                        ExtensionLoader.getExtensionLoader(Protocol.class).getSupportedExtensions()));
                continue;
            }
            // 合并provider端配置数据（如：服务端IP和port等）
            URL url = mergeUrl(providerUrl);
    
            String key = url.toFullString();
            // 忽略重复推送列表
            if (keys.contains(key)) {
        
                continue;
            }
            keys.add(key);
            // 缓存键是不与用户端参数合并的url，无论用户如何合并参数，如果服务器url更改，则再次引用
            Map<String, Invoker<T>> localUrlInvokerMap = this.urlInvokerMap;
            Invoker<T> invoker = localUrlInvokerMap == null ? null : localUrlInvokerMap.get(key);
            if (invoker == null) {
        
                try {
        
                    boolean enabled = true;
                    if (url.hasParameter(DISABLED_KEY)) {
        
                        enabled = !url.getParameter(DISABLED_KEY, false);
                    } else {
        
                        enabled = url.getParameter(ENABLED_KEY, true);
                    }
                    if (enabled) {
        
                    	// 使用具体协议创建远程调用
                        invoker = new InvokerDelegate<>(protocol.refer(serviceType, url), url, providerUrl);
                    }
                } catch (Throwable t) {
        
                    logger.error("Failed to refer invoker for interface:" + serviceType + ",url:(" + url + ")" + t.getMessage(), t);
                }
                if (invoker != null) {
        
                	// 缓存invoker
                    newUrlInvokerMap.put(key, invoker);
                }
            } else {
        
            	// 缓存invoker
                newUrlInvokerMap.put(key, invoker);
            }
        }
        keys.clear();
        return newUrlInvokerMap;
    }

### 多注册中心消费原理 ###

在实际使用中，我们更多遇到的是但注册中心场景，但是当跨机房消费时，Dubbo框架允许同时消费多个机房的服务。默认Dubbo消费机房服务的顺序按照配置注册中心的顺序决定，配置靠前优先消费。（多注册中心场景下，默认使用的集群策略是available）  
处理逻辑在上方`ReferenceConfig#createProxy`第57至78行。

### 直连服务消费原理 ###

Dubbo可以绕过注册中心直接向指定一台或多台服务（直接指定目标IP和端口）发起RPC调用，使用直连模式可以方便在某些场景下使用，比如压测指定机器等。  
处理逻辑在上方`ReferenceConfig#createProxy`第13至31行。

相关文章：  
[Davids原理探究：Dubbo源码编译（2.7.8）][Davids_Dubbo_2.7.8]  
[Davids原理探究：Dubbo SPI和Java SPI实现原理][Davids_Dubbo SPI_Java SPI]  
[Davids原理探究：Dubbo注册中心（ZooKeeper、Redis）实现原理][Davids_Dubbo_ZooKeeper_Redis]  
[Davids原理探究：Dubbo配置解析原理][Davids_Dubbo]  
[Davids原理探究：Dubbo服务暴露原理][Dubbo]  
[Davids原理探究：Dubbo服务消费原理][Davids_Dubbo 1]  
[Davids原理探究：Dubbo优雅停机原理][Davids_Dubbo 2]  
[Davids原理探究：Dubbo调用流程图][Davids_Dubbo 3]  
[Davids原理探究：Dubbo路由实现原理][Davids_Dubbo 4]  
[Davids原理探究：Dubbo负载均衡实现原理][Davids_Dubbo 5]  
[Davids原理探究：Dubbo过滤器原理][Davids_Dubbo 6]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0Rhdmlkc18_size_16_color_FFFFFF_t_70_pic_center]: https://img-blog.csdnimg.cn/20200817133417155.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0Rhdmlkc18=,size_16,color_FFFFFF,t_70#pic_center
[Dubbo]: https://blog.csdn.net/Davids_/article/details/106669184
[Dubbo 1]: https://img-blog.csdnimg.cn/20200625094929582.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0Rhdmlkc18=,size_16,color_FFFFFF,t_70
[Davids_Dubbo_2.7.8]: https://blog.csdn.net/Davids_/article/details/106603254
[Davids_Dubbo SPI_Java SPI]: https://blog.csdn.net/Davids_/article/details/106611949
[Davids_Dubbo_ZooKeeper_Redis]: https://blog.csdn.net/Davids_/article/details/106590954
[Davids_Dubbo]: https://blog.csdn.net/Davids_/article/details/106951976
[Davids_Dubbo 1]: https://blog.csdn.net/Davids_/article/details/106951941
[Davids_Dubbo 2]: https://blog.csdn.net/Davids_/article/details/106952238
[Davids_Dubbo 3]: https://blog.csdn.net/Davids_/article/details/106721401
[Davids_Dubbo 4]: https://blog.csdn.net/Davids_/article/details/106709918
[Davids_Dubbo 5]: https://blog.csdn.net/Davids_/article/details/106720555
[Davids_Dubbo 6]: https://blog.csdn.net/Davids_/article/details/106750014