Dubbo原理和源码解析之服务引用

秒速五厘米 2022-12-27 14:09 137阅读 0赞

> # **github新增仓库 "dubbo-read"（[点此查看][Link 1]），集合所有《Dubbo原理和源码解析》系列文章，后续将继续补充该系列，同时将针对Dubbo所做的功能扩展也进行分享。不定期更新，欢迎Follow。** #

# 一、框架设计 #

在官方《[Dubbo 开发指南][Dubbo]》框架设计部分，给出了引用服务时序图：

![3c9c7e4202035d1c4b7749e057959d35.png][]

另外，在官方《[Dubbo 用户指南][Dubbo 1]》集群容错部分，给出了服务引用的各功能组件关系图：

![701f9cdc1a63d415e7756eec01c118ca.png][]

本文将根据以上两张图，分析服务引用的实现原理，并进行详细的代码跟踪与解析。

# 二、原理和源码解析 #

## 2.1 创建代理 ##

Dubbo 基于 Spring 的 Schema 扩展实现 XML 配置解析，DubboNamespaceHandler 会将 <dubbo:reference> 标签解析为 ReferenceBean，ReferenceBean 实现了 FactoryBean，因此当它在代码中有引用时，会调用 ReferenceBean\#getObject() 方法进入节点注册和服务发现流程。

**ReferenceBean.java**

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

**ReferenceConfig.java**

public synchronized T get() {
        if (destroyed){
            throw new IllegalStateException("Already destroyed!");
        }
        if (ref == null) {
            init();
        }
        return ref;
    }
    
    private void init() {
        //.......忽略
        ref = createProxy(map);
    }
    
    
    private T createProxy(Map<String, String> map) {
        //.....忽略
        invoker = refprotocol.refer(interfaceClass, urls.get(0));
        //.....忽略
        // 创建服务代理
        return (T) proxyFactory.getProxy(invoker);
    }

##  2.2 服务发现 ##

因为通过注册中心，因此在 ReferenceConfig.java\#createProxy() 方法中，进入 RegistryProtocol.java\#refer() 方法。

**RegistryProtocol.java**

private Cluster cluster;
    
    public void setCluster(Cluster cluster) {
        this.cluster = cluster;
    }
    
    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);
        URL subscribeUrl = new URL(Constants.CONSUMER_PROTOCOL, NetUtils.getLocalHost(), 0, type.getName(), directory.getUrl().getParameters());
        if (! Constants.ANY_VALUE.equals(url.getServiceInterface())
                && url.getParameter(Constants.REGISTER_KEY, true)) {
            registry.register(subscribeUrl.addParameters(Constants.CATEGORY_KEY, Constants.CONSUMERS_CATEGORY,
                    Constants.CHECK_KEY, String.valueOf(false)));
        }
        directory.subscribe(subscribeUrl.addParameter(Constants.CATEGORY_KEY, 
                Constants.PROVIDERS_CATEGORY 
                + "," + Constants.CONFIGURATORS_CATEGORY 
                + "," + Constants.ROUTERS_CATEGORY));
        return cluster.join(directory);
    }

RegistryDirectory 通过 RegistryDirectory\#subscribeUrl() 向 Zookeeper 订阅服务节点信息并 watch 变更，这样就实现了服务自动发现。

### 2.3 Invoker选取 ###

### 2.3.1 Cluster ###

上面我之所以把设置 Cluster 的代码贴上，是因为此处涉及到一个 Dubbo 服务框架核心的概念——微内核和插件机制（*此处会单独一篇文章详细介绍*）：

![85efd80c9deaf41f0f7086e9494b3806.png][]

有关 Dubbo 的设计原则，请查看Dubbo《[一些设计上的基本常识][Link 2]》。

Cluster 类的定义如下：

**Cluster.java**

@SPI(FailoverCluster.NAME)
    public interface Cluster {
    
        /**
         * Merge the directory invokers to a virtual invoker.
         */
        @Adaptive
        <T> Invoker<T> join(Directory<T> directory) throws RpcException;
    }

cluster 的类型是 Cluster$Adaptive，实际上是一个通用的代理类，它会根据 URL 中的 cluster 参数值定位到实际的 Cluster 实现类（默认是 FailoverCluster）。 由于 ExtensionLoader 在实例化对象时，会在实例化完成之后自动套上 Wrapper 类，而 MockerClusterWrapper 就是这样一个 Wrapper。

