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

這裏的 Invoker 是 Provider 的一個可調用 Service 的抽象，Invoker 封裝了 Provider 地址及 Service 接口信息
Directory 表明多個 Invoker，能夠把它當作 List<Invoker> ，但與 List 不一樣的是，它的值多是動態變化的，好比註冊中心推送變動
Cluster 將 Directory 中的多個 Invoker 假裝成一個 Invoker，對上層透明，假裝過程包含了容錯邏輯，調用失敗後，重試另外一個
Router 負責從多個 Invoker 中按路由規則選出子集，好比讀寫分離，應用隔離等
LoadBalance 負責從多個 Invoker 中選出具體的一個用於本次調用，選的過程包含了負載均衡算法，調用失敗後，須要重選

集羣容錯模式

能夠自行擴展集羣容錯策略，參見：集羣擴展api

Failover Cluster

失敗自動切換，當出現失敗，重試其它服務器 [1]。一般用於讀操做，但重試會帶來更長延遲。可經過 retries="2" 來設置重試次數(不含第一次)。緩存

重試次數配置以下：

<dubbo:service retries="2" />

或

<dubbo:reference retries="2" />

或

<dubbo:reference>
    <dubbo:method name="findFoo" retries="2" />
</dubbo:reference>

Failfast Cluster

快速失敗，只發起一次調用，失敗當即報錯。一般用於非冪等性的寫操做，好比新增記錄。

Failsafe Cluster

失敗安全，出現異常時，直接忽略。一般用於寫入審計日誌等操做。

Failback Cluster

失敗自動恢復，後臺記錄失敗請求，定時重發。一般用於消息通知操做。

Forking Cluster

並行調用多個服務器，只要一個成功即返回。一般用於實時性要求較高的讀操做，但須要浪費更多服務資源。可經過 forks="2" 來設置最大並行數。

Broadcast Cluster

廣播調用全部提供者，逐個調用，任意一臺報錯則報錯 [2]。一般用於通知全部提供者更新緩存或日誌等本地資源信息。

集羣模式配置

按照如下示例在服務提供方和消費方配置集羣模式

<dubbo:service cluster="failsafe" />

或

<dubbo:reference cluster="failsafe" />

Dubbo集羣容錯的架構分析

經過官網上這張圖咱們能大體的瞭解到一個請求過來，在集羣中的調用過程。那麼咱們就根據這個調用過程來進行分析吧。

整個在調用的過程當中這三個關鍵詞接下來會貫穿全文,他們就是Directory,Router,LoadBalance

咱們只要緊緊的抓住這幾個關鍵字就能貫穿整個調用鏈

先看下時序圖，來看下調用的過程

最初咱們一個方法調用

咱們使用的是官方的dubbo-demo的dubbo-demo-consumer

public static void main(String[] args) {
    DemoService demoService = (DemoService) context.getBean("demoService"); // get remote service proxy
    String hello = demoService.sayHello("world"); // call remote method
    System.out.println(hello); // get result
   
}

調用 InvokerInvocationHandler#invoker方法代理類的調用

public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
    String methodName = method.getName();
    Class<?>[] parameterTypes = method.getParameterTypes();
    if (method.getDeclaringClass() == Object.class) {
        return method.invoke(invoker, args);
    }
    if ("toString".equals(methodName) && parameterTypes.length == 0) {
        return invoker.toString();
    }
    if ("hashCode".equals(methodName) && parameterTypes.length == 0) {
        return invoker.hashCode();
    }
    if ("equals".equals(methodName) && parameterTypes.length == 1) {
        return invoker.equals(args[0]);
    }

    RpcInvocation invocation;
    if (RpcUtils.hasGeneratedFuture(method)) {
        Class<?> clazz = method.getDeclaringClass();
        String syncMethodName = methodName.substring(0, methodName.length() - Constants.ASYNC_SUFFIX.length());
        Method syncMethod = clazz.getMethod(syncMethodName, method.getParameterTypes());
        invocation = new RpcInvocation(syncMethod, args);
        invocation.setAttachment(Constants.FUTURE_GENERATED_KEY, "true");
        invocation.setAttachment(Constants.ASYNC_KEY, "true");
    } else {
        invocation = new RpcInvocation(method, args);
        if (RpcUtils.hasFutureReturnType(method)) {
            invocation.setAttachment(Constants.FUTURE_RETURNTYPE_KEY, "true");
            invocation.setAttachment(Constants.ASYNC_KEY, "true");
        }
    }
    //這裏使用的是動態代理的方式獲取到指定的代理類
   // <1>
    return invoker.invoke(invocation).recreate();
}

