1.簡介

為了避免單點故障碗降，現(xiàn)在的應用通常至少會部署在兩臺服務器上勾给。對于一些負載比較高的服務漱贱，會部署更多的服務器憔辫。這樣仇轻，在同一環(huán)境下的服務提供者數(shù)量會大于1帮坚。對于服務消費者來說晶乔，同一環(huán)境下出現(xiàn)了多個服務提供者较木。這時會出現(xiàn)一個問題辉川，服務消費者需要決定選擇哪個服務提供者進行調用表蝙。另外服務調用失敗時的處理措施也是需要考慮的，是重試呢乓旗，還是拋出異常府蛇，亦或是只打印異常等。為了處理這些問題屿愚，Dubbo 定義了集群接口 Cluster 以及 Cluster Invoker汇跨。集群 Cluster 用途是將多個服務提供者合并為一個 Cluster Invoker，并將這個 Invoker 暴露給服務消費者妆距。這樣一來穷遂，服務消費者只需通過這個 Invoker 進行遠程調用即可，至于具體調用哪個服務提供者娱据，以及調用失敗后如何處理等問題蚪黑，現(xiàn)在都交給集群模塊去處理。集群模塊是服務提供者和服務消費者的中間層，為服務消費者屏蔽了服務提供者的情況忌穿，這樣服務消費者就可以專心處理遠程調用相關事宜抒寂。比如發(fā)請求，接受服務提供者返回的數(shù)據(jù)等掠剑。這就是集群的作用屈芜。

Dubbo 提供了多種集群實現(xiàn)，包含但不限于 Failover Cluster朴译、Failfast Cluster 和 Failsafe Cluster 等井佑。每種集群實現(xiàn)類的用途不同，接下來會一一進行分析眠寿。

2. 集群容錯

在對集群相關代碼進行分析之前躬翁，這里有必要先來介紹一下集群容錯的所有組件。包含 Cluster盯拱、Cluster Invoker姆另、Directory、Router 和 LoadBalance 等坟乾。

img

集群工作過程可分為兩個階段：

• 第一個階段是在服務消費者初始化期間，集群 Cluster 實現(xiàn)類為服務消費者創(chuàng)建 Cluster Invoker 實例蝶防，即上圖中的 merge 操作甚侣。

• 第二個階段是在服務消費者進行遠程調用時。以 FailoverClusterInvoker 為例间学，該類型 Cluster Invoker 首先會調用 Directory 的 list 方法列舉 Invoker 列表（可將 Invoker 簡單理解為服務提供者）殷费。Directory 的用途是保存 Invoker，可簡單類比為 List<Invoker>低葫。其實現(xiàn)類 RegistryDirectory 是一個動態(tài)服務目錄详羡，可感知注冊中心配置的變化，它所持有的 Invoker 列表會隨著注冊中心內容的變化而變化嘿悬。每次變化后实柠，RegistryDirectory 會動態(tài)增刪 Invoker，并調用 Router 的 route 方法進行路由善涨，過濾掉不符合路由規(guī)則的 Invoker窒盐。當 FailoverClusterInvoker 拿到 Directory 返回的 Invoker 列表后，它會通過 LoadBalance 從 Invoker 列表中選擇一個 Invoker钢拧。最后 FailoverClusterInvoker 會將參數(shù)傳給 LoadBalance 選擇出的 Invoker 實例的 invoke 方法蟹漓，進行真正的遠程調用。

以上就是集群工作的整個流程源内，這里并沒介紹集群是如何容錯的葡粒。Dubbo 主要提供了這樣幾種容錯方式：

