dubbo服務引用與叢集容錯

阿新 • • 發佈：2018-12-23

服務引用無非就是做了兩件事

將spring的schemas標籤資訊轉換bean,然後通過這個bean的資訊,連線、訂閱zookeeper節點資訊建立一個invoker
將invoker的資訊建立一個動態代理物件

時序圖：

最終返回一個被呼叫介面的動態代理物件。

在呼叫代理物件的方法時，會進入InvokerInvocationHandle類的邏輯。

跟蹤原始碼的時候，發現消費端呼叫invoke的時候要呼叫一連串的Invoker實現類，一直糾結這些Invoker是用來做什麼的？

Invoker的建立應該是入口，也就是從referenceConfig類開始

然後找到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);
    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);
}

也就是這一行：
cluster.join(directory);

在執行join方法的時候，會通過SPI機制找到cluster的擴充套件例項，預設的時候FailoverCluster
但是除錯發現第一步建立的例項化物件是MockClusterWrapper類而不是FailoverCluster 查閱資料 dubbo中的mock機制再結合原始碼總結如下：在dubbo的配置檔案 classpath:/META-INF/dubbo/internal/com.alibaba.dubbo.rpc.cluster.Cluster中，failover對應的是FailoverCluster類：
但是ExtensionLoader在例項化物件時，有個比較特殊的地方，那就是在例項化完成之後，會自動套上當前的ExtensionLoader中的Wrapper類，上面的mock所對應的MockClusterWrapper就是這樣的一個Wrapper：也就是例項化出來的FailoverCluster會被套上一層MockClusterWrapper，總結一下就是： Cluster$Adaptive -> 定位到內部key為failover的物件 ->FailoverCluster->外部套上MockClusterWrapper 所以時序圖是這樣的：

官網叢集容錯介紹圖：

根據以上時序圖檢視原始碼如下： 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
            //執行到這一行的時候開始進入叢集 cluster -> AbstractClusterInvoker
            result = this.invoker.invoke(invocation);
        } else if (value.startsWith("force")) {
            if (logger.isWarnEnabled()) {
                logger.warn("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.warn("fail-mock: " + invocation.getMethodName() + " fail-mock enabled , url : " + directory.getUrl(), e);
                    }
                    result = doMockInvoke(invocation, e);
                }
            }
        }
        return result;
    }

AbstractClusterInvoker.java

public Result invoke(final Invocation invocation) throws RpcException {
        checkWhetherDestroyed();

        // binding attachments into invocation.
        Map<String, String> contextAttachments = RpcContext.getContext().getAttachments();
        if (contextAttachments != null && contextAttachments.size() != 0) {
            ((RpcInvocation) invocation).addAttachments(contextAttachments);
        }
         
         //選擇出可用的invoker集合
        List<Invoker<T>> invokers = list(invocation);
        // 初始化負載均衡策略
        LoadBalance loadbalance = initLoadBalance(invokers, invocation);
        RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
        return doInvoke(invocation, invokers, loadbalance);
    }


   protected List<Invoker<T>> list(Invocation invocation) throws RpcException {
        // -> AbstractDirectory.java
        return directory.list(invocation);
    }

AbstractDirectory.java

public List<Invoker<T>> list(Invocation invocation) throws RpcException {
        if (destroyed) {
            throw new RpcException("Directory already destroyed .url: " + getUrl());
        }
        
        //  模板方法，由子類實現
        // -> RegistryDirectory.java 或者 StaticDirectory.java 

        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,開始進入路由,現在我們到了序號6,此時到了MockInvokersSelector類,
                         //他是Router介面的實現類,從官網的介紹圖中我們也可以看到Router分為Script和Condition兩種,翻譯過來也就是指令碼路由和條件路由
                        invokers = router.route(invokers, getConsumerUrl(), invocation);
                    }
                } catch (Throwable t) {
                    logger.error("Failed to execute router: " + getUrl() + ", cause: " + t.getMessage(), t);
                }
            }
        }
        return invokers;
    }

RegistryDirectory.java

