2. 程式人生 > >Dubbo 原始碼分析 - 叢集容錯之 Cluster

Dubbo 原始碼分析 - 叢集容錯之 Cluster

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

1.簡介

為了避免單點故障，現在的應用至少會部署在兩臺伺服器上。對於一些負載比較高的服務，會部署更多臺伺服器。這樣，同一環境下的服務提供者數量會大於1。對於服務消費者來說，同一環境下出現了多個服務提供者。這時會出現一個問題，服務消費者需要決定選擇哪個服務提供者進行呼叫。另外服務呼叫失敗時的處理措施也是需要考慮的，是重試呢，還是丟擲異常，亦或是隻列印異常等。為了處理這些問題，Dubbo 定義了叢集介面 Cluster 以及及 Cluster Invoker。叢集 Cluster 用途是將多個服務提供者合併為一個 Cluster Invoker，並將這個 Invoker 暴露給服務消費者。這樣一來，服務消費者只需通過這個 Invoker 進行遠端呼叫即可，至於具體呼叫哪個服務提供者，以及呼叫失敗後如何處理等問題，現在都交給叢集模組去處理。叢集模組是服務提供者和服務消費者的中間層，為服務消費者遮蔽了服務提供者的情況，這樣服務消費者就可以處理遠端呼叫相關事宜。比如發請求，接受服務提供者返回的資料等。這就是叢集的作用。

Dubbo 提供了多種叢集實現，包含但不限於 Failover Cluster、Failfast Cluster 和 Failsafe Cluster 等。每種叢集實現類的用途不同，接下來我會一一進行分析。

 2. 叢集容錯

在對叢集相關程式碼進行分析之前，這裡有必要先來介紹一下叢集容錯的所有元件。包含 Cluster、Cluster Invoker、Directory、Router 和 LoadBalance 等，先來看圖。

* 圖片來源：Dubbo 官方文件

這張圖來自 Dubbo 官方文件，接下來我會按照這張圖介紹叢集工作過程。叢集工作過程可分為兩個階段，第一個階段是在服務消費者初始化期間，叢集 Cluster 實現類為服務消費者建立 Cluster Invoker 例項，即上圖中的 merge 操作。第二個階段是在服務消費者進行遠端呼叫時。以 FailoverClusterInvoker 為例，該型別 Cluster Invoker 首先會呼叫 Directory 的 list 方法列舉 Invoker 列表（可將 Invoker 簡單理解為服務提供者）。Directory 的用途是儲存 Invoker，可簡單類比為 List<Invoker>。其實現類 RegistryDirectory 是一個動態服務目錄，可感知註冊中心配置的變化，它所持有的 Inovker 列表會隨著註冊中心內容的變化而變化。每次變化後，RegistryDirectory 會動態增刪 Inovker，並呼叫 Router 的 route 方法進行路由，過濾掉不符合路由規則的 Invoker。回到上圖，Cluster Invoker 實際上並不會直接呼叫 Router 進行路由。當 FailoverClusterInvoker 拿到 Directory 返回的 Invoker 列表後，它會通過 LoadBalance 從 Invoker 列表中選擇一個 Inovker。最後 FailoverClusterInvoker 會將引數傳給 LoadBalance 選擇出的 Invoker 例項的 invoker 方法，進行真正的 RPC 呼叫。

以上就是叢集工作的整個流程，這裡並沒介紹叢集是如何容錯的。Dubbo 主要提供了這樣幾種容錯方式：

  • Failover Cluster - 失敗自動切換
  • Failfast Cluster - 快速失敗
  • Failsafe Cluster - 失敗安全
  • Failback Cluster - 失敗自動恢復
  • Forking Cluster - 並行呼叫多個服務提供者

這裡暫時只對這幾種容錯模式進行簡單的介紹，在接下來的章節中，我會重點分析這幾種容錯模式的具體實現。好了，關於叢集的工作流程和容錯模式先說到這，接下來進入原始碼分析階段。

 3.原始碼分析

 3.1 Cluster 實現類分析

我在上一章提到了叢集介面 Cluster 和 Cluster Invoker，這兩者是不同的。Cluster 是介面，而 Cluster Invoker 是一種 Invoker。服務提供者的選擇邏輯，以及遠端呼叫失敗後的的處理邏輯均是封裝在 Cluster Invoker 中。那麼 Cluster 介面和相關實現類有什麼用呢？用途比較簡單，用於生成 Cluster Invoker，僅此而已。下面我們來看一下原始碼。