• Failover Cluster - 失敗自動切換，dubbo默認，常用于讀操作嗽交。
• Failfast Cluster - 一次調用失敗就立即失敗卿嘲，常見于非冪等性操作，比如新增操作轮纫。
• Failsafe Cluster - 出現(xiàn)異常時忽略腔寡，常用于日志記錄等不是很重要的接口調用。
• Failback Cluster - 失敗了在后臺自動記錄請求掌唾，然后定時重發(fā)放前。比較適合寫消息隊列等操作。
• Forking Cluster - 并行調用多個服務提供者糯彬，只要有一個成功立即返回凭语。適用于實時性要求比較高的讀操作，但是對資源比較浪費撩扒。
• broadcast cluster-逐個調用所有的provider似扔。任何一個provider出錯則報錯。常用于通知所有提供者更新緩存或日志等本地資源信息搓谆。

下面開始分析源碼炒辉。

3.源碼分析

3.1 Cluster 實現(xiàn)類分析

我們在上一章看到了兩個概念，分別是集群接口 Cluster 和 Cluster Invoker泉手，這兩者是不同的黔寇。Cluster 是接口，而 Cluster Invoker 是一種 Invoker斩萌。服務提供者的選擇邏輯缝裤，以及遠程調用失敗后的的處理邏輯均是封裝在 Cluster Invoker 中。那么 Cluster 接口和相關實現(xiàn)類有什么用呢颊郎？用途比較簡單憋飞，僅用于生成 Cluster Invoker。

首先我們先看一下繼承關系：

AbstractCluster.png

下面我們來看一下源碼姆吭。

@SPI(FailoverCluster.NAME)// Dubbo默認
public interface Cluster {

    /**
     * Merge the directory invokers to a virtual invoker.
     *
     * @param <T>
     * @param directory
     * @return cluster invoker
     * @throws RpcException
     */
    @Adaptive
    <T> Invoker<T> join(Directory<T> directory) throws RpcException;

}

public abstract class AbstractCluster implements Cluster {

    // 最終返回的對象是 InterceptorInvokerNode 榛做，該類是內部類，定義在下面  
    private <T> Invoker<T> buildClusterInterceptors(AbstractClusterInvoker<T> clusterInvoker, String key) {
        AbstractClusterInvoker<T> last = clusterInvoker;
        List<ClusterInterceptor> interceptors = ExtensionLoader.getExtensionLoader(ClusterInterceptor.class).getActivateExtension(clusterInvoker.getUrl(), key);

        if (!interceptors.isEmpty()) {
            for (int i = interceptors.size() - 1; i >= 0; i--) {
                final ClusterInterceptor interceptor = interceptors.get(i);
                final AbstractClusterInvoker<T> next = last;
                last = new InterceptorInvokerNode<>(clusterInvoker, interceptor, next);
            }
        }
        return last;
    }

    // Directory 指的是上節(jié)的服務字典（保存著privider的各種信息）
    @Override
    public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
        return buildClusterInterceptors(doJoin(directory), directory.getUrl().getParameter(REFERENCE_INTERCEPTOR_KEY));
    }

    //子類實現(xiàn)
    protected abstract <T> AbstractClusterInvoker<T> doJoin(Directory<T> directory) throws RpcException;

    //內部類
    protected class InterceptorInvokerNode<T> extends AbstractClusterInvoker<T> {

        private AbstractClusterInvoker<T> clusterInvoker;
        private ClusterInterceptor interceptor;
        private AbstractClusterInvoker<T> next;

        public InterceptorInvokerNode(AbstractClusterInvoker<T> clusterInvoker,
                                      ClusterInterceptor interceptor,
                                      AbstractClusterInvoker<T> next) {
            this.clusterInvoker = clusterInvoker;
            this.interceptor = interceptor;
            this.next = next;
        }

        @Override
        public Class<T> getInterface() {
            return clusterInvoker.getInterface();
        }

        @Override
        public URL getUrl() {
            return clusterInvoker.getUrl();
        }

        @Override
        public boolean isAvailable() {
            return clusterInvoker.isAvailable();
        }

        @Override
        public Result invoke(Invocation invocation) throws RpcException {
            Result asyncResult;
            try {
                // before方法
                interceptor.before(next, invocation);
                // 異步執(zhí)行
                asyncResult = interceptor.intercept(next, invocation);
            } catch (Exception e) {
                // onError callback
                if (interceptor instanceof ClusterInterceptor.Listener) {
                    ClusterInterceptor.Listener listener = (ClusterInterceptor.Listener) interceptor;
                    listener.onError(e, clusterInvoker, invocation);
                }
                throw e;
            } finally {
                // 最終會執(zhí)行 after 方法
                interceptor.after(next, invocation);
            }
            return asyncResult.whenCompleteWithContext((r, t) -> {
                // onResponse callback
                if (interceptor instanceof ClusterInterceptor.Listener) {
                    ClusterInterceptor.Listener listener = (ClusterInterceptor.Listener) interceptor;
                    if (t == null) {
                        listener.onMessage(r, clusterInvoker, invocation);
                    } else {
                        listener.onError(t, clusterInvoker, invocation);
                    }
                }
            });
        }

        @Override
        public void destroy() {
            clusterInvoker.destroy();
        }

        @Override
        public String toString() {
            return clusterInvoker.toString();
        }

        @Override
        protected Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
            // The only purpose is to build a interceptor chain, so the cluster related logic doesn't matter.
            return null;
        }
    }
}

