摘要：集群用途是將多個(gè)服務(wù)提供者合并為一個(gè)，并將這個(gè)暴露給服務(wù)消費(fèi)者。比如發(fā)請(qǐng)求，接受服務(wù)提供者返回的數(shù)據(jù)等。如果包含，表明對(duì)應(yīng)的服務(wù)提供者可能因網(wǎng)絡(luò)原因未能成功提供服務(wù)。如果不包含，此時(shí)還需要進(jìn)行可用性檢測(cè)，比如檢測(cè)服務(wù)提供者網(wǎng)絡(luò)連通性等。

為了避免單點(diǎn)故障，現(xiàn)在的應(yīng)用至少會(huì)部署在兩臺(tái)服務(wù)器上。對(duì)于一些負(fù)載比較高的服務(wù)，會(huì)部署更多臺(tái)服務(wù)器。這樣，同一環(huán)境下的服務(wù)提供者數(shù)量會(huì)大于1。對(duì)于服務(wù)消費(fèi)者來(lái)說(shuō)，同一環(huán)境下出現(xiàn)了多個(gè)服務(wù)提供者。這時(shí)會(huì)出現(xiàn)一個(gè)問(wèn)題，服務(wù)消費(fèi)者需要決定選擇哪個(gè)服務(wù)提供者進(jìn)行調(diào)用。另外服務(wù)調(diào)用失敗時(shí)的處理措施也是需要考慮的，是重試呢，還是拋出異常，亦或是只打印異常等。為了處理這些問(wèn)題，Dubbo 定義了集群接口 Cluster 以及及 Cluster Invoker。集群 Cluster 用途是將多個(gè)服務(wù)提供者合并為一個(gè) Cluster Invoker，并將這個(gè) Invoker 暴露給服務(wù)消費(fèi)者。這樣一來(lái)，服務(wù)消費(fèi)者只需通過(guò)這個(gè) Invoker 進(jìn)行遠(yuǎn)程調(diào)用即可，至于具體調(diào)用哪個(gè)服務(wù)提供者，以及調(diào)用失敗后如何處理等問(wèn)題，現(xiàn)在都交給集群模塊去處理。集群模塊是服務(wù)提供者和服務(wù)消費(fèi)者的中間層，為服務(wù)消費(fèi)者屏蔽了服務(wù)提供者的情況，這樣服務(wù)消費(fèi)者就可以處理遠(yuǎn)程調(diào)用相關(guān)事宜。比如發(fā)請(qǐng)求，接受服務(wù)提供者返回的數(shù)據(jù)等。這就是集群的作用。

Dubbo 提供了多種集群實(shí)現(xiàn)，包含但不限于 Failover Cluster、Failfast Cluster 和 Failsafe Cluster 等。每種集群實(shí)現(xiàn)類的用途不同，接下來(lái)我會(huì)一一進(jìn)行分析。

在對(duì)集群相關(guān)代碼進(jìn)行分析之前，這里有必要先來(lái)介紹一下集群容錯(cuò)的所有組件。包含 Cluster、Cluster Invoker、Directory、Router 和 LoadBalance 等，先來(lái)看圖。

* 圖片來(lái)源：Dubbo 官方文檔