1 執行invoke就要開始進入MockClusterInvoker#invoker

public Result invoke(Invocation invocation) throws RpcException {
    Result result = null;
    // 得到 「mock」 配置項，有多種配置方式
    String value = directory.getUrl().getMethodParameter(invocation.getMethodName(), Constants.MOCK_KEY, Boolean.FALSE.toString()).trim();
    //【第一種】無 mock
    if (value.length() == 0 || value.equalsIgnoreCase("false")) {
        //no mock
        // 調用原 Invoker ，發起 RPC 調用
        // 調用 invoker方法，進入到集羣也就是CLuster類中
        //<2>
        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
        // 直接調用 Mock Invoker ，執行本地 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);
                }
                // 失敗後，調用 Mock Invoker ，執行本地 Mock 邏輯
                result = doMockInvoke(invocation, e);
            }
        }
    }
    return result;
}

2 進入到 invoke就要開始進入到集羣,也就是Cluster

/**
 * 調用服務提供者
 * @param invocation
 * @return
 * @throws RpcException
 */
@Override
public Result invoke(final Invocation invocation) throws RpcException {
    // 校驗是否銷燬
    checkWhetherDestroyed();

    //TODO
    // binding attachments into invocation.
    Map<String, String> contextAttachments = RpcContext.getContext().getAttachments();
    if (contextAttachments != null && contextAttachments.size() != 0) {
        ((RpcInvocation) invocation).addAttachments(contextAttachments);
    }

    // 得到全部服務提供者 Invoker 集合
    // <下面的list方法>
    List<Invoker<T>> invokers = list(invocation);
    // 得到 LoadBalance 對象
    LoadBalance loadbalance = initLoadBalance(invokers, invocation);
    // 設置調用編號，如果異步調用
    RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
    // 執行調用
    return doInvoke(invocation, invokers, loadbalance);
}

/**
 *  得到全部服務提供者 Invoker 集合
 * @param invocation
 * @return
 * @throws RpcException
 */
protected List<Invoker<T>> list(Invocation invocation) throws RpcException {
    // 經過directory 進入到 AbstractDirectory 中選擇 directory
    //<3 進入3 裏面>
    return directory.list(invocation);
}

3 進入到 AbstractDirectory 進行 directory的選擇

/**
 * 得到全部服務 Invoker 集合
 * @param invocation
 * @return Invoker 集合
 * @throws RpcException
 */
@Override
public List<Invoker<T>> list(Invocation invocation) throws RpcException {
    //當銷燬時拋出異常
    if (destroyed) {
        throw new RpcException("Directory already destroyed .url: " + getUrl());
    }
    // 得到全部 Invoker 集合
    // <4 RegistryDirectory 選擇 invoker>
    List<Invoker<T>> invokers = doList(invocation);
    //根據路由規則，篩選Invoker集合
    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);
            }
        }
    }
      //< 6 獲取 router 即將進入 MockInvokersSelector 類中>
    return invokers;
}

調用list方法會進入到 RegistryDirectory#doList

/**
 *  得到對應的 Invoker 集合。
 * @param invocation
 * @return
 */
