集群容錯？看看Dubbo是如何實現的

大家好，我是java吳彥祖！

隨著網際網路的發展，傳統垂直架構已經無法應對滿足網際網路企業經常面對的高並發、集群和大數據量等特點，所以這就是Dubbo誕生的目的，本文會以最簡單的方式給大家講述Dubbo中集群容錯實現邏輯以及使用場景。

圖1：集群容錯流程

Dubbo集群容錯實現邏輯

在分布式項目中為了做到服務的高可用，我們往往需要服務集群化。如果集群服務中出現了某個服務故障則就需要集群實現容錯機制。

容錯的實現是在Dubbo的容錯層中，具體還是要先看AbstractClusterInvoker這個抽象類，它定義了最基本的整個集群容錯層的流程模版。

通過AbstractClusterInvoker我大致可以看到整個集群容錯層的調用流程是這樣的：

1.調用AbstractClusterInvoker#invoke(final Invocation invocation);初始化LoadBalance和Invoker列表2.調用AbstractClusterInvoker#doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance);3.doInvoker方法的具體實現是在各個實現類中的如下圖2所示；4.實現類中會調用調用AbstractClusterInvoker（父類）中的select方法做負載均衡策略；

圖2：集群容錯實現類

以Failsafe容錯策略為例，具體的容錯的實現代碼是放在doInvoker方法中。

public class FailsafeClusterInvoker<T> extends AbstractClusterInvoker<T> { private static final Logger logger = LoggerFactory.getLogger(FailsafeClusterInvoker.class); public FailsafeClusterInvoker(Directory<T> directory) { super(directory); } // 實現 public Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException { try { this.checkInvokers(invokers, invocation); Invoker<T> invoker = this.select(loadbalance, invocation, invokers, (List)null); return invoker.invoke(invocation); } catch (Throwable var5) { logger.error("Failsafe ignore exception: " + var5.getMessage(), var5); return AsyncRpcResult.newDefaultAsyncResult((Object)null, (Throwable)null, invocation); } }}

既然知道了容錯的實現流程和邏輯，下面還有一個問題就是如何選擇容錯策略。這個其實和負載均衡是一樣的，也是利用的Dubbo的SPI機制。實現的流出大致是這樣的：

1.集群容錯層中有一個Cluster接口，該接口我們可以看到被@SPI註解標註、方法上被@Adaptive註解標識（SPI機制後續會詳細說明）；

// 默認用的是failover（故障轉移）@SPI("failover")public interface Cluster { @Adaptive <T> Invoker<T> join(Directory<T> directory) throws RpcException;}

2.我們點一下實現類AbstractCluster（抽象類）中有一個doJoin的方法，點一下實現類，發現自帶有八種實現類；

圖3：容錯實現類

3.隨便點一個實現類以上面的FailSafeCluster為例，我們可以看到具體的實現方式；

public class FailsafeCluster extends AbstractCluster { public static final String NAME = "failsafe"; public FailsafeCluster() { } public <T> AbstractClusterInvoker<T> doJoin(Directory<T> directory) throws RpcException { return new FailsafeClusterInvoker(directory); }}

我們看到FailsafeCluster在創建時還會傳入一個Directory對象（List<Invoker>）,Directory其實就是一組Invoker。

protected List<Invoker<T>> list(Invocation invocation) throws RpcException { return this.directory.list(invocation); }

4.最後SPI機制肯定是需要配置文件

圖4：Cluster SPI配置文件

Dubbo中自帶的集群容錯機制

說完了dubbo的集群容錯的實現邏輯，我們看看Dubbo給我們提供了哪些容錯的策略。

官網其實已經給我們說了有哪些集群容錯可以選擇（http://dubbo.apache.org/zh-cn/docs/user/demos/fault-tolerent-strategy.html）。我這邊主要是說一下幾種在工作中常用的集群容錯的策略。

• Failover Cluster（失敗自動切換）：失敗自動切換，當出現失敗，重試其它伺服器。通常用於讀操作，但重試會帶來更長延遲。可通過 retries="2" 來設置重試次數(不含第一次)。Failover Cluster其實一般使用在冪等的操作是比較合適的。
• Failfast Cluster（快速失敗）：只發起一次調用，失敗立即報錯。通常用於非冪等性的寫操作，比如新增記錄。
• Failsafe Cluster（失敗安全）：出現異常時，直接忽略。通常用於寫入審計日誌等操作。一般用於可以缺失的非主流程，業務中用的不多；

其實的容錯策略直接去官網查看即可，現實業務中基本不可能使用（反正我是除了面試外，沒用到過）。

Dubbo中如何自定義集群容錯機制

在dubbo中如何進行自定義的集群容錯機制，其實和負載均衡流程是一樣的，畢竟都是基於Dubbo SPI擴展的。我這邊就直接基於Dubbo官網和大家說一下即可，畢竟這個擴展還是比較簡單的（http://dubbo.apache.org/zh-cn/docs/dev/impls/cluster.html）。

• 擴展說明

當有多個服務提供方時，將多個服務提供方組織成一個集群，並偽裝成一個提供方。

• 擴展接口

org.apache.dubbo.rpc.cluster.Cluster

• 擴展配置

<dubbo:protocol cluster="xxx" /><!-- 預設值配置，如果<dubbo:protocol>沒有配置cluster時，使用此配置 --><dubbo:provider cluster="xxx" />

• 已知擴展

• org.apache.dubbo.rpc.cluster.support.FailoverCluster
• org.apache.dubbo.rpc.cluster.support.FailfastCluster
• org.apache.dubbo.rpc.cluster.support.FailsafeCluster
• org.apache.dubbo.rpc.cluster.support.FailbackCluster
• org.apache.dubbo.rpc.cluster.support.ForkingCluster
• org.apache.dubbo.rpc.cluster.support.AvailableCluster

• 擴展示例

Maven 項目結構：

src |-main |-java |-com |-xxx |-XxxCluster.java (實現Cluster接口) |-resources |-META-INF |-dubbo |-org.apache.dubbo.rpc.cluster.Cluster (純文本文件，內容為：xxx=com.xxx.XxxCluster)

XxxCluster.java：

package com.xxx; import org.apache.dubbo.rpc.cluster.Cluster;import org.apache.dubbo.rpc.cluster.support.AbstractClusterInvoker;import org.apache.dubbo.rpc.cluster.Directory;import org.apache.dubbo.rpc.cluster.LoadBalance;import org.apache.dubbo.rpc.Invoker;import org.apache.dubbo.rpc.Invocation;import org.apache.dubbo.rpc.Result;import org.apache.dubbo.rpc.RpcException; public class XxxCluster implements Cluster { public <T> Invoker<T> merge(Directory<T> directory) throws RpcException { return new AbstractClusterInvoker<T>(directory) { public Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException { // ... } }; }}

META-INF/dubbo/org.apache.dubbo.rpc.cluster.Cluster：

xxx=com.xxx.XxxCluster