1
2
3
4
5
6
7
8
9
10

public class FailoverCluster implements Cluster {

    public final static String NAME = "failover";

    @Override
    public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
        // 建立並返回 FailoverClusterInvoker 物件
        return new FailoverClusterInvoker<T>(directory);
    }
}

如上，FailoverCluster 總共就包含這幾行程式碼，用於建立 FailoverClusterInvoker 物件，很簡單。下面再看一個。

1
2
3
4
5
6
7
8
9
10
11

public class FailbackCluster implements Cluster {

    public final static String NAME = "failback";

    @Override
    public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
        // 建立並返回 FailbackClusterInvoker 物件
        return new FailbackClusterInvoker<T>(directory);
    }

}

如上，FailbackCluster 的邏輯也是很簡單，無需解釋了。所以接下來，我們把重點放在各種 Cluster Invoker 上

 3.2 Cluster Invoker 分析

我們首先從各種 Cluster Invoker 的父類 AbstractClusterInvoker 原始碼開始說起。前面說過，叢集工作過程可分為兩個階段，第一個階段是在服務消費者初始化期間，這個在服務引用那篇文章中已經分析過了，這裡不再贅述。第二個階段是在服務消費者進行遠端呼叫時，此時 AbstractClusterInvoker 的 invoke 方法會被呼叫。列舉 Invoker，負載均衡等操作均會在此階段被執行。因此下面先來看一下 invoke 方法的邏輯。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

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

    // 繫結 attachments 到 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);
    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);
    
    // 呼叫 doInvoke 進行後續操作
    return doInvoke(invocation, invokers, loadbalance);
}

// 抽象方法，由子類實現
protected abstract Result doInvoke(Invocation invocation, List<Invoker<T>> invokers,
                                       LoadBalance loadbalance) throws RpcException;

AbstractClusterInvoker 的 invoke 方法主要用於列舉 Invoker，以及載入 LoadBalance。最後再呼叫模板方法 doInvoke 進行後續操作。下面我們來看一下 Invoker 列舉方法 list(Invocation) 的邏輯，如下：

1
2
3
4
5

protected List<Invoker<T>> list(Invocation invocation) throws RpcException {
    // 呼叫 Directory 的 list 方法
    List<Invoker<T>> invokers = directory.list(invocation);
    return invokers;
}

如上，AbstractClusterInvoker 中的 list 方法做的事情很簡單，只是簡單的呼叫了 Directory 的 list 方法，沒有其他更多的邏輯了。Directory 的 list 方法我在前面的文章中已經分析過了，這裡就不贅述了。

接下來，我們把目光轉移到 AbstractClusterInvoker 的各種實現類上，來看一下這些實現類是如何實現 doInvoke 方法邏輯的。

 3.2.1 FailoverClusterInvoker

FailoverClusterInvoker 在呼叫失敗時，會自動切換 Invoker 進行重試。在無明確配置下，Dubbo 會使用這個類作為預設 Cluster Invoker。下面來看一下該類的邏輯。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

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

    // 省略部分程式碼

    @Override
    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;
        }
        RpcException le = null;
        List<Invoker<T>> invoked = new ArrayList<Invoker<T>>(copyinvokers.size());
        Set<String> providers = new HashSet<String>(len);
        // 迴圈呼叫，失敗重試
        for (int i = 0; i < len; i++) {
            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);
                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 ...");
    }
}