@Override
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;
    //從methodInvokerMap中取出invokers
    Map<String, List<Invoker<T>>> localMethodInvokerMap = this.methodInvokerMap; // local reference
    // 得到 Invoker 集合
    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
        }
        // 【第二】根據方法名得到 Invoker 集合
        if (invokers == null) {
            invokers = localMethodInvokerMap.get(methodName);
        }
        // 【第三】使用全量 Invoker 集合。例如，`#$echo(name)` ，回聲方法
        if (invokers == null) {
            invokers = localMethodInvokerMap.get(Constants.ANY_VALUE);
        }
        // 【第四】使用 `methodInvokerMap` 第一個 Invoker 集合。防護性編程。
        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;
}

6 進入 MockInvokersSelector 類中根據路由規則拿到正常執行的invokers

/**
 * ，根據 "invocation.need.mock" 路由匹配對應類型的 Invoker 集合：
 * @param invokers Invoker 集合
 * @param url        refer url
 * @param invocation
 * @param <T>
 * @return
 * @throws RpcException
 */
@Override
public <T> List<Invoker<T>> route(final List<Invoker<T>> invokers,
                                  URL url, final Invocation invocation) throws RpcException {
    // 得到普通 Invoker 集合
    if (invocation.getAttachments() == null) {
        //<7> 拿到能正常執行的invokers,並將其返回.也就是序號7
        return getNormalInvokers(invokers);
    } else {
        // 得到 "invocation.need.mock" 配置項
        String value = invocation.getAttachments().get(Constants.INVOCATION_NEED_MOCK);
        // 得到普通 Invoker 集合
        if (value == null)
            return getNormalInvokers(invokers);
        // 得到 MockInvoker 集合
        else if (Boolean.TRUE.toString().equalsIgnoreCase(value)) {
            return getMockedInvokers(invokers);
        }
    }
    // 其它，不匹配，直接返回 `invokers` 集合
    return invokers;
}

<7> 拿到能正常執行的invokers,並將其返回

/**
 * 得到普通 Invoker 集合
 * @param invokers
 * @param <T>
 * @return
 */
private <T> List<Invoker<T>> getNormalInvokers(final List<Invoker<T>> invokers) {
    // 不包含 MockInvoker 的狀況下，直接返回 `invokers` 集合
    if (!hasMockProviders(invokers)) {
        return invokers;
    } else {
        // 若包含 MockInvoker 的狀況下，過濾掉 MockInvoker ，建立普通 Invoker 集合
        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;
    }
}

8 拿到 invoker 返回到 AbstractClusterInvoker這個類

對於上面的這些步驟，主要用於作兩件事

在Directory中找出本次集羣中的所有invokers

在Router中,將上一步的所有invokers挑選出能正常執行的invokers

在時序圖的序號5和序號7處，作了上訴的處理。

在有多個集羣的狀況下，並且兩個集羣都是正常的，那麼到底須要執行哪一個？

AbstractClusterInvoker#invoke

/**
 * 調用服務提供者
 * @param invocation
 * @return
 * @throws RpcException
 */
@Override
public Result invoke(final Invocation invocation) throws RpcException {
    // 校驗是否銷燬
    checkWhetherDestroyed();

    //TODO
    // 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 loadbalance = initLoadBalance(invokers, invocation);
    // 設置調用編號，如果異步調用
    RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
    // 執行調用
    return doInvoke(invocation, invokers, loadbalance);
}

/**
 *  實現子 Cluster 的 Invoker 實現類的服務調用的差別邏輯，
 * @param invocation
 * @param invokers
 * @param loadbalance
 * @return
 * @throws RpcException
 */
 // 抽象方法，子類自行的實現  由於咱們使用的默認配置，因此 咱們將會是FailoverClusterInvoker 這個類
 //< 8 doInvoker 方法>
protected abstract Result doInvoke(Invocation invocation, List<Invoker<T>> invokers,
                                   LoadBalance loadbalance) throws RpcException;