如下猾编，F(xiàn)ailoverCluster 總共就包含這幾行代碼瘤睹，用于創(chuàng)建 FailoverClusterInvoker 對象，很簡單答倡。

public class FailoverCluster extends AbstractCluster {

    public final static String NAME = "failover";

    @Override
    public <T> AbstractClusterInvoker<T> doJoin(Directory<T> directory) throws RpcException {
        return new FailoverClusterInvoker<>(directory);
    }

}

3.2 Cluster Invoker 分析

我們首先從各種 Cluster Invoker 的父類 AbstractClusterInvoker 源碼開始說起轰传。前面說過，集群工作過程可分為兩個階段瘪撇，第一個階段是在服務消費者初始化期間获茬，這個在服務引用那篇文章中分析過港庄，就不贅述。第二個階段是在服務消費者進行遠程調用時恕曲，此時 AbstractClusterInvoker 的 invoke 方法會被調用鹏氧。列舉 Invoker，負載均衡等操作均會在此階段被執(zhí)行佩谣。因此下面先來看一下 invoke 方法的邏輯把还。

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

    // binding attachments into invocation.
    // 綁定 attachments 到 invocation 中
    Map<String, Object> contextAttachments = RpcContext.getContext().getObjectAttachments();
    if (contextAttachments != null && contextAttachments.size() != 0) {
        ((RpcInvocation) invocation).addObjectAttachments(contextAttachments);
    }

    // list可用的 invokers
    List<Invoker<T>> invokers = list(invocation);
    // 負載均衡選擇 通過 SPI 加載 Loadbalance，默認為 RandomLoadBalance
    LoadBalance loadbalance = initLoadBalance(invokers, invocation);
    RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
    //子類方法實現(xiàn)
    return doInvoke(invocation, invokers, loadbalance);
}

    // 子類實現(xiàn)
    protected abstract Result doInvoke(Invocation invocation, List<Invoker<T>> invokers,
                                       LoadBalance loadbalance) throws RpcException;

    // 調用了 Directory 的 list 方法
    protected List<Invoker<T>> list(Invocation invocation) throws RpcException {
        return directory.list(invocation);
    }

AbstractClusterInvoker 的 invoke 方法主要用于列舉 Invoker茸俭，以及加載 LoadBalance吊履。最后再調用模板方法 doInvoke 進行后續(xù)操作。

3.2.1 FailoverClusterInvoker

FailoverClusterInvoker 在調用失敗時调鬓，會自動切換 Invoker 進行重試艇炎。默認配置下，Dubbo 會使用這個類作為缺省 Cluster Invoker腾窝。下面來看一下該類的邏輯缀踪。