這張圖來(lái)自 Dubbo 官方文檔，接下來(lái)我會(huì)按照這張圖介紹集群工作過(guò)程。集群工作過(guò)程可分為兩個(gè)階段，第一個(gè)階段是在服務(wù)消費(fèi)者初始化期間，集群 Cluster 實(shí)現(xiàn)類為服務(wù)消費(fèi)者創(chuàng)建 Cluster Invoker 實(shí)例，即上圖中的 merge 操作。第二個(gè)階段是在服務(wù)消費(fèi)者進(jìn)行遠(yuǎn)程調(diào)用時(shí)。以 FailoverClusterInvoker 為例，該類型 Cluster Invoker 首先會(huì)調(diào)用 Directory 的 list 方法列舉 Invoker 列表（可將 Invoker 簡(jiǎn)單理解為服務(wù)提供者）。Directory 的用途是保存 Invoker，可簡(jiǎn)單類比為 List。其實(shí)現(xiàn)類 RegistryDirectory 是一個(gè)動(dòng)態(tài)服務(wù)目錄，可感知注冊(cè)中心配置的變化，它所持有的 Inovker 列表會(huì)隨著注冊(cè)中心內(nèi)容的變化而變化。每次變化后，RegistryDirectory 會(huì)動(dòng)態(tài)增刪 Inovker，并調(diào)用 Router 的 route 方法進(jìn)行路由，過(guò)濾掉不符合路由規(guī)則的 Invoker?；氐缴蠄D，Cluster Invoker 實(shí)際上并不會(huì)直接調(diào)用 Router 進(jìn)行路由。當(dāng) FailoverClusterInvoker 拿到 Directory 返回的 Invoker 列表后，它會(huì)通過(guò) LoadBalance 從 Invoker 列表中選擇一個(gè) Inovker。最后 FailoverClusterInvoker 會(huì)將參數(shù)傳給 LoadBalance 選擇出的 Invoker 實(shí)例的 invoker 方法，進(jìn)行真正的 RPC 調(diào)用。

以上就是集群工作的整個(gè)流程，這里并沒(méi)介紹集群是如何容錯(cuò)的。Dubbo 主要提供了這樣幾種容錯(cuò)方式：

Failover Cluster - 失敗自動(dòng)切換

Failfast Cluster - 快速失敗

Failsafe Cluster - 失敗安全

Failback Cluster - 失敗自動(dòng)恢復(fù)

Forking Cluster - 并行調(diào)用多個(gè)服務(wù)提供者

這里暫時(shí)只對(duì)這幾種容錯(cuò)模式進(jìn)行簡(jiǎn)單的介紹，在接下來(lái)的章節(jié)中，我會(huì)重點(diǎn)分析這幾種容錯(cuò)模式的具體實(shí)現(xiàn)。好了，關(guān)于集群的工作流程和容錯(cuò)模式先說(shuō)到這，接下來(lái)進(jìn)入源碼分析階段。

我在上一章提到了集群接口 Cluster 和 Cluster Invoker，這兩者是不同的。Cluster 是接口，而 Cluster Invoker 是一種 Invoker。服務(wù)提供者的選擇邏輯，以及遠(yuǎn)程調(diào)用失敗后的的處理邏輯均是封裝在 Cluster Invoker 中。那么 Cluster 接口和相關(guān)實(shí)現(xiàn)類有什么用呢？用途比較簡(jiǎn)單，用于生成 Cluster Invoker，僅此而已。下面我們來(lái)看一下源碼。

public class FailoverCluster implements Cluster {

    public final static String NAME = "failover";

    @Override
    public  Invoker join(Directory directory) throws RpcException {
        // 創(chuàng)建并返回 FailoverClusterInvoker 對(duì)象
        return new FailoverClusterInvoker(directory);
    }
}

如上，F(xiàn)ailoverCluster 總共就包含這幾行代碼，用于創(chuàng)建 FailoverClusterInvoker 對(duì)象，很簡(jiǎn)單。下面再看一個(gè)。

public class FailbackCluster implements Cluster {

    public final static String NAME = "failback";

    @Override
    public  Invoker join(Directory directory) throws RpcException {
        // 創(chuàng)建并返回 FailbackClusterInvoker 對(duì)象
        return new FailbackClusterInvoker(directory);
    }

}

如上，F(xiàn)ailbackCluster 的邏輯也是很簡(jiǎn)單，無(wú)需解釋了。所以接下來(lái)，我們把重點(diǎn)放在各種 Cluster Invoker 上

我們首先從各種 Cluster Invoker 的父類 AbstractClusterInvoker 源碼開(kāi)始說(shuō)起。前面說(shuō)過(guò)，集群工作過(guò)程可分為兩個(gè)階段，第一個(gè)階段是在服務(wù)消費(fèi)者初始化期間，這個(gè)在服務(wù)引用那篇文章中已經(jīng)分析過(guò)了，這里不再贅述。第二個(gè)階段是在服務(wù)消費(fèi)者進(jìn)行遠(yuǎn)程調(diào)用時(shí)，此時(shí) AbstractClusterInvoker 的 invoke 方法會(huì)被調(diào)用。列舉 Invoker，負(fù)載均衡等操作均會(huì)在此階段被執(zhí)行。因此下面先來(lái)看一下 invoke 方法的邏輯。