9 進入到相應的集羣容錯方案類中由於咱們使用的默認配置，因此咱們將會是FailoverClusterInvoker 這個類

/**
 * 實際邏輯很簡單：循環，查找一個 Invoker 對象，進行調用，直到成功
 * @param invocation
 * @param invokers
 * @param loadbalance
 * @return
 * @throws RpcException
 */
@Override
@SuppressWarnings({"unchecked", "rawtypes"})
public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
    List<Invoker<T>> copyinvokers = invokers;
    // 檢查copyinvokers便可用Invoker集合是否爲空，若是爲空，那麼拋出異常
    checkInvokers(copyinvokers, invocation);
    String methodName = RpcUtils.getMethodName(invocation);
    // 獲得最大可調用次數：最大可重試次數+1，默認最大可重試次數Constants.DEFAULT_RETRIES=2
    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);
    // failover機制核心實現：若是出現調用失敗，那麼重試其餘服務器
    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.
        // 重試時，進行從新選擇，避免重試時invoker列表已發生變化.
        // 注意：若是列表發生了變化，那麼invoked判斷會失效，由於invoker示例已經改變
        if (i > 0) {
            checkWhetherDestroyed();
            // 根據Invocation調用信息從Directory中獲取全部可用Invoker
            copyinvokers = list(invocation);
            // check again
            // 從新檢查一下
            checkInvokers(copyinvokers, invocation);
        }
        // 根據負載均衡機制從copyinvokers中選擇一個Invoker
         //< 9 select ------------------------- 下面將進行 invoker選擇----------------------------------->
        Invoker<T> invoker = select(loadbalance, invocation, copyinvokers, invoked);
        // 保存每次調用的Invoker
        invoked.add(invoker);
        // 設置已經調用的 Invoker 集合，到 Context 中
        RpcContext.getContext().setInvokers((List) invoked);
        try {
            // RPC 調用獲得 Result
            Result result = invoker.invoke(invocation);
            // 重試過程當中，將最後一次調用的異常信息以 warn 級別日誌輸出
            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;
            }
            // 其餘性質的異常統一封裝成RpcException
            le = e;
        } catch (Throwable e) {
            le = new RpcException(e.getMessage(), e);
        } finally {
            providers.add(invoker.getUrl().getAddress());
        }
    }
    // 最大可調用次數用完還獲得Result的話，拋出RpcException異常：重試了N次仍是失敗，並輸出最後一次異常信息
    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);
}

9 select 使用 loadbalance 選擇 invoker

/**
 * Select a invoker using loadbalance policy.</br>
 * a) Firstly, select an invoker using loadbalance. If this invoker is in previously selected list, or,
 * if this invoker is unavailable, then continue step b (reselect), otherwise return the first selected invoker</br>
 * <p>
 * b) Reselection, the validation rule for reselection: selected > available. This rule guarantees that
 * the selected invoker has the minimum chance to be one in the previously selected list, and also
 * guarantees this invoker is available.
 *
 * @param loadbalance load balance policy
 * @param invocation  invocation
 * @param invokers    invoker candidates
 * @param selected    exclude selected invokers or not
 * @return the invoker which will final to do invoke.
 * @throws RpcException
 */