public class FailoverClusterInvoker<T> extends AbstractClusterInvoker<T> {

    public FailoverClusterInvoker(Directory<T> directory) {
        super(directory);
    }

    @Override
    @SuppressWarnings({"unchecked", "rawtypes"})
    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);
        // 獲取重試次數(shù)
        int len = getUrl().getMethodParameter(methodName, RETRIES_KEY, 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);
        // 循環(huán)調用，失敗重試
        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();
                // 在進行重試前重新列舉 Invoker虹脯，這樣做的好處是驴娃，如果某個服務掛了，
                // 通過調用 list 可得到最新可用的 Invoker 列表
                copyInvokers = list(invocation);
                // 對 copyinvokers 進行判空檢查
                checkInvokers(copyInvokers, invocation);
            }
             // 通過負載均衡選擇 Invoker
            Invoker<T> invoker = select(loadbalance, invocation, copyInvokers, invoked);
            // 添加到 invoker 到 invoked 列表中
            invoked.add(invoker);
            // 設置 invoked 到 RPC 上下文中
            RpcContext.getContext().setInvokers((List) invoked);
            try {
                // 調用目標 Invoker 的 invoke 方法
                Result result = invoker.invoke(invocation);
                if (le != null && logger.isWarnEnabled()) {
                    // 日志
                }
                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(..., "Failed to invoke the method ...");
    }

}

如上暇榴，F(xiàn)ailoverClusterInvoker 的 doInvoke 方法首先是獲取重試次數(shù)，然后根據(jù)重試次數(shù)進行循環(huán)調用蕉世，失敗后進行重試蔼紧。在 for 循環(huán)內，首先是通過負載均衡組件選擇一個 Invoker狠轻，然后再通過這個 Invoker 的 invoke 方法進行遠程調用奸例。如果失敗了，記錄下異常向楼，并進行重試查吊。重試時會再次調用父類的 list 方法列舉 Invoker。整個流程大致如此湖蜕，不是很難理解逻卖。下面我們看一下 select 方法的邏輯。

/**
 *
 *  使用loadbalance策略選擇一個調用程序昭抒。
 *  a)首先评也，使用loadbalance選擇一個調用程序炼杖。如果此調用程序在先前選擇的列表中，或者盗迟，如果此調用程序不可用坤邪，則繼續(xù)步驟b(重新選擇)，否則返回第一個選擇的調用程序
 *  b)重新選擇罚缕，用于重新選擇的驗證規(guī)則:selected > available艇纺。該規(guī)則保證所選的調用程序在之前選擇的列表中有最小的機會成為一個調用程序，并且保證該調用程序可用邮弹。
 *
 */