**MockerClusterWrapper.java**

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

也就是说，实例化出来的 FailoverCluster 会作为参数赋予 MockerClusterWrapper\#cluster，而 MockClusterWrapper 会作为参数赋予 RegistryProtocol\#cluster。因此 RegistryProtocol\#doRefer() 中调用 cluster.join(directory) 实际上是调用的 MockClusterWrapper\#join(directory)。 使用这种机制，可以把一些公共的处理放在 Wrapper 类中，实现代码和功能收敛。

**MockClusterInvoker.java**

public Result invoke(Invocation invocation) throws RpcException {
        Result result = null;
        
        String value = directory.getUrl().getMethodParameter(invocation.getMethodName(),
                                 Constants.MOCK_KEY, Boolean.FALSE.toString()).trim(); 
        if (value.length() == 0 || value.equalsIgnoreCase("false")){
            //no mock
            result = this.invoker.invoke(invocation);
        } else if (value.startsWith("force")) {
            if (logger.isWarnEnabled()) {
                logger.info("force-mock: " + invocation.getMethodName() + 
                            " force-mock enabled , url : " +  directory.getUrl());
            }
            //force:direct mock
            result = doMockInvoke(invocation, null);
        } else {
            //fail-mock
            try {
                result = this.invoker.invoke(invocation);
            }catch (RpcException e) {
                if (e.isBiz()) {
                    throw e;
                } else {
                    if (logger.isWarnEnabled()) {
                        logger.info("fail-mock: " + invocation.getMethodName() + 
                                " fail-mock enabled , url : " +  directory.getUrl(), e);
                    }
                    //fail:mock
                    result = doMockInvoke(invocation, e);
                }
            }
        }
        return result;
    }

这里还涉及到 Dubbo 另外一个核心机制——Mock。Mock 可以在测试中模拟服务调用的各种异常情况，还用来实现服务降级。 从 MockClusterWrapper.join() 方法可知，实际创建的 ClusterInvoker 是封装了 FailoverClusterInvoker 的 MockerClusterInvoker。

在 MockerClusterInvoker 中，调用之前 Dubbo 会先检查 URL 中是否有 mock 参数（通过服务治理后台 Consumer 端的屏蔽和容错进行设置，或者直接动态设置 mock 参数值），如果存在且以 force 开头，则不发起远程调用直接执行降级逻辑；如果存在且以 fail 开头，则在远程调用异常时才会执行降级逻辑。

因此，通过 MockerClusterWrapper 成功地在 Invoker 中植入了 Mock 机制。

### 2.3.2 Directory ###

在 RegistryProtocol\#doRefer() 中可以看到，服务发现过程是通过 RegistryDirectory 向 Zookeeper 订阅来实现的。 先看看 Directory 类之间的关系：

![fdc654a869f6743354c81329d795af23.png][]

看下 Directory 接口的定义：

**Directory.java**

public interface Directory<T> extends Node {
        
        Class<T> getInterface();
    
        List<Invoker<T>> list(Invocation invocation) throws RpcException;
    }

Directory 可以看做是对应 Interface 的 Invoker 列表，而这个列表可能是动态变化的，比如注册中心推送变更。

通过 ReferenceConfig\#createProxy() 方法可知，StaticDirectory 主要用于多注册中心引用的场景，它的 invoker 列表是通过参数传入的、固定的。在此不做更详细的解析了。

RegistryDirectory 用于使用单注册中心发现服务的场景。RegistryDirectory 没有重写 list() 方法，所以使用 AbstractDirectory\#list() 方法：

**AbstractDirectory.java**

public List<Invoker<T>> list(Invocation invocation) throws RpcException {
        if (destroyed) {
            throw new RpcException("Directory already destroyed .url: " + getUrl());
        }
        List<Invoker<T>> invokers = doList(invocation);
        List<Router> localRouters = this.routers; // local reference
        if (localRouters != null && !localRouters.isEmpty()) {
            for (Router router : localRouters) {
                try {
                    if (router.getUrl() == null || 
                             router.getUrl().getParameter(Constants.RUNTIME_KEY, false)) {
                        invokers = router.route(invokers, getConsumerUrl(), invocation);
                    }
                } catch (Throwable t) {
                    logger.error("Failed to execute router: " + getUrl() + 
                                  ", cause: " + t.getMessage(), t);
                }
            }
        }
        return invokers;
    }