如上，FailoverClusterInvoker 的 doInvoke 方法首先是獲取重試次數，然後根據重試次數進行迴圈呼叫，失敗後進行重試。在 for 迴圈內，首先是通過負載均衡元件選擇一個 Invoker，然後再通過這個 Invoker 的 invoke 方法進行遠端呼叫。如果失敗了，記錄下異常，並進行重試。重試時會再次呼叫父類的 list 方法列舉 Invoker。整個流程大致如此，不是很難理解。下面我們看一下 select 方法的邏輯。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

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();

    // 獲取 sticky 配置，sticky 表示粘滯連線。所謂粘滯連線是指讓服務消費者儘可能的
    // 呼叫同一個服務提供者，除非該提供者掛了再進行切換
    boolean sticky = invokers.get(0).getUrl().getMethodParameter(methodName, Constants.CLUSTER_STICKY_KEY, Constants.DEFAULT_CLUSTER_STICKY);
    {
        // 檢測 invokers 列表是否包含 stickyInvoker，如果不包含，
        // 說明 stickyInvoker 代表的服務提供者掛了，此時需要將其置空
        if (stickyInvoker != null && !invokers.contains(stickyInvoker)) {
            stickyInvoker = null;
        }
        
        // 在 sticky 為 true，且 stickyInvoker != null 的情況下。如果 selected 包含 
        // stickyInvoker，表明 stickyInvoker 對應的服務提供者可能因網路原因未能成功提供服務。
        // 但是該提供者並沒掛，此時 invokers 列表中仍存在該服務提供者對應的 Invoker。
        if (sticky && stickyInvoker != null && (selected == null || !selected.contains(stickyInvoker))) {
            // availablecheck 表示是否開啟了可用性檢查，如果開啟了，則呼叫 stickyInvoker 的 
            // isAvailable 方法進行檢查，如果檢查通過，則直接返回 stickyInvoker。
            if (availablecheck && stickyInvoker.isAvailable()) {
                return stickyInvoker;
            }
        }
    }
    
    // 如果執行緒走到當前程式碼處，說明前面的 stickyInvoker 為空，或者不可用。
    // 此時呼叫繼續呼叫 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 在此之前沒有成功提供服務（但其仍然處於存活狀態）。此時我們認為這個服務不可靠，不應該在重試期間內再次被呼叫，因此這個時候不會返回該 stickyInvoker。如果 selected 不包含 stickyInvoker，此時還需要進行可用性檢測，比如檢測服務提供者網路連通性等。當可用性檢測通過，才可返回 stickyInvoker，否則呼叫 doSelect 方法選擇 Invoker。如果 sticky 為 true，此時會將 doSelect 方法選出的 Invoker 賦值給 stickyInvoker。