public List<Invoker<T>> doList(Invocation invocation) {
        if (forbidden) {
            // 1. No service provider 2. Service providers are disabled
            throw new RpcException(RpcException.FORBIDDEN_EXCEPTION,
                    "No provider available from registry " + getUrl().getAddress() + " for service " + getConsumerUrl().getServiceKey() + " on consumer " + NetUtils.getLocalHost()
                            + " use dubbo version " + Version.getVersion() + ", please check status of providers(disabled, not registered or in blacklist).");
        }
        List<Invoker<T>> invokers = null;
        Map<String, List<Invoker<T>>> localMethodInvokerMap = this.methodInvokerMap; // local reference
        if (localMethodInvokerMap != null && localMethodInvokerMap.size() > 0) {
            String methodName = RpcUtils.getMethodName(invocation);
            Object[] args = RpcUtils.getArguments(invocation);
            if (args != null && args.length > 0 && args[0] != null
                    && (args[0] instanceof String || args[0].getClass().isEnum())) {
                invokers = localMethodInvokerMap.get(methodName + "." + args[0]); // The routing can be enumerated according to the first parameter
            }
            if (invokers == null) {
                invokers = localMethodInvokerMap.get(methodName);
            }
            if (invokers == null) {
                invokers = localMethodInvokerMap.get(Constants.ANY_VALUE);
            }
            if (invokers == null) {
                Iterator<List<Invoker<T>>> iterator = localMethodInvokerMap.values().iterator();
                if (iterator.hasNext()) {
                    invokers = iterator.next();
                }
            }
        }
        return invokers == null ? new ArrayList<Invoker<T>>(0) : invokers;
    }

MockInvokersSelector.java

public <T> List<Invoker<T>> route(final List<Invoker<T>> invokers,
                                      URL url, final Invocation invocation) throws RpcException {
        if (invocation.getAttachments() == null) {
            return getNormalInvokers(invokers);
        } else {
            String value = invocation.getAttachments().get(Constants.INVOCATION_NEED_MOCK);
            if (value == null) {
                //拿到能正常執行的invokers,並將其返回
                return getNormalInvokers(invokers);
            } else if (Boolean.TRUE.toString().equalsIgnoreCase(value)) {
                return getMockedInvokers(invokers);
            }
        }
        return invokers;
    }


    private <T> List<Invoker<T>> getNormalInvokers(final List<Invoker<T>> invokers) {
        if (!hasMockProviders(invokers)) {
            return invokers;
        } else {
            List<Invoker<T>> sInvokers = new ArrayList<Invoker<T>>(invokers.size());
            for (Invoker<T> invoker : invokers) {
                if (!invoker.getUrl().getProtocol().equals(Constants.MOCK_PROTOCOL)) {
                    sInvokers.add(invoker);
                }
            }
            return sInvokers;
        }
    }

上面出現的這兩個關鍵詞,其實無非就是做兩件事在Directory中找出本次叢集中的全部invokers
在Router中,將上一步的全部invokers挑選出能正常執行的invokers 回到AbstractClusterInvoker.java

      ......
        //選擇出可用的invoker集合
        List<Invoker<T>> invokers = list(invocation);
         // 初始化負載均衡策略
        LoadBalance loadbalance = initLoadBalance(invokers, invocation);
        RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
        return doInvoke(invocation, invokers, loadbalance);

從上面步驟已經挑選出能正常執行的invokers了,但是假如2個做叢集,但是這兩個都是正常的,到底要執行哪一個呢? 根據官網的描述

在叢集呼叫失敗時，Dubbo 提供了多種容錯方案，預設為 failover 重試。

所以這個時候是到了FailoverClusterInvoker類,但是如果你配置的是Failfast Cluster(快速失敗),Failsafe Cluster(失敗安全),Failback Cluster(失敗自動恢復),Forking Cluster(並行呼叫多個伺服器，只要一個成功即返回),Broadcast Cluster(廣播呼叫所有提供者，逐個呼叫，任意一臺報錯則報錯)他也會到達相應的類 FailoverClusterInvoker.java