/**
 * 使用 loadbalance 選擇 invoker.
 * a)
 * @param loadbalance  對象，提供負責均衡策略
 * @param invocation 對象
 * @param invokers 候選的 Invoker 集合
 * @param selected 已選過的 Invoker 集合. 注意：輸入保證不重複
 * @return 最終的 Invoker 對象
 * @throws RpcException
 */
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;
    // 得到 sticky 配置項，方法級
    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
        // 若 stickyInvoker 不存在於 invokers 中，說明不在候選中，須要置空，從新選擇
        if (stickyInvoker != null && !invokers.contains(stickyInvoker)) {
            stickyInvoker = null;
        }
        //ignore concurrency problem
        // 若開啓粘滯鏈接的特性，且 stickyInvoker 不存在於 selected 中，則返回 stickyInvoker 這個 Invoker 對象
        if (sticky && stickyInvoker != null && (selected == null || !selected.contains(stickyInvoker))) {
            // 若開啓排除非可用的 Invoker 的特性，則校驗 stickyInvoker 是否可用。若可用，則進行返回
            if (availablecheck && stickyInvoker.isAvailable()) {
                return stickyInvoker;
            }
        }
    }
    // 執行選擇
    //<  10 -----------------------------進行選擇------------------------------------->
    Invoker<T> invoker = doSelect(loadbalance, invocation, invokers, selected);

    // 若開啓粘滯鏈接的特性，記錄最終選擇的 Invoker 到 stickyInvoker
    if (sticky) {
        stickyInvoker = invoker;
    }
    return invoker;
}

10 doSelect 從候選的 Invoker 集合，選擇一個最終調用的 Invoker 對象

/**
 *  從候選的 Invoker 集合，選擇一個最終調用的 Invoker 對象
 * @param loadbalance
 * @param invocation
 * @param invokers
 * @param selected
 * @return
 * @throws RpcException
 */
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;
    // 若是隻有一個 Invoker ，直接選擇
    if (invokers.size() == 1)
        return invokers.get(0);
    // 使用 Loadbalance ，選擇一個 Invoker 對象。
    //<11 ------------------ 根據LoadBalance(負載均衡) 選擇一個合適的invoke>
    Invoker<T> invoker = loadbalance.select(invokers, getUrl(), invocation);

    //If the `invoker` is in the  `selected` or invoker is unavailable && availablecheck is true, reselect.
    // 若是 selected中包含（優先判斷） 或者 不可用&&availablecheck=true 則重試.
    if ((selected != null && selected.contains(invoker))
            || (!invoker.isAvailable() && getUrl() != null && availablecheck)) {
        try {
            //重選一個 Invoker 對象
            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.
                //看下第一次選的位置，若是不是最後，選+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;
}

11 AbstractLoadBalance#select

此方法是抽象方法，須要各類子類去實現

/**
 *  抽象方法，下面的實現類來實現這個選擇invoker 的方法
 *  各個負載均衡的類自行實現提供自定義的負載均衡策略。
 * @param invokers
 * @param url
 * @param invocation
 * @param <T>
 * @return
 */
protected abstract <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation);

12 RoundRobinLoadBalance 實現父類的抽象方法

@Override
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
    String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();
    int length = invokers.size(); // Number of invokers
    int maxWeight = 0; // The maximum weight
    int minWeight = Integer.MAX_VALUE; // The minimum weight
    final LinkedHashMap<Invoker<T>, IntegerWrapper> invokerToWeightMap = new LinkedHashMap<Invoker<T>, IntegerWrapper>();
    int weightSum = 0;
    // 計算最小、最大權重，總的權重和。
    for (int i = 0; i < length; i++) {
        int weight = getWeight(invokers.get(i), invocation);
        maxWeight = Math.max(maxWeight, weight); // Choose the maximum weight
        minWeight = Math.min(minWeight, weight); // Choose the minimum weight
        if (weight > 0) {
            invokerToWeightMap.put(invokers.get(i), new IntegerWrapper(weight));
            weightSum += weight;
        }
    }
    // 計算最小、最大權重，總的權重和。
    AtomicPositiveInteger sequence = sequences.get(key);
    if (sequence == null) {
        sequences.putIfAbsent(key, new AtomicPositiveInteger());
        sequence = sequences.get(key);
    }
    // 得到當前順序號，並遞增 + 1
    int currentSequence = sequence.getAndIncrement();
    // 權重不相等，順序根據權重分配
    if (maxWeight > 0 && minWeight < maxWeight) {
        int mod = currentSequence % weightSum;// 剩餘權重
        for (int i = 0; i < maxWeight; i++) {// 循環最大權重
            for (Map.Entry<Invoker<T>, IntegerWrapper> each : invokerToWeightMap.entrySet()) {
                final Invoker<T> k = each.getKey();
                final IntegerWrapper v = each.getValue();
                // 剩餘權重歸 0 ，當前 Invoker 還有剩餘權重，返回該 Invoker 對象
                if (mod == 0 && v.getValue() > 0) {
                    return k;
                }
                // 若 Invoker 還有權重值，扣除它( value )和剩餘權重( mod )。
                if (v.getValue() > 0) {
                    v.decrement();
                    mod--;
                }
            }
        }
    }
    // 權重相等，平均順序得到
    // Round robin
    //<13 --------------------------->
    return invokers.get(currentSequence % length);
}