以上就是 select 方法的邏輯，這段邏輯看起來不是很複雜，但是資訊量比較大。不搞懂 invokers 和 selected 兩個入參的含義，以及粘滯連線特性，這段程式碼應該是沒法看懂的。大家在閱讀這段程式碼時，不要忽略了對背景知識的理解。其他的不多說了，繼續向下分析。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

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);
    if (loadbalance == null) {
        // 如果 loadbalance 為空，這裡通過 SPI 載入 Loadbalance，預設為 RandomLoadBalance
        loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(Constants.DEFAULT_LOADBALANCE);
    }
    
    // 通過負載均衡元件選擇 Invoker
    Invoker<T> invoker = loadbalance.select(invokers, getUrl(), invocation);

	// 如果 selected 包含負載均衡選擇出的 Invoker，或者該 Invoker 無法經過可用性檢查，此時進行重選
    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 {
                // rinvoker 為空，定位 invoker 在 invokers 中的位置
                int index = invokers.indexOf(invoker);
                try {
                    // 獲取 index + 1 位置處的 Invoker，以下程式碼等價於：
                    //     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 主要做了兩件事，第一是通過負載均衡元件選擇 Invoker。第二是，如果選出來的 Invoker 不穩定，或不可用，此時需要呼叫 reselect 方法進行重選。若 reselect 選出來的 Invoker 為空，此時定位 invoker 在 invokers 列表中的位置 index，然後獲取 index + 1 處的 invoker，這也可以看做是重選邏輯的一部分。關於負載均衡的選擇邏輯，我將會在下篇文章進行詳細分析。下面我們來看一下 reselect 方法的邏輯。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

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

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

    // 根據 availablecheck 進行不同的處理
    if (availablecheck) {
        // 遍歷 invokers 列表
        for (Invoker<T> invoker : invokers) {
            // 檢測可用性
            if (invoker.isAvailable()) {
                // 如果 selected 列表不包含當前 invoker，則將其新增到 reselectInvokers 中
                if (selected == null || !selected.contains(invoker)) {
                    reselectInvokers.add(invoker);
                }
            }
        }
        
        // reselectInvokers 不為空，此時通過負載均衡元件進行選擇
        if (!reselectInvokers.isEmpty()) {
            return loadbalance.select(reselectInvokers, getUrl(), invocation);
        }

    // 不檢查 Invoker 可用性
    } else {
        for (Invoker<T> invoker : invokers) {
            // 如果 selected 列表不包含當前 invoker，則將其新增到 reselectInvokers 中
            if (selected == null || !selected.contains(invoker)) {
                reselectInvokers.add(invoker);
            }
        }
        if (!reselectInvokers.isEmpty()) {
            // 通過負載均衡元件進行選擇
            return loadbalance.select(reselectInvokers, getUrl(), invocation);
        }
    }

    {
        // 若執行緒走到此處，說明 reselectInvokers 集合為空，此時不會呼叫負載均衡元件進行篩選。
        // 這裡從 selected 列表中查詢可用的 Invoker，並將其新增到 reselectInvokers 集合中
        if (selected != null) {
            for (Invoker<T> invoker : selected) {
                if ((invoker.isAvailable())
                        && !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 列表中繼續查詢。關於 reselect 方法就先分析到這，繼續分析其他的 Cluster Invoker。

 3.2.2 FailbackClusterInvoker

FailbackClusterInvoker 會在呼叫失敗後，返回一個空結果給服務提供者。並通過定時任務對失敗的呼叫進行重傳，適合執行訊息通知等操作。下面來看一下它的實現邏輯。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77

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

    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<Invocation, AbstractClusterInvoker<?>> failed = new ConcurrentHashMap<Invocation, AbstractClusterInvoker<?>>();
    private volatile ScheduledFuture<?> retryFuture;

    @Override
    protected Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
        try {
            checkInvokers(invokers, invocation);
            // 選擇 Invoker
            Invoker<T> invoker = select(loadbalance, invocation, invokers, null);
            // 進行呼叫
            return invoker.invoke(invocation);
        } catch (Throwable e) {
            // 如果呼叫過程中發生異常，此時僅列印錯誤日誌，不丟擲異常
            logger.error("Failback to invoke method ...");
            
            // 記錄呼叫資訊
            addFailed(invocation, this);
            // 返回一個空結果給服務消費者
            return new RpcResult();
        }
    }

    private void addFailed(Invocation invocation, AbstractClusterInvoker<?> router) {
        if (retryFuture == null) {
            synchronized (this) {
                if (retryFuture == null) {
                    // 建立定時任務，每隔5秒執行一次
                    retryFuture = scheduledExecutorService.scheduleWithFixedDelay(new Runnable() {

                        @Override
                        public void run() {
                            try {
                                // 對失敗的呼叫進行重試
                                retryFailed();
                            } catch (Throwable t) {
                                // 如果發生異常，僅列印異常日誌，不丟擲
                                logger.error("Unexpected error occur at collect statistic", t);
                            }
                        }
                    }, RETRY_FAILED_PERIOD, RETRY_FAILED_PERIOD, TimeUnit.MILLISECONDS);
                }
            }
        }
        
        // 新增 invocation 和 invoker 到 failed 中，
        // 這裡的把 invoker 命名為 router，很奇怪，明顯名不副實
        failed.put(invocation, router);
    }

    void retryFailed() {
        if (failed.size() == 0) {
            return;
        }
        
        // 遍歷 failed，對失敗的呼叫進行重試
        for (Map.Entry<Invocation, AbstractClusterInvoker<?>> entry : new HashMap<Invocation, AbstractClusterInvoker<?>>(failed).entrySet()) {
            Invocation invocation = entry.getKey();
            Invoker<?> invoker = entry.getValue();
            try {
                // 再次進行呼叫
                invoker.invoke(invocation);
                // 呼叫成功，則從 failed 中移除 invoker
                failed.remove(invocation);
            } catch (Throwable e) {
                // 僅列印異常，不丟擲
                logger.error("Failed retry to invoke method ...");
            }
        }
    }
}

這個類主要由3個方法組成，首先是 doInvoker，該方法負責初次的遠端呼叫。若遠端呼叫失敗，則通過 addFailed 方法將呼叫資訊存入到 failed 中，等待定時重試。addFailed 在開始階段會根據 retryFuture 為空與否，來決定是否開啟定時任務。retryFailed 方法則是包含了失敗重試的邏輯，該方法會對 failed 進行遍歷，然後依次對 Invoker 進行呼叫。呼叫成功則將 Invoker 從 failed 中移除，呼叫失敗則忽略失敗原因。

以上就是 FailbackClusterInvoker 的執行邏輯，不是很複雜，繼續往下看。

 3.2.3 FailfastClusterInvoker

FailfastClusterInvoker 只會進行一次呼叫，失敗後立即丟擲異常。適用於冪等操作，比如新增記錄。樓主日常開發中碰到過一次程式連續插入三條同樣的記錄問題，原因是新增記錄過程中包含了一些耗時操作，導致介面超時。而我當時使用的是 Dubbo 預設的 Cluster Invoker，即 FailoverClusterInvoker。其會在呼叫失敗後進行重試，所以導致插入服務提供者插入了3條同樣的資料。如果當時考慮使用 FailfastClusterInvoker，就不會出現這種問題了。當然此時介面仍然會超時，所以更合理的做法是使用 Dubbo 非同步特性。或者優化服務邏輯，避免超時。

其他的不多說了，下面直接看原始碼吧。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

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

    @Override
    public Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
        checkInvokers(invokers, invocation);
        // 選擇 Invoker
        Invoker<T> invoker = select(loadbalance, invocation, invokers, null);
        try {
            // 呼叫 Invoker
            return invoker.invoke(invocation);
        } catch (Throwable e) {
            if (e instanceof RpcException && ((RpcException) e).isBiz()) {
                // 丟擲異常
                throw (RpcException) e;
            }
            // 丟擲異常
            throw new RpcException(..., "Failfast invoke providers ...");
        }
    }
}

上面程式碼比較簡單了，首先是通過 select 方法選擇 Invoker，然後進行遠端呼叫。如果呼叫失敗，則立即丟擲異常。FailfastClusterInvoker 就先分析到這，下面分析 FailsafeClusterInvoker。

 3.2.4 FailsafeClusterInvoker

FailsafeClusterInvoker 是一種失敗安全的 Cluster Invoker。所謂的失敗安全是指，當呼叫過程中出現異常時，FailsafeClusterInvoker 僅會列印異常，而不會丟擲異常。Dubbo 官方給出的應用場景是寫入審計日誌等操作，這個場景我在日常開發中沒遇到過，沒發言權，就不多說了。下面直接分析原始碼。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

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

    @Override
    public Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
        try {
            checkInvokers(invokers, invocation);
            // 選擇 Invoker
            Invoker<T> invoker = select(loadbalance, invocation, invokers, null);
            // 進行遠端呼叫
            return invoker.invoke(invocation);
        } catch (Throwable e) {
			// 列印錯誤日誌，但不丟擲
            logger.error("Failsafe ignore exception: " + e.getMessage(), e);
            // 返回空結果忽略錯誤
            return new RpcResult();
        }
    }
}