public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
        List<Invoker<T>> copyinvokers = invokers;
        checkInvokers(copyinvokers, invocation);
        String methodName = RpcUtils.getMethodName(invocation);
        int len = getUrl().getMethodParameter(methodName, 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()); // invoked invokers.
        Set<String> providers = new HashSet<String>(len);
        for (int i = 0; i < len; i++) {
            //Reselect before retry to avoid a change of candidate `invokers`.
            //NOTE: if `invokers` changed, then `invoked` also lose accuracy.
            if (i > 0) {
                checkWhetherDestroyed();
                copyinvokers = list(invocation);
                // check again
                checkInvokers(copyinvokers, invocation);
            }
            // 通過負載均衡演算法選擇一個Invoker,然後呼叫
            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("Although retry the method " + methodName
                            + " in the service " + getInterface().getName()
                            + " was successful by the provider " + invoker.getUrl().getAddress()
                            + ", but there have been failed providers " + providers
                            + " (" + providers.size() + "/" + copyinvokers.size()
                            + ") from the registry " + directory.getUrl().getAddress()
                            + " on the consumer " + NetUtils.getLocalHost()
                            + " using the dubbo version " + Version.getVersion() + ". Last error is: "
                            + le.getMessage(), 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(le.getCode(), "Failed to invoke the method "
                + methodName + " in the service " + getInterface().getName()
                + ". Tried " + len + " times of the providers " + providers
                + " (" + providers.size() + "/" + copyinvokers.size()
                + ") from the registry " + directory.getUrl().getAddress()
                + " on the consumer " + NetUtils.getLocalHost() + " using the dubbo version "
                + Version.getVersion() + ". Last error is: "
                + le.getMessage(), le.getCause() != null ? le.getCause() : le);
    }

    protected Invoker<T> select(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {
        if (invokers == null || invokers.isEmpty()) {
            return null;
        }
        String methodName = invocation == null ? "" : invocation.getMethodName();

        boolean sticky = invokers.get(0).getUrl().getMethodParameter(methodName, Constants.CLUSTER_STICKY_KEY, Constants.DEFAULT_CLUSTER_STICKY);
        {
            //ignore overloaded method
            if (stickyInvoker != null && !invokers.contains(stickyInvoker)) {
                stickyInvoker = null;
            }
            //ignore concurrency problem
            if (sticky && stickyInvoker != null && (selected == null || !selected.contains(stickyInvoker))) {
                if (availablecheck && stickyInvoker.isAvailable()) {
                    return stickyInvoker;
                }
            }
        }
        Invoker<T> invoker = doSelect(loadbalance, invocation, invokers, selected);

        if (sticky) {
            stickyInvoker = invoker;
        }
        return invoker;
    }

    private Invoker<T> doSelect(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {
        if (invokers == null || invokers.isEmpty())
            return null;
        if (invokers.size() == 1)
            return invokers.get(0);
        Invoker<T> invoker = loadbalance.select(invokers, getUrl(), invocation);

        //If the `invoker` is in the  `selected` or invoker is unavailable && availablecheck is true, reselect.
        if ((selected != null && selected.contains(invoker))
                || (!invoker.isAvailable() && getUrl() != null && availablecheck)) {
            try {
                Invoker<T> rinvoker = reselect(loadbalance, invocation, invokers, selected, availablecheck);
                if (rinvoker != null) {
                    invoker = rinvoker;
                } else {
                    //Check the index of current selected invoker, if it's not the last one, choose the one at index+1.
                    int index = invokers.indexOf(invoker);
                    try {
                        //Avoid collision
                        invoker = index < invokers.size() - 1 ? invokers.get(index + 1) : invokers.get(0);
                    } catch (Exception e) {
                        logger.warn(e.getMessage() + " may because invokers list dynamic change, ignore.", e);
                    }
                }
            } catch (Throwable t) {
                logger.error("cluster reselect fail reason is :" + t.getMessage() + " if can not solve, you can set cluster.availablecheck=false in url", t);
            }
        }
        return invoker;
    }

在這個叢集容錯的整體架構過程中,dubbo其實也就是三件事 1.在Directory中找出本次叢集中的全部invokers 2.在Router中,將上一步的全部invokers挑選出能正常執行的invokers 3.在LoadBalance中,將上一步的能正常的執行invokers中,根據配置的負載均衡策略,挑選出需要執行的invoker 參考肥朝 dubbo原始碼解析-叢集容錯架構設計

dubbo服務引用與叢集容錯

dubbo服務引用與叢集容錯

dubbo高可用之zookeeper宕機、Dubbo直連、負載均衡、服務降級、叢集容錯

dubbo原始碼解析五 --- 叢集容錯架構設計與原理分析

Dubbo服務引用原理

dubbo高可用：叢集容錯（十四）

dubbo工作原理，叢集容錯，負載均衡

Dubbo學習(四)：叢集容錯

搞懂Dubbo服務釋出與服務註冊

Dubbo服務暴露與註冊

Dubbo服務註冊與發現

Dubbo服務叢集，常見容錯機制：failover ，failsafe，failfase ，failback，forking

Dubbo服務叢集容錯配置（四）

DUBBO叢集容錯與負載均衡

轉載：java分布式服務框架Dubbo的介紹與使用

dubbo refrence bean（服務引用）

dubbo 叢集容錯模式原始碼學習---ForkingCluster

dubbo 叢集容錯模式原始碼學習 -- FaibackCluster

Dubbo原始碼解析之服務釋出與註冊

dubbo叢集容錯

SpringCloud-Eureka服務註冊與發現之叢集配置(五)

dubbo服務引用與叢集容錯

相關推薦