Dubbo服务消费过程

阳光穿透心脏的1/2处 2023-06-18 09:55 28阅读 0赞

在我们使用dubbo进行远程服务消费时，可以通过@Reference注解或<dubbo:reference>来配置要引用的接口，最终转化成ReferenceBean，这样我们便可以像调用本地接口方法一样使用远程服务提供的功能。

在功能实现的背后，dubbo为我们完成了很多复杂的过程：

1、生成远程服务的代理。

2、获得目标服务的url地址。

3、实现远程网络通信。

4、实现负载均衡。

5、实现集群容错。

### ReferenceBean ###

ReferenceBean继承自ReferenceConfig，同时实现了FactoryBean, ApplicationContextAware, InitializingBean, DisposableBean接口。其中最主要的实现方法是getObject与afterPropertiesSet。

Dubbo 服务引用的时机有两个，第一个是在 Spring 容器调用 ReferenceBean 的 afterPropertiesSet 方法时引用服务，第二个是在 ReferenceBean 对应的服务被注入到其他类中时引用，而入口都是getObject方法。

public class ReferenceBean<T> extends ReferenceConfig<T> implements FactoryBean, ApplicationContextAware, InitializingBean, DisposableBean {
        .....
    }

### **getObject** ###

@Override
        public Object getObject() {
            // 调用父类ReferenceConfig的get方法
            return get();
        }
    
        public synchronized T get() {
            // 检查和修改配置
            checkAndUpdateSubConfigs();
    
            if (destroyed) {
                throw new IllegalStateException("The invoker of ReferenceConfig(" + url + ") has already destroyed!");
            }
            // 检测 ref 是否为空，为空则通过 init 方法创建
            if (ref == null) {
                // init 方法主要用于处理配置，以及调用 createProxy 生成代理类
                init();
            }
            return ref;
        }

### **init** ###

初始化过程与服务发布类似，包括配置解析、校验以及参数组装等，这里只关注创建代理类方法createProxy。

private void init() {
            ...
            ref = createProxy(map);
            ...
        }

### createProxy ###

private T createProxy(Map<String, String> map) {
            // 判断是否在同一个JVM进程中调用
            if (shouldJvmRefer(map)) {
                URL url = new URL(LOCAL_PROTOCOL, LOCALHOST_VALUE, 0, interfaceClass.getName()).addParameters(map);
                invoker = REF_PROTOCOL.refer(interfaceClass, url);
                if (logger.isInfoEnabled()) {
                    logger.info("Using injvm service " + interfaceClass.getName());
                }
            } else {
                urls.clear(); // reference retry init will add url to urls, lead to OOM
                // 如果url不为空，说明是点对点调用
                if (url != null && url.length() > 0) { // user specified URL, could be peer-to-peer address, or register center's address.
                    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);
                            }
                            // 检测url协议是否为registry，若是，表明用户想使用指定注册中心
                            if (REGISTRY_PROTOCOL.equals(url.getProtocol())) {
                                // 将 map 转换为查询字符串，并作为 refer 参数的值添加到 url 中
                                urls.add(url.addParameterAndEncoded(REFER_KEY, StringUtils.toQueryString(map)));
                            } else {
                                // 合并 url，移除服务提供者的一些配置（这些配置来源于用户配置的 url 属性），
                                // 比如线程池相关配置。并保留服务提供者的部分配置，比如版本，group，时间戳等
                                // 最后将合并后的配置设置为 url 查询字符串中。
                                urls.add(ClusterUtils.mergeUrl(url, map));
                            }
                        }
                    }
                } else { // assemble URL from register center's configuration
                    // if protocols not injvm checkRegistry
                    if (!LOCAL_PROTOCOL.equalsIgnoreCase(getProtocol())){
                        // 校验注册中心的配置以及是否有必要从配置中心组装url
                        checkRegistry();
                        List<URL> us = loadRegistries(false);
                        if (CollectionUtils.isNotEmpty(us)) {
                            for (URL u : us) {
                                URL monitorUrl = loadMonitor(u);
                                if (monitorUrl != null) {
                                    map.put(MONITOR_KEY, URL.encode(monitorUrl.toFullString()));
                                }
                                urls.add(u.addParameterAndEncoded(REFER_KEY, StringUtils.toQueryString(map)));
                            }
                        }
                        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.");
                        }
                    }
                }
    
                // 如果只配置了一个注册中心或一个provider
                if (urls.size() == 1) {
                    // 创建invoker
                    invoker = REF_PROTOCOL.refer(interfaceClass, urls.get(0));
                } else {
                    List<Invoker<?>> invokers = new ArrayList<Invoker<?>>();
                    URL registryURL = null;
                    // 遍历urls生成多个invoker
                    for (URL url : urls) {
                        invokers.add(REF_PROTOCOL.refer(interfaceClass, url));
                        if (REGISTRY_PROTOCOL.equals(url.getProtocol())) {
                            registryURL = url; // use last registry url
                        }
                    }
                    if (registryURL != null) { // registry url is available
                        // use RegistryAwareCluster only when register's CLUSTER is available
                        URL u = registryURL.addParameter(CLUSTER_KEY, RegistryAwareCluster.NAME);
                        // The invoker wrap relation would be: RegistryAwareClusterInvoker(StaticDirectory) -> FailoverClusterInvoker(RegistryDirectory, will execute route) -> Invoker
                        // 创建 StaticDirectory 实例，并由 Cluster 对多个 Invoker 进行合并
                        invoker = CLUSTER.join(new StaticDirectory(u, invokers));
                    } else { // not a registry url, must be direct invoke.
                        invoker = CLUSTER.join(new StaticDirectory(invokers));
                    }
                }
            }
    
            if (shouldCheck() && !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());
            }
            /**
             * @since 2.7.0
             * ServiceData Store
             */
            MetadataReportService metadataReportService = null;
            if ((metadataReportService = getMetadataReportService()) != null) {
                URL consumerURL = new URL(CONSUMER_PROTOCOL, map.remove(REGISTER_IP_KEY), 0, map.get(INTERFACE_KEY), map);
                metadataReportService.publishConsumer(consumerURL);
            }
            // create service proxy
            // 生成代理类
            return (T) PROXY_FACTORY.getProxy(invoker);
        }