FailsafeClusterInvoker 的邏輯和 FailfastClusterInvoker 的邏輯一樣簡單，因此就不多說了。繼續下面分析。

 3.2.5 ForkingClusterInvoker

ForkingClusterInvoker 會在執行時通過執行緒池建立多個執行緒，併發呼叫多個服務提供者。只要有一個服務提供者成功返回了結果，doInvoke 方法就會立即結束執行。ForkingClusterInvoker 的應用場景是在一些對實時性要求比較高讀操作（注意是讀操作，並行寫操作可能不安全）下使用，但這將會耗費更多的服務資源。下面來看該類的實現。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80

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

    @Override
    public Result doInvoke(final Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
        try {
            checkInvokers(invokers, invocation);
            final List<Invoker<T>> selected;
            // 獲取 forks 配置
            final int forks = getUrl().getParameter(Constants.FORKS_KEY, Constants.DEFAULT_FORKS);
            // 獲取超時配置
            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<Invoker<T>>();
                // 迴圈選出 forks 個 Invoker，並新增到 selected 中
                for (int i = 0; i < forks; i++) {
                    // 選擇 Invoker
                    Invoker<T> 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<Object> ref = new LinkedBlockingQueue<Object>();
            // 遍歷 selected 列表
            for (final Invoker<T> invoker : selected) {
                // 為每個 Invoker 建立一個執行執行緒
                executor.execute(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            // 進行遠端呼叫
                            Result result = invoker.invoke(invocation);
                            // 將結果存到阻塞佇列中
                            ref.offer(result);
                        } catch (Throwable e) {
                            int value = count.incrementAndGet();
                            // 僅在 value 大於等於 selected.size() 時，才將異常物件
                            // 放入阻塞佇列中，請大家思考一下為什麼要這樣做。
                            if (value >= selected.size()) {
                                // 將異常物件存入到阻塞佇列中
                                ref.offer(e);
                            }
                        }
                    }
                });
            }
            
            // ----------------------✨ 分割線2 ✨---------------------- //
            
            try {
                // 從阻塞佇列中取出遠端呼叫結果
                Object ret = ref.poll(timeout, TimeUnit.MILLISECONDS);
                
                // 如果結果型別為 Throwable，則丟擲異常
                if (ret instanceof Throwable) {
                    Throwable e = (Throwable) ret;
                    throw new RpcException(..., "Failed to forking invoke provider ...");
                }
                
                // 返回結果
                return (Result) ret;
            } catch (InterruptedException e) {
                throw new RpcException("Failed to forking invoke provider ...");
            }
        } finally {
            RpcContext.getContext().clearAttachments();
        }
    }
}