protected Invoker<T> select(LoadBalance loadbalance, Invocation invocation,
                            List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {

    if (CollectionUtils.isEmpty(invokers)) {
        return null;
    }
    // 獲取調用方法名
    String methodName = invocation == null ? StringUtils.EMPTY_STRING : invocation.getMethodName();

    // 獲取 sticky 配置黔衡，sticky 表示粘滯連接。所謂粘滯連接是指讓服務消費者盡可能的
    // 調用同一個服務提供者肠鲫，除非該提供者掛了再進行切換
    boolean sticky = invokers.get(0).getUrl()
            .getMethodParameter(methodName, CLUSTER_STICKY_KEY, DEFAULT_CLUSTER_STICKY);

    //ignore overloaded method
    // 檢測 invokers 列表是否包含 stickyInvoker员帮，如果不包含，
    // 說明 stickyInvoker 代表的服務提供者掛了导饲，此時需要將其置空
    if (stickyInvoker != null && !invokers.contains(stickyInvoker)) {
        stickyInvoker = null;
    }
    //ignore concurrency problem
    // 在 sticky 為 true捞高，且 stickyInvoker != null 的情況下。如果 selected 包含 
    // stickyInvoker渣锦，表明 stickyInvoker 對應的服務提供者可能因網(wǎng)絡原因未能成功提供服務硝岗。
    // 但是該提供者并沒掛，此時 invokers 列表中仍存在該服務提供者對應的 Invoker袋毙。
    if (sticky && stickyInvoker != null && (selected == null || !selected.contains(stickyInvoker))) {
         // availablecheck 表示是否開啟了可用性檢查型檀，如果開啟了，則調用 stickyInvoker 的 
         // isAvailable 方法進行檢查听盖，如果檢查通過胀溺，則直接返回 stickyInvoker。
        if (availablecheck && stickyInvoker.isAvailable()) {
            return stickyInvoker;
        }
    }

    // 如果線程走到當前代碼處皆看，說明前面的 stickyInvoker 為空仓坞，或者不可用。
    // 此時繼續(xù)調用 doSelect 選擇 Invoker
    Invoker<T> invoker = doSelect(loadbalance, invocation, invokers, selected);

    // 如果 sticky 為 true腰吟，則將負載均衡組件選出的 Invoker 賦值給 stickyInvoker
    if (sticky) {
        stickyInvoker = invoker;
    }
    return invoker;
}

如上无埃，select 方法的主要邏輯集中在了對粘滯連接特性的支持上。首先是獲取 sticky 配置毛雇，然后再檢測 invokers 列表中是否包含 stickyInvoker嫉称，如果不包含，則認為該 stickyInvoker 不可用灵疮，此時將其置空织阅。這里的 invokers 列表可以看做是存活著的服務提供者列表，如果這個列表不包含 stickyInvoker震捣，那自然而然的認為 stickyInvoker 掛了蒲稳，所以置空氮趋。如果 stickyInvoker 存在于 invokers 列表中，此時要進行下一項檢測 — 檢測 selected 中是否包含 stickyInvoker江耀。如果包含的話剩胁，說明 stickyInvoker 在此之前沒有成功提供服務（但其仍然處于存活狀態(tài)）。此時我們認為這個服務不可靠祥国，不應該在重試期間內再次被調用昵观，因此這個時候不會返回該 stickyInvoker。如果 selected 不包含 stickyInvoker舌稀，此時還需要進行可用性檢測啊犬，比如檢測服務提供者網(wǎng)絡連通性等。當可用性檢測通過壁查，才可返回 stickyInvoker觉至，否則調用 doSelect 方法選擇 Invoker。如果 sticky 為 true睡腿，此時會將 doSelect 方法選出的 Invoker 賦值給 stickyInvoker语御。

以上就是 select 方法的邏輯，這段邏輯看起來不是很復雜席怪，但是信息量比較大应闯。不搞懂 invokers 和 selected 兩個入?yún)⒌暮x，以及粘滯連接特性挂捻，這段代碼是不容易看懂的碉纺。所以大家在閱讀這段代碼時，不要忽略了對背景知識的理解刻撒。關于 select 方法先分析這么多骨田，繼續(xù)向下分析。