**RegistryDirectory.java**

/**
     * 获取 invoker 列表
     */
    public List<Invoker<T>> doList(Invocation invocation) {
        if (forbidden) {
            throw new RpcException(RpcException.FORBIDDEN_EXCEPTION, "Forbid consumer " +
                          NetUtils.getLocalHost() + " access service " + getInterface().getName() + 
                          " from registry " + getUrl().getAddress() + " use dubbo version " +
                           Version.getVersion() + 
                           ", Please check registry access list (whitelist/blacklist).");
        }
        List<Invoker<T>> invokers = null;
        Map<String, List<Invoker<T>>> localMethodInvokerMap = this.methodInvokerMap;  //本地缓存
        if (localMethodInvokerMap != null && localMethodInvokerMap.size() > 0) {
            String methodName = RpcUtils.getMethodName(invocation);
            //根据方法名从本地缓存中获取invoker列表，此处略
            //……
        }
        return invokers == null ? new ArrayList<Invoker<T>>(0) : invokers;
    }
    
    
    /**
     * 节点变更通知
     */
    public synchronized void notify(List<URL> urls) {
        List<URL> invokerUrls = new ArrayList<URL>();
        List<URL> routerUrls = new ArrayList<URL>();
        List<URL> configuratorUrls = new ArrayList<URL>();
        for (URL url : urls) {
            String protocol = url.getProtocol();
            String category = url.getParameter(Constants.CATEGORY_KEY, Constants.DEFAULT_CATEGORY);
            if (Constants.ROUTERS_CATEGORY.equals(category) 
                    || Constants.ROUTE_PROTOCOL.equals(protocol)) {
                routerUrls.add(url);
            } else if (Constants.CONFIGURATORS_CATEGORY.equals(category) 
                    || Constants.OVERRIDE_PROTOCOL.equals(protocol)) {
                configuratorUrls.add(url);
            } else if (Constants.PROVIDERS_CATEGORY.equals(category)) {
                invokerUrls.add(url);
            } else {
                logger.warn("Unsupported category " + category + " in notified url: " + 
                            url + " from registry " + getUrl().getAddress() + 
                            " to consumer " + NetUtils.getLocalHost());
            }
        }
        // configurators 
        if (configuratorUrls != null && configuratorUrls.size() >0 ){
            this.configurators = toConfigurators(configuratorUrls);
        }
        // routers
        if (routerUrls != null && routerUrls.size() >0 ){
            List<Router> routers = toRouters(routerUrls);
            if(routers != null){ // null - do nothing
                setRouters(routers);
            }
        }
        List<Configurator> localConfigurators = this.configurators; // local reference
        // 合并override参数
        this.overrideDirectoryUrl = directoryUrl;
        if (localConfigurators != null && localConfigurators.size() > 0) {
            for (Configurator configurator : localConfigurators) {
                this.overrideDirectoryUrl = configurator.configure(overrideDirectoryUrl);
            }
        }
        // providers
        refreshInvoker(invokerUrls);
    }
    
    
    /**
     * 根据invokerURL列表转换为invoker列表
     */
    private void refreshInvoker(List<URL> invokerUrls){
        //......
        Map<String, Invoker<T>> newUrlInvokerMap = toInvokers(invokerUrls) ;// 将URL列表转成Invoker列表
        //......
    }
    
    
    /**
     * 合并url参数 顺序为override > -D >Consumer > Provider
     */
    private Map<String, Invoker<T>> toInvokers(List<URL> urls) {
        //......
        URL url = mergeUrl(providerUrl);
        //......
    }

在 dolist() 方法中，如果通过服务治理禁止 Consumer 访问的话，此处直接抛出响应的异常。

RegistryDirectory 实现了 NotifyListener，在 ZK 节点变化时能收到通知更新内存缓存，其中 RegistryDirectory\#mergeUrl() 方法中会按照优先级合并参数（动态配置在此处生效）。

服务引用时从内存缓存中获取并返回invoker列表，并根据路由规则再进行一次过滤。