ForkingClusterInvoker 的 doInvoker 方法比較長，這裡我通過兩個分割線將整個方法劃分為三個邏輯塊。從方法開始，到分割線1之間的程式碼主要是用於選出 forks 個 Invoker，為接下來的併發呼叫提供輸入。分割線1和分割線2之間的邏輯主要是通過執行緒池併發呼叫多個 Invoker，並將結果儲存在阻塞佇列中。分割線2到方法結尾之間的邏輯主要用於從阻塞佇列中獲取返回結果，並對返回結果型別進行判斷。如果為異常型別，則直接丟擲，否則返回。

以上就是ForkingClusterInvoker 的 doInvoker 方法大致過程。我在分割線1和分割線2之間的程式碼上留了一個問題，問題是這樣的：為什麼要在 value >= selected.size() 的情況下，才將異常物件新增到阻塞佇列中？這裡來解答一下。原因是這樣的，在並行呼叫多個服務提供者的情況下，哪怕只有一個服務提供者成功返回結果，而其他全部失敗。此時 ForkingClusterInvoker 仍應該返回成功的結果，而非丟擲異常。在 value >= selected.size() 時將異常物件放入阻塞佇列中，可以保證異常物件不會出現在正常結果的前面，這樣可從阻塞佇列中優先取出正常的結果。

好了，關於 ForkingClusterInvoker 就先分析到這，接下來分析最後一個 Cluster Invoker。

 3.2.6 BroadcastClusterInvoker

本章的最後，我們再來看一下 BroadcastClusterInvoker。BroadcastClusterInvoker 會逐個呼叫每個服務提供者，如果其中一臺報錯，在迴圈呼叫結束後，BroadcastClusterInvoker 會丟擲異常。看官方文件上的說明，該類通常用於通知所有提供者更新快取或日誌等本地資源資訊。這個使用場景筆者也沒遇到過，沒法詳細說明了，所以下面還是直接分析原始碼吧。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

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

    @Override
    public Result doInvoke(final Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
        checkInvokers(invokers, invocation);
        RpcContext.getContext().setInvokers((List) invokers);
        RpcException exception = null;
        Result result = null;
        // 遍歷 Invoker 列表，逐個呼叫
        for (Invoker<T> invoker : invokers) {
            try {
                // 進行遠端呼叫
                result = invoker.invoke(invocation);
            } catch (RpcException e) {
                exception = e;
                logger.warn(e.getMessage(), e);
            } catch (Throwable e) {
                exception = new RpcException(e.getMessage(), e);
                logger.warn(e.getMessage(), e);
            }
        }
        
        // exception 不為空，則丟擲異常
        if (exception != null) {
            throw exception;
        }
        return result;
    }
}

以上就是 BroadcastClusterInvoker 的程式碼，比較簡單，就不多說了。

 4.總結

本篇文章較為詳細的分析了 Dubbo 叢集容錯方面的內容，並詳細分析了叢集容錯的幾種實現方式。叢集容錯對於 Dubbo 框架來說，是很重要的邏輯。叢集模組處於服務提供者和消費者之間，對於服務消費者來說，叢集可向其遮蔽服務提供者叢集的情況，使其能夠專心進行遠端呼叫。除此之外，通過叢集模組，我們還可以對服務之間的呼叫鏈路進行編排優化，治理服務。總的來說，對於 Dubbo 而言，叢集容錯相關邏輯是非常重要的。想要對 Dubbo 有比較深的理解，叢集容錯是繞不過去的。因此，對於這部分內容，大家要認真看一下。