createProxy方法的主要逻辑：

1、首先根据配置检查是否为本地调用，若是，则调用 InjvmProtocol 的 refer 方法生成 InjvmInvoker 实例。

2、判断是否是点对点调用，如果是则把url保存到urls集合，如果urls数量为1，执行步骤4，如果大于1，则执行步骤5。

3、如果配置了注册中心，则遍历注册中心，把url添加到urls集合，如果urls数量为1，执行步骤4，如果大于1，则执行步骤5。

4、构建直连的invoker。

5、构建invokers集合，并通过cluster合并invokers。

6、最后调用ProxyFactory生成代理类。

### REF\_PROTOCOL.refer ###

这里通过指定的协议来调用refer生成一个invoker对象，invoker是一个代理对象。那么在当前的消费端而言，invoker主要用于执行远程调用。

private static final Protocol REF_PROTOCOL = ExtensionLoader.getExtensionLoader(Protocol.class).getAdaptiveExtension();

REF\_PROTOCOL是一个自适应扩展点，当有注册中心时，解析url得到的协议是registry，所以最终调用的是RegistryProtocol的refer方法。

### RegistryProtocol.refer ###

主要逻辑：

1、组装注册中心协议的url  
2、判断是否配置legroup，如果有，则cluster=getMergeableCluster()，构建invoker  
3、doRefer构建invoker

@Override
        @SuppressWarnings("unchecked")
        public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
            // 根据配置的协议，生成注册中心的url: zookeeper://
            url = URLBuilder.from(url)
                    .setProtocol(url.getParameter(REGISTRY_KEY, DEFAULT_REGISTRY))
                    .removeParameter(REGISTRY_KEY)
                    .build();
            Registry registry = registryFactory.getRegistry(url);
            if (RegistryService.class.equals(type)) {
                return proxyFactory.getInvoker((T) registry, type, url);
            }
    
            // group="a,b" or group="*"
            // 解析group参数，根据group决定cluster的类型
            Map<String, String> qs = StringUtils.parseQueryString(url.getParameterAndDecoded(REFER_KEY));
            String group = qs.get(GROUP_KEY);
            if (group != null && group.length() > 0) {
                if ((COMMA_SPLIT_PATTERN.split(group)).length > 1 || "*".equals(group)) {
                    return doRefer(getMergeableCluster(), registry, type, url);
                }
            }
            return doRefer(cluster, registry, type, url);
        }

### doRefer ###

主要逻辑：

1、构建一个RegistryDirectory。  
2、构建一个consumer://协议的地址注册到注册中心。  
3、订阅zookeeper中节点的变化。  
4、调用cluster.join方法返回invoker。

private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) {
            // RegistryDirectory初始化
            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());
            // 注册consumer://协议的url
            URL subscribeUrl = new URL(CONSUMER_PROTOCOL, parameters.remove(REGISTER_IP_KEY), 0, type.getName(), parameters);
            if (!ANY_VALUE.equals(url.getServiceInterface()) && url.getParameter(REGISTER_KEY, true)) {
                directory.setRegisteredConsumerUrl(getRegisteredConsumerUrl(subscribeUrl, url));
                registry.register(directory.getRegisteredConsumerUrl());
            }
            directory.buildRouterChain(subscribeUrl);
            // 订阅事件监听
            directory.subscribe(subscribeUrl.addParameter(CATEGORY_KEY,
                    PROVIDERS_CATEGORY + "," + CONFIGURATORS_CATEGORY + "," + ROUTERS_CATEGORY));
            // 构建invoker
            Invoker invoker = cluster.join(directory);
            ProviderConsumerRegTable.registerConsumer(invoker, url, subscribeUrl, directory);
            return invoker;
        }