14 FailoverClusterInvoker # doInvoke 方法

/**
 * 實際邏輯很簡單：循環，查找一個 Invoker 對象，進行調用，直到成功
 * @param invocation
 * @param invokers
 * @param loadbalance
 * @return
 * @throws RpcException
 */
@Override
@SuppressWarnings({"unchecked", "rawtypes"})
public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
    List<Invoker<T>> copyinvokers = invokers;
    // 檢查copyinvokers便可用Invoker集合是否爲空，若是爲空，那麼拋出異常
    checkInvokers(copyinvokers, invocation);
    String methodName = RpcUtils.getMethodName(invocation);
    // 獲得最大可調用次數：最大可重試次數+1，默認最大可重試次數Constants.DEFAULT_RETRIES=2
    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);
    // failover機制核心實現：若是出現調用失敗，那麼重試其餘服務器
    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.
        // 重試時，進行從新選擇，避免重試時invoker列表已發生變化.
        // 注意：若是列表發生了變化，那麼invoked判斷會失效，由於invoker示例已經改變
        if (i > 0) {
            checkWhetherDestroyed();
            // 根據Invocation調用信息從Directory中獲取全部可用Invoker
            copyinvokers = list(invocation);
            // check again
            // 從新檢查一下
            checkInvokers(copyinvokers, invocation);
        }
        // 根據負載均衡機制從copyinvokers中選擇一個Invoker
        //< ------------------------- 下面將進行 invoker選擇----------------------------------->
        Invoker<T> invoker = select(loadbalance, invocation, copyinvokers, invoked);
        // 保存每次調用的Invoker
        invoked.add(invoker);
        // 設置已經調用的 Invoker 集合，到 Context 中
        RpcContext.getContext().setInvokers((List) invoked);
        try {
            // RPC 調用獲得 Result
            Result result = invoker.invoke(invocation);
            // 重試過程當中，將最後一次調用的異常信息以 warn 級別日誌輸出
            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;
            }
            // 其餘性質的異常統一封裝成RpcException
            le = e;
        } catch (Throwable e) {
            le = new RpcException(e.getMessage(), e);
        } finally {
            providers.add(invoker.getUrl().getAddress());
        }
    }
    // 最大可調用次數用完還獲得Result的話，拋出RpcException異常：重試了N次仍是失敗，並輸出最後一次異常信息
    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);
}

以上是一個完成調用過程的源碼分析以及架構分析

Dubbo集羣容錯源碼解析

** 咱們後面的分析重點就是Cluster的調用過程**

Cluster的做用

Cluster 將 Directory 中的多個 Invoker 假裝成一個 Invoker，對上層透明，假裝過程包含了容錯邏輯，調用失敗後，重試另外一個

應對出錯狀況採起的策略,在某次出現錯誤後將會採用何種方式進行下次重試

集羣模式歸納

下面對這些一個一個分析

Failover Cluster

失敗自動切換，當出現失敗，重試其它服務器 [1]。一般用於讀操做，但重試會帶來更長延遲。可經過 retries="2" 來設置重試次數(不含第一次)。