好了，本篇文章就先到這，感謝大家的閱讀。

 附錄：Dubbo 原始碼分析系列文章

時間 文章
2018-10-01 Dubbo 原始碼分析 - SPI 機制
2018-10-13 Dubbo 原始碼分析 - 自適應拓展原理
2018-10-31 Dubbo 原始碼分析 - 服務匯出
2018-11-12 Dubbo 原始碼分析 - 服務引用
2018-11-17 Dubbo 原始碼分析 - 叢集容錯之 Directory
2018-11-20 Dubbo 原始碼分析 - 叢集容錯之 Router
2018-11-24 Dubbo 原始碼分析 - 叢集容錯之 Cluster

http://www.tianxiaobo.com/2018/11/24/Dubbo-%E6%BA%90%E7%A0%81%E5%88%86%E6%9E%90-%E9%9B%86%E7%BE%A4%E5%AE%B9%E9%94%99%E4%B9%8B-Cluster/ 

相關推薦

Dubbo 原始碼分析 - 叢集容錯之 Cluster

1.簡介 為了避免單點故障，現在的應用至少會部署在兩臺伺服器上。對於一些負載比較高的服務，會部署更多臺伺服器。這樣，同一環境下的服務提供者數量會大於1。對於服務消費者來說，同一環境下出現了多個服務提供者。這時會出現一個問題，服務消費者需要決定選擇哪個服務提供者進行呼叫。另外服務呼叫失敗時的處理措

Dubbo 原始碼分析 - 叢集容錯之 Router

1. 簡介 上一篇文章分析了叢集容錯的第一部分 – 服務目錄 Directory。服務目錄在重新整理 Invoker 列表的過程中，會通過 Router 進行服務路由。上一篇文章關於服務路由相關邏輯沒有細緻分析，一筆帶過了，本篇文章將對此進行詳細的分析。首先，先來介紹一下服務目錄是什麼。服務路由包含一條路由

Dubbo 原始碼分析 - 叢集容錯之 LoadBalance

1.簡介 LoadBalance 中文意思為負載均衡，它的職責是將網路請求，或者其他形式的負載“均攤”到不同的機器上。避免叢集中部分伺服器壓力過大，而另一些伺服器比較空閒的情況。通過負載均衡，可以讓每臺伺服器獲取到適合自己處理能力的負載。在為高負載的伺服器分流的同時，還可以避免資源浪費，一舉兩得。負載均衡可

Dubbo 原始碼分析 - 叢集容錯之Directory

1. 簡介 前面文章分析了服務的匯出與引用過程，從本篇文章開始，我將開始分析 Dubbo 叢集容錯方面的原始碼。這部分原始碼包含四個部分，分別是服務目錄 Directory、服務路由 Router、叢集 Cluster 和負載均衡 LoadBalance。這幾個部分的原始碼邏輯比較獨立，我會分四

14. Dubbo原理解析-叢集&容錯之Cluster

Dubbo作為一個分散式的服務治理框架，提供了叢集部署，路由，軟負載均衡及容錯機制 下圖描述了dubbo呼叫過程中的對於叢集，負載等的呼叫關係。 Cluster 將Directory中的多個Invoker偽裝成一個Invoker, 對上層透明，包含叢集的容錯機制 Cl

dubbo原始碼分析-叢集容錯(二)

開發十年，就只剩下這套架構體系了！ >>>   

dubbo原始碼分析8 -- DubboProtocol 之提供端釋出服務export

在前面提到，在RegistryProtocol釋出服務時，首先會dubbo對外提供介面 根據url的地址，協議是dubbo，呼叫protocol.export(…), 但是根據ExtensionLoader.getExtensionLoader獲取的到的pro

Dubbo 原始碼分析系列之三 —— 架構原理

1 核心功能 首先要了解Dubbo提供的三大核心功能： Remoting:遠端通訊 提供對多種NIO框架抽象封裝，包括“同步轉非同步”和“請求-響應”模式的資訊交換方式。 Cluster: 服務框架 提供基於介面方法的透明遠端過程呼叫，包括多協議支援，以及