public Result invoke(final Invocation invocation) throws RpcException {
    checkWhetherDestroyed();
    LoadBalance loadbalance = null;

    // 綁定 attachments 到 invocation 中.
    Map contextAttachments = RpcContext.getContext().getAttachments();
    if (contextAttachments != null && contextAttachments.size() != 0) {
        ((RpcInvocation) invocation).addAttachments(contextAttachments);
    }

    // 列舉 Invoker
    List> invokers = list(invocation);
    if (invokers != null && !invokers.isEmpty()) {
        // 加載 LoadBalance
        loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(invokers.get(0).getUrl()
                .getMethodParameter(RpcUtils.getMethodName(invocation), Constants.LOADBALANCE_KEY, Constants.DEFAULT_LOADBALANCE));
    }
    RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
    
    // 調(diào)用 doInvoke 進(jìn)行后續(xù)操作
    return doInvoke(invocation, invokers, loadbalance);
}

// 抽象方法，由子類實(shí)現(xiàn)
protected abstract Result doInvoke(Invocation invocation, List> invokers,
                                       LoadBalance loadbalance) throws RpcException;

AbstractClusterInvoker 的 invoke 方法主要用于列舉 Invoker，以及加載 LoadBalance。最后再調(diào)用模板方法 doInvoke 進(jìn)行后續(xù)操作。下面我們來(lái)看一下 Invoker 列舉方法 list(Invocation) 的邏輯，如下：

protected List> list(Invocation invocation) throws RpcException {
    // 調(diào)用 Directory 的 list 方法
    List> invokers = directory.list(invocation);
    return invokers;
}

如上，AbstractClusterInvoker 中的 list 方法做的事情很簡(jiǎn)單，只是簡(jiǎn)單的調(diào)用了 Directory 的 list 方法，沒(méi)有其他更多的邏輯了。Directory 的 list 方法我在前面的文章中已經(jīng)分析過(guò)了，這里就不贅述了。

接下來(lái)，我們把目光轉(zhuǎn)移到 AbstractClusterInvoker 的各種實(shí)現(xiàn)類上，來(lái)看一下這些實(shí)現(xiàn)類是如何實(shí)現(xiàn) doInvoke 方法邏輯的。

FailoverClusterInvoker 在調(diào)用失敗時(shí)，會(huì)自動(dòng)切換 Invoker 進(jìn)行重試。在無(wú)明確配置下，Dubbo 會(huì)使用這個(gè)類作為缺省 Cluster Invoker。下面來(lái)看一下該類的邏輯。

public class FailoverClusterInvoker extends AbstractClusterInvoker {

    // 省略部分代碼