### 2.3.3 Router ###

Router 的作用就是从 Directory 的 invoker 列表中刷选出符合路由规则的 invoker 子集。目前 Dubbo 提供了基于IP、应用名和协议等的静态路由功能，功能和实现比较简单，在此不做过多解释。

### 2.3.4 LoadBalance ###

通过 Direcotry 和 Router 之后，返回的是可用的 invoker 子集；在发起服务调用时，需要通过 LoadBalance 选择最终的一个目标 invoker。

在上面的 Cluster 章节中我们知道，调用时首先会经过 MockerClusterInvoker 拦截 Mock 设置，然后再根据配置调用实际的 Invoker（默认是 FailoverClusterInvoker）。

FailoverClusterInvoker 继承 AbstractClusterInvoker，在 AbstractClusterInvoker 中：

**AbstractClusterInvoker.java**

public Result invoke(final Invocation invocation) throws RpcException {
    
        checkWheatherDestoried();
    
        LoadBalance loadbalance;
        
        List<Invoker<T>> invokers = list(invocation);
        if (invokers != null && invokers.size() > 0) {
            loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class)
                         .getExtension(invokers.get(0).getUrl()
                         .getMethodParameter(invocation.getMethodName(),
                                             Constants.LOADBALANCE_KEY, 
                                             Constants.DEFAULT_LOADBALANCE));
        } else {
            loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class)
                         .getExtension(Constants.DEFAULT_LOADBALANCE);
        }
        RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
        return doInvoke(invocation, invokers, loadbalance);
    }

**FailoverClusterInvoker.java**

public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
        List<Invoker<T>> copyinvokers = invokers;
        checkInvokers(copyinvokers, invocation);
        int len = getUrl().getMethodParameter(invocation.getMethodName(), 
                           Constants.RETRIES_KEY, Constants.DEFAULT_RETRIES) + 1;
        if (len <= 0) {
            len = 1;
        }
        // retry loop.
        RpcException le = null; // last exception.
        List<Invoker<T>> invoked = new ArrayList<Invoker<T>>(copyinvokers.size());
        Set<String> providers = new HashSet<String>(len);
        for (int i = 0; i < len; i++) {
            //重试时，进行重新选择，避免重试时invoker列表已发生变化.
            //注意：如果列表发生了变化，那么invoked判断会失效，因为invoker示例已经改变
            if (i > 0) {
                checkWheatherDestoried();
                copyinvokers = list(invocation);
                //重新检查一下
                checkInvokers(copyinvokers, invocation);
            }
            Invoker<T> invoker = select(loadbalance, invocation, copyinvokers, invoked);
            invoked.add(invoker);
            RpcContext.getContext().setInvokers((List)invoked);
            try {
                Result result = invoker.invoke(invocation);
                if (le != null && logger.isWarnEnabled()) {
                    logger.warn("", le);
                }
                return result;
            } catch (RpcException e) {
                if (e.isBiz()) { // biz exception.
                    throw e;
                }
                le = e;
            } catch (Throwable e) {
                le = new RpcException(e.getMessage(), e);
            } finally {
                providers.add(invoker.getUrl().getAddress());
            }
        }
        throw new RpcException();
    }

在调用或重试时，每次都通过 LoadBalance 选出一个 Invoker 进行调用。

至此，调用流程结束。

[Link 1]: https://github.com/cyfonly/dubbo-read
[Dubbo]: http://dubbo.apache.org/books/dubbo-dev-book/design.html
[3c9c7e4202035d1c4b7749e057959d35.png]: /images/20221120/bf97aa7492964056bff22833a1d73ffa.png
[Dubbo 1]: http://dubbo.apache.org/books/dubbo-user-book/demos/fault-tolerent-strategy.html
[701f9cdc1a63d415e7756eec01c118ca.png]: /images/20221120/e0a5adda37f1436cab238326c585da14.png
[85efd80c9deaf41f0f7086e9494b3806.png]: /images/20221120/4aa306fe477b438e990ac998e5449074.png
[Link 2]: http://dubbo.apache.org/books/dubbo-dev-book/principals/general-knowledge.html
[fdc654a869f6743354c81329d795af23.png]: /images/20221120/ca131fae07db48699956d1d781eceb62.png