### Cluster ###

private Cluster cluster;
    
        public void setCluster(Cluster cluster) {
            this.cluster = cluster;
        }
    
        // 构建invoker
        Invoker invoker = cluster.join(directory);

cluster其实是在RegistryProtocol中通过set方法完成依赖注入的，并且，它还是一个被包装的。

Cluster扩展点的定义, 由于它是一个自适应扩展点，那么会动态生成一个Cluster$Adaptive的动态代理类。

### Cluster$Adaptive ###

public class Cluster$Adaptive implements org.apache.dubbo.rpc.cluster.Cluster {
        public org.apache.dubbo.rpc.Invoker
        join(org.apache.dubbo.rpc.cluster.Directory arg0) throws
                org.apache.dubbo.rpc.RpcException {
            if (arg0 == null) throw new
                    IllegalArgumentException("org.apache.dubbo.rpc.cluster.Directory argument == null");
            if (arg0.getUrl() == null) throw new
                    IllegalArgumentException("org.apache.dubbo.rpc.cluster.Directory argument getUrl() == null");
            org.apache.dubbo.common.URL url = arg0.getUrl();
            String extName = url.getParameter("cluster", "failover");
            if (extName == null)
                throw new IllegalStateException("Failed to get extension (org.apache.dubbo.rpc.cluster.Cluster) name from url (" +
                        url.toString() + ") use keys([cluster])");
            org.apache.dubbo.rpc.cluster.Cluster extension =
                    (org.apache.dubbo.rpc.cluster.Cluster) ExtensionLoader.getExtensionLoader(org.apache.dubbo.rpc.cluster.Cluster.class).getExtension(extName);
            return extension.join(arg0);
        }
    }

在动态适配的类中会基于extName，选择一个合适的扩展点进行适配，由于默认情况下cluster默认failover，所以  
getExtension("failover")理论上应该返回FailOverCluster。但实际上，这里做了包装：MockClusterWrapper(FailOverCluster)

META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.Cluster配置文件：

mock=org.apache.dubbo.rpc.cluster.support.wrapper.MockClusterWrapper
    failover=org.apache.dubbo.rpc.cluster.support.FailoverCluster
    failfast=org.apache.dubbo.rpc.cluster.support.FailfastCluster
    failsafe=org.apache.dubbo.rpc.cluster.support.FailsafeCluster
    failback=org.apache.dubbo.rpc.cluster.support.FailbackCluster
    forking=org.apache.dubbo.rpc.cluster.support.ForkingCluster
    available=org.apache.dubbo.rpc.cluster.support.AvailableCluster
    mergeable=org.apache.dubbo.rpc.cluster.support.MergeableCluster
    broadcast=org.apache.dubbo.rpc.cluster.support.BroadcastCluster
    registryaware=org.apache.dubbo.rpc.cluster.support.RegistryAwareCluster

### cluster.join ###

所以再回到doRefer方法，cluster.join(directory)方法实际调用的是MockClusterWrapper(FailOverCluster).join，返回的是MockClusterInvoker对象。

Invoker invoker = cluster.join(directory);

public class MockClusterWrapper implements Cluster {
    
        private Cluster cluster;
    
        public MockClusterWrapper(Cluster cluster) {
            this.cluster = cluster;
        }
    
        @Override
        public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
            return new MockClusterInvoker<T>(directory,
                    this.cluster.join(directory));
        }
    
    }

### proxyFactory.getProxy ###

回到ReferenceConfig.createProxy方法的最后一行，根据拿到的invoker生成动态代理：

// 生成代理类
        return (T) PROXY_FACTORY.getProxy(invoker);

PROXY\_FACTORY是一个自适应扩展点，因此找到ProxyFactory扩展点的默认实现JavassistProxyFactory的getProxy方法。

private static final ProxyFactory PROXY_FACTORY = ExtensionLoader.getExtensionLoader(ProxyFactory.class).getAdaptiveExtension();

### JavassistProxyFactory.getProxy ###

getProxy方法为我们的目标接口生成了一个动态代理类，并且对invoker再做了一层处理。 意味着后续发起服务调用的时候，会由InvokerInvocationHandler来进行处理。

通过@Reference注入的对象本质就是这个动态代理，执行具体方法会通过InvokerInvocationHandler的invoke方法调用

@Override
        @SuppressWarnings("unchecked")
        public <T> T getProxy(Invoker<T> invoker, Class<?>[] interfaces) {
            return (T) Proxy.getProxy(interfaces).newInstance(new InvokerInvocationHandler(invoker));
        }