Dubbo原始碼分析（三）：Dubbo之服務端（Service）

         如上圖所示的Dubbo的暴露服務的過程，不難看出它也和消費者端很像，也需要一個像reference的物件來維護service關聯的所有物件及其屬性，這裡的reference就是provider。由於ServiceBean實現了  Initializ

dubbo叢集容錯之LoadBalance

原文地址：Dubbo 原始碼分析 - 叢集容錯之 LoadBalance dubbo 提供了4種負載均衡實現，分別是基於權重隨機演算法的 RandomLoadBalance、基於最少活躍呼叫數演算法的 LeastActiveLoadBalance、基於 hash 一致性的 ConsistentHashLoa

dubbo原始碼分析之動態編譯

我們執行的Java程式碼，都是編譯之後的位元組碼，dubbo為了實現了基於spi思想的擴充套件特性，特別是能夠靈活新增額外功能。dubbo作為一個高擴充套件性的框架，使的使用者能夠新增自己的需求，根據配置動態生成自己的設配類程式碼，這樣就需要在執行的時候去編譯載

Dubbo原始碼分析之四：服務的呼叫

在呼叫服務之前，先得獲得服務的引用。 ReferenceBean 就是服務的引用。它實現了一個FactoryBean介面，在我們需要一個服務時，FactoryBean介面的getObject() 方法會被呼叫。 public Object getObje

5.dubbo原始碼分析 之 SPI分析

如果大家看過之前的 dubbo 核心 SPI 實現 – 2.dubbo原始碼分析 之 核心SPI實現, 有可能還是一頭霧水,下面我講一下dubbo的具體應用。最典型的應用就是 Protocol 介面。 Protocol 屬於 dubbo 十層結構 中的遠端呼叫

9、dubbo原始碼分析 之 服務遠端暴露(中)

在上一篇文章我們講解了一下 dubbo 遠端服務暴露過程中通過 Netty 進行 Socket 服務暴露。使得遠端客戶端可以訪問這個暴露的服務，這個只是解決了訪問之前點到點的服務呼叫。對於分步式環境當中，越來越多的服務我們如何管理並且治理這些服務是一個問題。因此 dubbo

7、dubbo原始碼分析 之 服務本地暴露

在上一篇文章我們分析了一下 dubbo 在服務暴露發生了哪些事，今天我們就來分析一下整個服務暴露中的本地暴露。(PS:其實我感覺本地暴露蠻雞肋的)。本地暴露需要服務提供方與服務消費方在同一個 JVM。下面我們來寫一個本地暴露使用的例子： 1) DemoS

Dubbo原始碼分析之 SPI（一）

一、概述     dubbo SPI 在dubbo的作用是基礎性的，要想分析研究dubbo的實現原理、dubbo原始碼，都繞不過 dubbo SPI，掌握dubbo SPI 是征服dubbo的必經之路。     本篇文章會詳細介紹dubbo SPI相關的內容，

Dubbo原始碼分析之SPI（二）

一、概述 　　本篇文章是dubbo SPI原始碼分析的第二篇，接著第一篇繼續分析dubbo SPI的內容，我們主要介紹 getDefaultExtension() 獲取預設擴充套件點方法。 由於此方法比較簡單，我們略過示例部分，直接分析原始碼。 二、原始碼分析 　　獲取預設擴充套件方法getDefau

Dubbo原始碼分析之SPI（三）

一、概述 　　本篇介紹自適應擴充套件，方法getAdaptiveExtension()的實現。ExtensionLoader類本身很多功能也使用到了自適應擴充套件。包括ExtensionFactory擴充套件。 　　通俗的講，自適應擴充套件要實現的邏輯是：呼叫擴充套件點的方法時，自動判斷要呼叫那個擴充套件點實

Dubbo原始碼分析：Dubbo自己實現的IOC

  在建立自適應例項時，都會呼叫ExtensionLoader的injectExtension方法： @SuppressWarnings("unchecked") private T createAdaptiveExtension() { try {

Dubbo 原始碼分析 - 服務匯出全過程解析

1.服務匯出過程 本篇文章，我們來研究一下 Dubbo 匯出服務的過程。Dubbo 服務匯出過程始於 Spring 容器釋出重新整理事件，Dubbo 在接收到事件後，會立即執行服務匯出邏輯。整個邏輯大致可分為三個部分，第一是前置工作，主要用於檢查引數，組裝 URL。第二是匯出服務，包含匯出服務到本地 (JV