private Invoker<T> doSelect(LoadBalance loadbalance, Invocation invocation,
                            List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {

    if (CollectionUtils.isEmpty(invokers)) {
        return null;
    }
    if (invokers.size() == 1) {
        return invokers.get(0);
    }
    // 通過負載均衡組件選擇 Invoker
    Invoker<T> invoker = loadbalance.select(invokers, getUrl(), invocation);

    //If the `invoker` is in the  `selected` or invoker is unavailable && availablecheck is true, reselect.
    // 如果 selected 包含負載均衡選擇出的 Invoker声怔，或者該 Invoker 無法經(jīng)過可用性檢查盛撑，此時進行重選
    if ((selected != null && selected.contains(invoker))
            || (!invoker.isAvailable() && getUrl() != null && availablecheck)) {
        try {
             // 進行重選
            Invoker<T> rInvoker = reselect(loadbalance, invocation, invokers, selected, availablecheck);
            if (rInvoker != null) {
                // 如果 rinvoker 不為空，則將其賦值給 invoker
                invoker = rInvoker;
            } else {
                //Check the index of current selected invoker, if it's not the last one, choose the one at index+1.
                // rinvoker 為空捧搞，定位 invoker 在 invokers 中的位置
                int index = invokers.indexOf(invoker);
                try {
                    //Avoid collision
                    // 獲取 index + 1 位置處的 Invoker，以下代碼等價于：
                    //     invoker = invokers.get((index + 1) % invokers.size());
                    invoker = invokers.get((index + 1) % invokers.size());
                } 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;
}

doSelect 主要做了兩件事狮荔，第一是通過負載均衡組件選擇 Invoker胎撇。第二是，如果選出來的 Invoker 不穩(wěn)定殖氏，或不可用晚树，此時需要調用 reselect 方法進行重選。若 reselect 選出來的 Invoker 為空雅采，此時定位 invoker 在 invokers 列表中的位置 index爵憎，然后獲取 index + 1 處的 invoker慨亲，這也可以看做是重選邏輯的一部分。下面我們來看一下 reselect 方法的邏輯宝鼓。

    private Invoker<T> reselect(LoadBalance loadbalance, Invocation invocation,
                            List<Invoker<T>> invokers, List<Invoker<T>> selected, boolean availablecheck) throws RpcException {

    //Allocating one in advance, this list is certain to be used.
    List<Invoker<T>> reselectInvokers = new ArrayList<>(
            invokers.size() > 1 ? (invokers.size() - 1) : invokers.size());

    // First, try picking a invoker not in `selected`.
    // 遍歷 invokers 列表
    for (Invoker<T> invoker : invokers) {
        if (availablecheck && !invoker.isAvailable()) {
            continue;
        }

        // 如果 selected 列表不包含當前 invoker刑棵，則將其添加到 reselectInvokers 中
        if (selected == null || !selected.contains(invoker)) {
            reselectInvokers.add(invoker);
        }
    }

    // reselectInvokers 不為空，此時通過負載均衡組件進行選擇
    if (!reselectInvokers.isEmpty()) {
        return loadbalance.select(reselectInvokers, getUrl(), invocation);
    }

    // Just pick an available invoker using loadbalance policy
    if (selected != null) {
        for (Invoker<T> invoker : selected) {
            // 如果 selected 列表不包含當前 invoker愚铡，則將其添加到 reselectInvokers 中
            if ((invoker.isAvailable()) // available first
                    && !reselectInvokers.contains(invoker)) {
                reselectInvokers.add(invoker);
            }
        }
    }
    if (!reselectInvokers.isEmpty()) {
        // 通過負載均衡組件進行選擇
        return loadbalance.select(reselectInvokers, getUrl(), invocation);
    }

    return null;
}

reselect 方法總結下來其實只做了兩件事情蛉签，第一是查找可用的 Invoker，并將其添加到 reselectInvokers 集合中沥寥。第二碍舍，如果 reselectInvokers 不為空，則通過負載均衡組件再次進行選擇邑雅。其中第一件事情又可進行細分片橡，一開始，reselect 從 invokers 列表中查找有效可用的 Invoker淮野，若未能找到捧书，此時再到 selected 列表中繼續(xù)查找。

3.2.2 FailbackClusterInvoker

FailbackClusterInvoker 會在調用失敗后录煤，返回一個空結果給服務消費者鳄厌。并通過定時任務對失敗的調用進行重傳，適合執(zhí)行消息通知等操作妈踊。

public class FailbackClusterInvoker<T> extends AbstractClusterInvoker<T> {

    private static final Logger logger = LoggerFactory.getLogger(FailbackClusterInvoker.class);

    private static final long RETRY_FAILED_PERIOD = 5;

    private final int retries;

    private final int failbackTasks;

     // Timer類 
    private volatile Timer failTimer;

    ...

    private void addFailed(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, Invoker<T> lastInvoker) {
        if (failTimer == null) {
            synchronized (this) {
                if (failTimer == null) {
                    failTimer = new HashedWheelTimer(
                            new NamedThreadFactory("failback-cluster-timer", true),
                            1,
                            TimeUnit.SECONDS, 32, failbackTasks);
                }
            }
        }
        RetryTimerTask retryTimerTask = new RetryTimerTask(loadbalance, invocation, invokers, lastInvoker, retries, RETRY_FAILED_PERIOD);
        try {
            // 默認 5 秒執(zhí)行一次
            failTimer.newTimeout(retryTimerTask, RETRY_FAILED_PERIOD, TimeUnit.SECONDS);
        } catch (Throwable e) {
            ...
        }
    }

    @Override
    protected Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
        Invoker<T> invoker = null;
        try {
            checkInvokers(invokers, invocation);
            invoker = select(loadbalance, invocation, invokers, null);
            return invoker.invoke(invocation);
        } catch (Throwable e) {
            //失敗后了嚎，加入一個任務，用于定時重試
            addFailed(loadbalance, invocation, invokers, invoker);
            return AsyncRpcResult.newDefaultAsyncResult(null, null, invocation); // ignore
        }
    }

    @Override
    public void destroy() {
        super.destroy();
        if (failTimer != null) {
            failTimer.stop();
        }
    }

    /**
     * RetryTimerTask 內部類
     */
    private class RetryTimerTask implements TimerTask {
        private final Invocation invocation;
        private final LoadBalance loadbalance;
        private final List<Invoker<T>> invokers;
        private final int retries;
        private final long tick;
        private Invoker<T> lastInvoker;
        private int retryTimes = 0;

        RetryTimerTask(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, Invoker<T> lastInvoker, int retries, long tick) {
            this.loadbalance = loadbalance;
            this.invocation = invocation;
            this.invokers = invokers;
            this.retries = retries;
            this.tick = tick;
            this.lastInvoker=lastInvoker;
        }

        @Override
        public void run(Timeout timeout) {
            try {
                //重新選擇廊营，重新調用
                Invoker<T> retryInvoker = select(loadbalance, invocation, invokers, Collections.singletonList(lastInvoker));
                lastInvoker = retryInvoker;
                retryInvoker.invoke(invocation);
            } catch (Throwable e) {
                // log
                if ((++retryTimes) >= retries) {
                   // log
                } else {
                    //重新放入執(zhí)行
                    rePut(timeout);
                }
            }
        }

        private void rePut(Timeout timeout) {
            if (timeout == null) {
                return;
            }

            Timer timer = timeout.timer();
            if (timer.isStop() || timeout.isCancelled()) {
                return;
            }

            timer.newTimeout(timeout.task(), tick, TimeUnit.SECONDS);
        }
    }
}

FailbackClusterInvoker 利用一個 Timer 去執(zhí)行重試操作歪泳，邏輯沒什么復雜的。其他的 invoker 不再分析了露筒。

4.總結

本篇文章詳細分析了集群容錯的幾種實現(xiàn)方式呐伞。集群容錯對于 Dubbo 框架來說，是很重要的邏輯慎式。集群模塊處于服務提供者和消費者之間伶氢，對于服務消費者來說，集群可向其屏蔽服務提供者集群的情況瘪吏，使其能夠專心進行遠程調用癣防。除此之外，通過集群模塊掌眠，我們還可以對服務之間的調用鏈路進行編排優(yōu)化蕾盯，治理服務±侗總的來說级遭，對于 Dubbo 而言望拖，集群容錯相關邏輯是非常重要的。想要對 Dubbo 有比較深的理解挫鸽，集群容錯是必須要掌握的说敏。

5.參考資料

本文參考于Dubbo官網(wǎng)，詳情以官網(wǎng)最新文檔為準掠兄。