Dubbo 源碼分析 - 集羣容錯之 Cluster
1.簡介
爲了不單點故障，如今的應用至少會部署在兩臺服務器上。對於一些負載比較高的服務，會部署更多臺服務器。這樣，同一環境下的服務提供者數量會大於1。對於服務消費者來講，同一環境下出現了多個服務提供者。這時會出現一個問題，服務消費者須要決定選擇哪一個服務提供者進行調用。另外服務調用失敗時的處理措施也是須要考慮的，是重試呢，仍是拋出異常，亦或是隻打印異常等。爲了處理這些問題，Dubbo 定義了集羣接口 Cluster 以及及 Cluster Invoker。集羣 Cluster 用途是將多個服務提供者合併爲一個 Cluster Invoker，並將這個 Invoker 暴露給服務消費者。這樣一來，服務消費者只需經過這個 Invoker 進行遠程調用便可，至於具體調用哪一個服務提供者，以及調用失敗後如何處理等問題，如今都交給集羣模塊去處理。集羣模塊是服務提供者和服務消費者的中間層，爲服務消費者屏蔽了服務提供者的狀況，這樣服務消費者就能夠處理遠程調用相關事宜。好比發請求，接受服務提供者返回的數據等。這就是集羣的做用。html
Dubbo 提供了多種集羣實現，包含但不限於 Failover Cluster、Failfast Cluster 和 Failsafe Cluster 等。每種集羣實現類的用途不一樣，接下來我會一一進行分析。java
2. 集羣容錯
在對集羣相關代碼進行分析以前，這裏有必要先來介紹一下集羣容錯的全部組件。包含 Cluster、Cluster Invoker、Directory、Router 和 LoadBalance 等，先來看圖。apache
* 圖片來源：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，僅此而已。下面咱們來看一下源碼。併發
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 對象，很簡單。下面再看一個。負載均衡
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 方法的邏輯。
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) 的邏輯，以下：
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。下面來看一下該類的邏輯。
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 方法的邏輯。
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 兩個入參的含義，以及粘滯鏈接特性，這段代碼應該是無法看懂的。你們在閱讀這段代碼時，不要忽略了對背景知識的理解。其餘的很少說了，繼續向下分析。
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 方法的邏輯。
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 會在調用失敗後，返回一個空結果給服務提供者。並經過定時任務對失敗的調用進行重傳，適合執行消息通知等操做。下面來看一下它的實現邏輯。
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 異步特性。或者優化服務邏輯，避免超時。
其餘的很少說了，下面直接看源碼吧。
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 官方給出的應用場景是寫入審計日誌等操做，這個場景我在平常開發中沒遇到過，沒發言權，就很少說了。下面直接分析源碼。
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 的應用場景是在一些對實時性要求比較高讀操做（注意是讀操做，並行寫操做可能不安全）下使用，但這將會耗費更多的服務資源。下面來看該類的實現。
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 會拋出異常。看官方文檔上的說明，該類一般用於通知全部提供者更新緩存或日誌等本地資源信息。這個使用場景筆者也沒遇到過，無法詳細說明了，因此下面仍是直接分析源碼吧。
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 |
本文在知識共享許可協議 4.0 下發布，轉載需在明顯位置處註明出處
做者：田小波
本文同步發佈在個人我的博客：http://www.tianxiaobo.com
本做品採用知識共享署名-非商業性使用-禁止演繹 4.0 國際許可協議進行許可。
- 1. dubbo源碼分析- 集羣容錯之Cluster（一）
- 2. Dubbo 源碼分析 - 集羣容錯之 Cluster
- 3. dubbo源碼分析-集羣容錯(二)
- 4. dubbo集羣容錯之cluster集羣
- 5. Dubbo源碼解析之Cluster集羣容錯策略介紹
- 6. dubbo源碼分析之集羣容錯failover-13
- 7. Dubbo 源碼分析 - 集羣容錯之 Directory
- 8. Dubbo 源碼分析 - 集羣容錯之Directory
- 9. Dubbo 源碼分析 - 集羣容錯之 LoadBalance
- 10. Dubbo剖析-集羣容錯
- 更多相關文章...
- • Swarm 集羣管理 - Docker教程
- • XML DOM 解析器錯誤 - XML DOM 教程
- • 互聯網組織的未來：剖析GitHub員工的任性之源
- • Java Agent入門實戰（二）-Instrumentation源碼概述
-
每一个你不满意的现在，都有一个你没有努力的曾经。
- 1. Appium入門
- 2. Spring WebFlux 源碼分析(2)-Netty 服務器啓動服務流程 --TBD
- 3. wxpython入門第六步（高級組件）
- 4. CentOS7.5安裝SVN和可視化管理工具iF.SVNAdmin
- 5. jedis 3.0.1中JedisPoolConfig對象缺少setMaxIdle、setMaxWaitMillis等方法，問題記錄
- 6. 一步一圖一代碼，一定要讓你真正徹底明白紅黑樹
- 7. 2018-04-12—（重點）源碼角度分析Handler運行原理
- 8. Spring AOP源碼詳細解析
- 9. Spring Cloud（1）
- 10. python簡單爬去油價信息發送到公衆號