    @Override
    public Result doInvoke(Invocation invocation, final List> invokers, LoadBalance loadbalance) throws RpcException {
        List> copyinvokers = invokers;
        checkInvokers(copyinvokers, invocation);
        // 獲取重試次數(shù)
        int len = getUrl().getMethodParameter(invocation.getMethodName(), Constants.RETRIES_KEY, Constants.DEFAULT_RETRIES) + 1;
        if (len <= 0) {
            len = 1;
        }
        RpcException le = null;
        List> invoked = new ArrayList>(copyinvokers.size());
        Set providers = new HashSet(len);
        // 循環(huán)調(diào)用，失敗重試
        for (int i = 0; i < len; i++) {
            if (i > 0) {
                checkWhetherDestroyed();
                // 在進(jìn)行重試前重新列舉 Invoker，這樣做的好處是，如果某個(gè)服務(wù)掛了，
                // 通過(guò)調(diào)用 list 可得到最新可用的 Invoker 列表
                copyinvokers = list(invocation);
                // 對(duì) copyinvokers 進(jìn)行判空檢查
                checkInvokers(copyinvokers, invocation);
            }

            // 通過(guò)負(fù)載均衡選擇 Invoker
            Invoker invoker = select(loadbalance, invocation, copyinvokers, invoked);
            // 添加到 invoker 到 invoked 列表中
            invoked.add(invoker);
            // 設(shè)置 invoked 到 RPC 上下文中
            RpcContext.getContext().setInvokers((List) invoked);
            try {
                // 調(diào)用目標(biāo) Invoker 的 invoke 方法
                Result result = invoker.invoke(invocation);
                return result;
            } catch (RpcException e) {
                if (e.isBiz()) {
                    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ù)進(jìn)行循環(huán)調(diào)用，失敗后進(jìn)行重試。在 for 循環(huán)內(nèi)，首先是通過(guò)負(fù)載均衡組件選擇一個(gè) Invoker，然后再通過(guò)這個(gè) Invoker 的 invoke 方法進(jìn)行遠(yuǎn)程調(diào)用。如果失敗了，記錄下異常，并進(jìn)行重試。重試時(shí)會(huì)再次調(diào)用父類的 list 方法列舉 Invoker。整個(gè)流程大致如此，不是很難理解。下面我們看一下 select 方法的邏輯。

protected Invoker select(LoadBalance loadbalance, Invocation invocation, List> invokers, List> selected) throws RpcException {
    if (invokers == null || invokers.isEmpty())
        return null;
    // 獲取調(diào)用方法名
    String methodName = invocation == null ? "" : invocation.getMethodName();

    // 獲取 sticky 配置，sticky 表示粘滯連接。所謂粘滯連接是指讓服務(wù)消費(fèi)者盡可能的
    // 調(diào)用同一個(gè)服務(wù)提供者，除非該提供者掛了再進(jìn)行切換
    boolean sticky = invokers.get(0).getUrl().getMethodParameter(methodName, Constants.CLUSTER_STICKY_KEY, Constants.DEFAULT_CLUSTER_STICKY);
    {
        // 檢測(cè) invokers 列表是否包含 stickyInvoker，如果不包含，
        // 說(shuō)明 stickyInvoker 代表的服務(wù)提供者掛了，此時(shí)需要將其置空
        if (stickyInvoker != null && !invokers.contains(stickyInvoker)) {
            stickyInvoker = null;
        }
        
        // 在 sticky 為 true，且 stickyInvoker != null 的情況下。如果 selected 包含 
        // stickyInvoker，表明 stickyInvoker 對(duì)應(yīng)的服務(wù)提供者可能因網(wǎng)絡(luò)原因未能成功提供服務(wù)。
        // 但是該提供者并沒(méi)掛，此時(shí) invokers 列表中仍存在該服務(wù)提供者對(duì)應(yīng)的 Invoker。
        if (sticky && stickyInvoker != null && (selected == null || !selected.contains(stickyInvoker))) {
            // availablecheck 表示是否開(kāi)啟了可用性檢查，如果開(kāi)啟了，則調(diào)用 stickyInvoker 的 
            // isAvailable 方法進(jìn)行檢查，如果檢查通過(guò)，則直接返回 stickyInvoker。
            if (availablecheck && stickyInvoker.isAvailable()) {
                return stickyInvoker;
            }
        }
    }
    
    // 如果線程走到當(dāng)前代碼處，說(shuō)明前面的 stickyInvoker 為空，或者不可用。
    // 此時(shí)調(diào)用繼續(xù)調(diào)用 doSelect 選擇 Invoker
    Invoker invoker = doSelect(loadbalance, invocation, invokers, selected);

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

如上，select 方法的主要邏輯集中在了對(duì)粘滯連接特性的支持上。首先是獲取 sticky 配置，然后再檢測(cè) invokers 列表中是否包含 stickyInvoker，如果不包含，則認(rèn)為該 stickyInvoker 不可用，此時(shí)將其置空。這里的 invokers 列表可以看做是存活著的服務(wù)提供者列表，如果這個(gè)列表不包含 stickyInvoker，那自然而然的認(rèn)為 stickyInvoker 掛了，所以置空。如果 stickyInvoker 存在于 invokers 列表中，此時(shí)要進(jìn)行下一項(xiàng)檢測(cè) ---- 檢測(cè) selected 中是否包含 stickyInvoker。如果包含的話，說(shuō)明 stickyInvoker 在此之前沒(méi)有成功提供服務(wù)（但其仍然處于存活狀態(tài)）。此時(shí)我們認(rèn)為這個(gè)服務(wù)不可靠，不應(yīng)該在重試期間內(nèi)再次被調(diào)用，因此這個(gè)時(shí)候不會(huì)返回該 stickyInvoker。如果 selected 不包含 stickyInvoker，此時(shí)還需要進(jìn)行可用性檢測(cè)，比如檢測(cè)服務(wù)提供者網(wǎng)絡(luò)連通性等。當(dāng)可用性檢測(cè)通過(guò)，才可返回 stickyInvoker，否則調(diào)用 doSelect 方法選擇 Invoker。如果 sticky 為 true，此時(shí)會(huì)將 doSelect 方法選出的 Invoker 賦值給 stickyInvoker。

以上就是 select 方法的邏輯，這段邏輯看起來(lái)不是很復(fù)雜，但是信息量比較大。不搞懂 invokers 和 selected 兩個(gè)入?yún)⒌暮x，以及粘滯連接特性，這段代碼應(yīng)該是沒(méi)法看懂的。大家在閱讀這段代碼時(shí)，不要忽略了對(duì)背景知識(shí)的理解。其他的不多說(shuō)了，繼續(xù)向下分析。

private Invoker doSelect(LoadBalance loadbalance, Invocation invocation, List> invokers, List> selected) throws RpcException {
    if (invokers == null || invokers.isEmpty())
        return null;
    if (invokers.size() == 1)
        return invokers.get(0);
    if (loadbalance == null) {
        // 如果 loadbalance 為空，這里通過(guò) SPI 加載 Loadbalance，默認(rèn)為 RandomLoadBalance
        loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(Constants.DEFAULT_LOADBALANCE);
    }
    
    // 通過(guò)負(fù)載均衡組件選擇 Invoker
    Invoker invoker = loadbalance.select(invokers, getUrl(), invocation);

    // 如果 selected 包含負(fù)載均衡選擇出的 Invoker，或者該 Invoker 無(wú)法經(jīng)過(guò)可用性檢查，此時(shí)進(jìn)行重選
    if ((selected != null && selected.contains(invoker))
            || (!invoker.isAvailable() && getUrl() != null && availablecheck)) {
        try {
            // 進(jìn)行重選
            Invoker rinvoker = reselect(loadbalance, invocation, invokers, selected, availablecheck);
            if (rinvoker != null) {
                // 如果 rinvoker 不為空，則將其賦值給 invoker
                invoker = rinvoker;
            } else {
                // rinvoker 為空，定位 invoker 在 invokers 中的位置
                int index = invokers.indexOf(invoker);
                try {
                    // 獲取 index + 1 位置處的 Invoker，以下代碼等價(jià)于：
                    //     invoker = invokers.get((index + 1) % invokers.size());
                    invoker = index < invokers.size() - 1 ? invokers.get(index + 1) : invokers.get(0);
                } catch (Exception e) {
                    logger.warn("... may because invokers list dynamic change, ignore.");
                }
            }
        } catch (Throwable t) {
            logger.error("cluster reselect fail reason is : ...");
        }
    }
    return invoker;
}

doSelect 主要做了兩件事，第一是通過(guò)負(fù)載均衡組件選擇 Invoker。第二是，如果選出來(lái)的 Invoker 不穩(wěn)定，或不可用，此時(shí)需要調(diào)用 reselect 方法進(jìn)行重選。若 reselect 選出來(lái)的 Invoker 為空，此時(shí)定位 invoker 在 invokers 列表中的位置 index，然后獲取 index + 1 處的 invoker，這也可以看做是重選邏輯的一部分。關(guān)于負(fù)載均衡的選擇邏輯，我將會(huì)在下篇文章進(jìn)行詳細(xì)分析。下面我們來(lái)看一下 reselect 方法的邏輯。

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

    List> reselectInvokers = new ArrayList>(invokers.size() > 1 ? (invokers.size() - 1) : invokers.size());

    // 根據(jù) availablecheck 進(jìn)行不同的處理
    if (availablecheck) {
        // 遍歷 invokers 列表
        for (Invoker invoker : invokers) {
            // 檢測(cè)可用性
            if (invoker.isAvailable()) {
                // 如果 selected 列表不包含當(dāng)前 invoker，則將其添加到 reselectInvokers 中
                if (selected == null || !selected.contains(invoker)) {
                    reselectInvokers.add(invoker);
                }
            }
        }
        
        // reselectInvokers 不為空，此時(shí)通過(guò)負(fù)載均衡組件進(jìn)行選擇
        if (!reselectInvokers.isEmpty()) {
            return loadbalance.select(reselectInvokers, getUrl(), invocation);
        }

    // 不檢查 Invoker 可用性
    } else {
        for (Invoker invoker : invokers) {
            // 如果 selected 列表不包含當(dāng)前 invoker，則將其添加到 reselectInvokers 中
            if (selected == null || !selected.contains(invoker)) {
                reselectInvokers.add(invoker);
            }
        }
        if (!reselectInvokers.isEmpty()) {
            // 通過(guò)負(fù)載均衡組件進(jìn)行選擇
            return loadbalance.select(reselectInvokers, getUrl(), invocation);
        }
    }

    {
        // 若線程走到此處，說(shuō)明 reselectInvokers 集合為空，此時(shí)不會(huì)調(diào)用負(fù)載均衡組件進(jìn)行篩選。
        // 這里從 selected 列表中查找可用的 Invoker，并將其添加到 reselectInvokers 集合中
        if (selected != null) {
            for (Invoker invoker : selected) {
                if ((invoker.isAvailable())
                        && !reselectInvokers.contains(invoker)) {
                    reselectInvokers.add(invoker);
                }
            }
        }
        if (!reselectInvokers.isEmpty()) {
            // 再次進(jìn)行選擇，并返回選擇結(jié)果
            return loadbalance.select(reselectInvokers, getUrl(), invocation);
        }
    }
    return null;
}

reselect 方法總結(jié)下來(lái)其實(shí)只做了兩件事情，第一是查找可用的 Invoker，并將其添加到 reselectInvokers 集合中。第二，如果 reselectInvokers 不為空，則通過(guò)負(fù)載均衡組件再次進(jìn)行選擇。其中第一件事情又可進(jìn)行細(xì)分，一開(kāi)始，reselect 從 invokers 列表中查找有效可用的 Invoker，若未能找到，此時(shí)再到 selected 列表中繼續(xù)查找。關(guān)于 reselect 方法就先分析到這，繼續(xù)分析其他的 Cluster Invoker。

FailbackClusterInvoker 會(huì)在調(diào)用失敗后，返回一個(gè)空結(jié)果給服務(wù)提供者。并通過(guò)定時(shí)任務(wù)對(duì)失敗的調(diào)用進(jìn)行重傳，適合執(zhí)行消息通知等操作。下面來(lái)看一下它的實(shí)現(xiàn)邏輯。

public class FailbackClusterInvoker extends AbstractClusterInvoker {

    private static final long RETRY_FAILED_PERIOD = 5 * 1000;

    private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(2,
            new NamedInternalThreadFactory("failback-cluster-timer", true));

    private final ConcurrentMap> failed = new ConcurrentHashMap>();
    private volatile ScheduledFuture retryFuture;

    @Override
    protected Result doInvoke(Invocation invocation, List> invokers, LoadBalance loadbalance) throws RpcException {
        try {
            checkInvokers(invokers, invocation);
            // 選擇 Invoker
            Invoker invoker = select(loadbalance, invocation, invokers, null);
            // 進(jìn)行調(diào)用
            return invoker.invoke(invocation);
        } catch (Throwable e) {
            // 如果調(diào)用過(guò)程中發(fā)生異常，此時(shí)僅打印錯(cuò)誤日志，不拋出異常
            logger.error("Failback to invoke method ...");
            
            // 記錄調(diào)用信息
            addFailed(invocation, this);
            // 返回一個(gè)空結(jié)果給服務(wù)消費(fèi)者
            return new RpcResult();
        }
    }

    private void addFailed(Invocation invocation, AbstractClusterInvoker router) {
        if (retryFuture == null) {
            synchronized (this) {
                if (retryFuture == null) {
                    // 創(chuàng)建定時(shí)任務(wù)，每隔5秒執(zhí)行一次
                    retryFuture = scheduledExecutorService.scheduleWithFixedDelay(new Runnable() {

                        @Override
                        public void run() {
                            try {
                                // 對(duì)失敗的調(diào)用進(jìn)行重試
                                retryFailed();
                            } catch (Throwable t) {
                                // 如果發(fā)生異常，僅打印異常日志，不拋出
                                logger.error("Unexpected error occur at collect statistic", t);
                            }
                        }
                    }, RETRY_FAILED_PERIOD, RETRY_FAILED_PERIOD, TimeUnit.MILLISECONDS);
                }
            }
        }
        
        // 添加 invocation 和 invoker 到 failed 中，
        // 這里的把 invoker 命名為 router，很奇怪，明顯名不副實(shí)
        failed.put(invocation, router);
    }

    void retryFailed() {
        if (failed.size() == 0) {
            return;
        }
        
        // 遍歷 failed，對(duì)失敗的調(diào)用進(jìn)行重試
        for (Map.Entry> entry : new HashMap>(failed).entrySet()) {
            Invocation invocation = entry.getKey();
            Invoker invoker = entry.getValue();
            try {
                // 再次進(jìn)行調(diào)用
                invoker.invoke(invocation);
                // 調(diào)用成功，則從 failed 中移除 invoker
                failed.remove(invocation);
            } catch (Throwable e) {
                // 僅打印異常，不拋出
                logger.error("Failed retry to invoke method ...");
            }
        }
    }
}

這個(gè)類主要由3個(gè)方法組成，首先是 doInvoker，該方法負(fù)責(zé)初次的遠(yuǎn)程調(diào)用。若遠(yuǎn)程調(diào)用失敗，則通過(guò) addFailed 方法將調(diào)用信息存入到 failed 中，等待定時(shí)重試。addFailed 在開(kāi)始階段會(huì)根據(jù) retryFuture 為空與非，來(lái)決定是否開(kāi)啟定時(shí)任務(wù)。retryFailed 方法則是包含了失敗重試的邏輯，該方法會(huì)對(duì) failed 進(jìn)行遍歷，然后依次對(duì) Invoker 進(jìn)行調(diào)用。調(diào)用成功則將 Invoker 從 failed 中移除，調(diào)用失敗則忽略失敗原因。

以上就是 FailbackClusterInvoker 的執(zhí)行邏輯，不是很復(fù)雜，繼續(xù)往下看。

FailfastClusterInvoker 只會(huì)進(jìn)行一次調(diào)用，失敗后立即拋出異常。適用于冪等操作，比如新增記錄。樓主日常開(kāi)發(fā)中碰到過(guò)一次程序連續(xù)插入三條同樣的記錄問(wèn)題，原因是新增記錄過(guò)程中包含了一些耗時(shí)操作，導(dǎo)致接口超時(shí)。而我當(dāng)時(shí)使用的是 Dubbo 默認(rèn)的 Cluster Invoker，即 FailoverClusterInvoker。其會(huì)在調(diào)用失敗后進(jìn)行重試，所以導(dǎo)致插入服務(wù)提供者插入了3條同樣的數(shù)據(jù)。如果當(dāng)時(shí)考慮使用 FailfastClusterInvoker，就不會(huì)出現(xiàn)這種問(wèn)題了。當(dāng)然此時(shí)接口仍然會(huì)超時(shí)，所以更合理的做法是使用 Dubbo 異步特性?；蛘邇?yōu)化服務(wù)邏輯，避免超時(shí)。

其他的不多說(shuō)了，下面直接看源碼吧。

public class FailfastClusterInvoker extends AbstractClusterInvoker {

    @Override
    public Result doInvoke(Invocation invocation, List> invokers, LoadBalance loadbalance) throws RpcException {
        checkInvokers(invokers, invocation);
        // 選擇 Invoker
        Invoker invoker = select(loadbalance, invocation, invokers, null);
        try {
            // 調(diào)用 Invoker
            return invoker.invoke(invocation);
        } catch (Throwable e) {
            if (e instanceof RpcException && ((RpcException) e).isBiz()) {
                // 拋出異常
                throw (RpcException) e;
            }
            // 拋出異常
            throw new RpcException(..., "Failfast invoke providers ...");
        }
    }
}

上面代碼比較簡(jiǎn)單了，首先是通過(guò) select 方法選擇 Invoker，然后進(jìn)行遠(yuǎn)程調(diào)用。如果調(diào)用失敗，則立即拋出異常。FailfastClusterInvoker 就先分析到這，下面分析 FailsafeClusterInvoker。

FailsafeClusterInvoker 是一種失敗安全的 Cluster Invoker。所謂的失敗安全是指，當(dāng)調(diào)用過(guò)程中出現(xiàn)異常時(shí)，F(xiàn)ailsafeClusterInvoker 僅會(huì)打印異常，而不會(huì)拋出異常。Dubbo 官方給出的應(yīng)用場(chǎng)景是寫(xiě)入審計(jì)日志等操作，這個(gè)場(chǎng)景我在日常開(kāi)發(fā)中沒(méi)遇到過(guò)，沒(méi)發(fā)言權(quán)，就不多說(shuō)了。下面直接分析源碼。

public class FailsafeClusterInvoker extends AbstractClusterInvoker {

    @Override
    public Result doInvoke(Invocation invocation, List> invokers, LoadBalance loadbalance) throws RpcException {
        try {
            checkInvokers(invokers, invocation);
            // 選擇 Invoker
            Invoker invoker = select(loadbalance, invocation, invokers, null);
            // 進(jìn)行遠(yuǎn)程調(diào)用
            return invoker.invoke(invocation);
        } catch (Throwable e) {
            // 打印錯(cuò)誤日志，但不拋出
            logger.error("Failsafe ignore exception: " + e.getMessage(), e);
            // 返回空結(jié)果忽略錯(cuò)誤
            return new RpcResult();
        }
    }
}

FailsafeClusterInvoker 的邏輯和 FailfastClusterInvoker 的邏輯一樣簡(jiǎn)單，因此就不多說(shuō)了。繼續(xù)下面分析。

ForkingClusterInvoker 會(huì)在運(yùn)行時(shí)通過(guò)線程池創(chuàng)建多個(gè)線程，并發(fā)調(diào)用多個(gè)服務(wù)提供者。只要有一個(gè)服務(wù)提供者成功返回了結(jié)果，doInvoke 方法就會(huì)立即結(jié)束運(yùn)行。ForkingClusterInvoker 的應(yīng)用場(chǎng)景是在一些對(duì)實(shí)時(shí)性要求比較高讀操作（注意是讀操作，并行寫(xiě)操作可能不安全）下使用，但這將會(huì)耗費(fèi)更多的服務(wù)資源。下面來(lái)看該類的實(shí)現(xiàn)。

public class ForkingClusterInvoker extends AbstractClusterInvoker {
    
    private final ExecutorService executor = Executors.newCachedThreadPool(
            new NamedInternalThreadFactory("forking-cluster-timer", true));

    @Override
    public Result doInvoke(final Invocation invocation, List> invokers, LoadBalance loadbalance) throws RpcException {
        try {
            checkInvokers(invokers, invocation);
            final List> selected;
            // 獲取 forks 配置
            final int forks = getUrl().getParameter(Constants.FORKS_KEY, Constants.DEFAULT_FORKS);
            // 獲取超時(shí)配置
            final int timeout = getUrl().getParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
            // 如果 forks 配置不合理，則直接將 invokers 賦值給 selected
            if (forks <= 0 || forks >= invokers.size()) {
                selected = invokers;
            } else {
                selected = new ArrayList>();
                // 循環(huán)選出 forks 個(gè) Invoker，并添加到 selected 中
                for (int i = 0; i < forks; i++) {
                    // 選擇 Invoker
                    Invoker invoker = select(loadbalance, invocation, invokers, selected);
                    if (!selected.contains(invoker)) {
                        selected.add(invoker);
                    }
                }
            }
            
            // ----------------------? 分割線1 ?---------------------- //
            
            RpcContext.getContext().setInvokers((List) selected);
            final AtomicInteger count = new AtomicInteger();
            final BlockingQueue