當前位置：首頁 > 编程资源 > 编程问答 >内容正文

编程问答

Dubbo 源码分析 - 集群容错之 Cluster

發布時間：2025/3/21 编程问答 34 豆豆

生活随笔收集整理的這篇文章主要介紹了 Dubbo 源码分析 - 集群容错之 Cluster 小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

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";@Overridepublic <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";@Overridepublic <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);}// 列舉 InvokerList<Invoker<T>> invokers = list(invocation);if (invokers != null && !invokers.isEmpty()) {// 加載 LoadBalanceloadbalance = 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> {// 省略部分代碼@Overridepublic 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);}// 通過負載均衡選擇 InvokerInvoker<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 選擇 InvokerInvoker<T> invoker = doSelect(loadbalance, invocation, invokers, selected);// 如果 sticky 為 true，則將負載均衡組件選出的 Invoker 賦值給 stickyInvokerif (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，默認為 RandomLoadBalanceloadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(Constants.DEFAULT_LOADBALANCE);}// 通過負載均衡組件選擇 InvokerInvoker<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 不為空，則將其賦值給 invokerinvoker = 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;@Overrideprotected Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {try {checkInvokers(invokers, invocation);// 選擇 InvokerInvoker<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() {@Overridepublic 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 中移除 invokerfailed.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> {@Overridepublic Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {checkInvokers(invokers, invocation);// 選擇 InvokerInvoker<T> invoker = select(loadbalance, invocation, invokers, null);try {// 調用 Invokerreturn 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> {@Overridepublic Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {try {checkInvokers(invokers, invocation);// 選擇 InvokerInvoker<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));@Overridepublic 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 賦值給 selectedif (forks <= 0 || forks >= invokers.size()) {selected = invokers;} else {selected = new ArrayList<Invoker<T>>();// 循環選出 forks 個 Invoker，并添加到 selected 中for (int i = 0; i < forks; i++) {// 選擇 InvokerInvoker<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() {@Overridepublic 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> {@Overridepublic 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

本文鏈接：?https://www.tianxiaobo.com/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的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。

上一篇： Dubbo 源码分析 - 集群容错之 R
下一篇： Dubbo 源码分析 - 集群容错之 L