更新時間:2022-09-16 來源:黑馬程序員 瀏覽量:
服務(wù)集群的概述
概述
為了避免單點故障，現(xiàn)在的應(yīng)用通常至少會部署在兩臺服務(wù)器上，這樣就組成了集群。集群就是單機的多實例，在多個服務(wù)器上部署多個服務(wù)，每個服務(wù)就是一個節(jié)點，部署N個節(jié)點，處理業(yè)務(wù)的能力就提升 N倍(大約)，這些節(jié)點的集合就叫做集群。
管理控制臺
目前的管理控制臺已經(jīng)發(fā)布0.1版本，結(jié)構(gòu)上采取了前后端分離的方式，前端使用Vue和Vuetify分別作為Javascript框架和UI框架，后端采用Spring Boot框架。既可以按照標(biāo)準(zhǔn)的Maven方式進行打包，部署，也可以采用前后端分離的部署方式，方便開發(fā)，功能上，目前具備了服務(wù)查詢，服務(wù)治理(包括Dubbo2.7中新增的治理規(guī)則)以及服務(wù)測試三部分內(nèi)容。
Maven方式部署
- 安裝
git clone https://github.com/apache/dubbo-admin.git cd dubbo-admin mvn clean package cd dubbo-admin-distribution/target java -jar dubbo-admin-0.1.jar
- 訪問
[http://localhost:8080](http://localhost:8080/)
前后端分離部署
- 前端
cd dubbo-admin-ui npm install npm run dev
- 后端
cd dubbo-admin-server mvn clean package cd target java -jar dubbo-admin-server-0.1.jar
- 訪問
[http://localhost:8081](http://localhost:8081/)
- 前后端分離模式下，前端的修改可以實時生效
環(huán)境搭建
略
集群調(diào)用存在的問題
負載均衡
集群容錯
服務(wù)治理
集群的調(diào)用過程
調(diào)用過程
在對集群相關(guān)代碼進行分析之前，這里有必要先來介紹一下集群容錯的所有組件。包含 Cluster、Cluster Invoker、Directory、Router 和 LoadBalance 等。
集群工作過程可分為兩個階段，第一個階段是在服務(wù)消費者初始化期間，集群 Cluster 實現(xiàn)類為服務(wù)消費者創(chuàng)建 Cluster Invoker 實例，即上圖中的 merge 操作。第二個階段是在服務(wù)消費者進行遠程調(diào)用時。以 FailoverClusterInvoker 為例，該類型 Cluster Invoker 首先會調(diào)用 Directory 的 list 方法列舉 Invoker 列表(可將 Invoker 簡單理解為服務(wù)提供者)。Directory 的用途是保存 Invoker，可簡單類比為 List。其實現(xiàn)類 RegistryDirectory 是一個動態(tài)服務(wù)目錄，可感知注冊中心配置的變化，它所持有的 Invoker 列表會隨著注冊中心內(nèi)容的變化而變化。每次變化后，RegistryDirectory 會動態(tài)增刪 Invoker，并調(diào)用 Router 的 route 方法進行路由，過濾掉不符合路由規(guī)則的 Invoker。當(dāng) FailoverClusterInvoker 拿到 Directory 返回的 Invoker 列表后，它會通過 LoadBalance 從 Invoker 列表中選擇一個 Invoker。最后 FailoverClusterInvoker 會將參數(shù)傳給 LoadBalance 選擇出的 Invoker 實例的 invoke 方法，進行真正的遠程調(diào)用。
組件介紹
Directory：它代表多個Invoker，從methodInvokerMap提取，但是他的值是動態(tài)，例如注冊中心的變更。
Router：負責(zé)從多個Invoker中按路由規(guī)則選出子集，例如應(yīng)用隔離或讀寫分離或灰度發(fā)布等等
Cluster：將Directory中的多個Invoker偽裝成一個Invoker,來容錯，調(diào)用失敗重試。
LoadBalance：從多個Invoker選取一個做本次調(diào)用，具體包含很多種負載均衡算法。
Invoker：Provider中的一個可調(diào)用接口。例如DemoService
集群容錯
在分布式系統(tǒng)中，集群某個某些節(jié)點出現(xiàn)問題是大概率事件，因此在設(shè)計分布式RPC框架的過程中，必須要把失敗作為設(shè)計的一等公民來對待。一次調(diào)用失敗之后，應(yīng)該如何選擇對失敗的選擇策略，這是一個見仁見智的問題，每種策略可能都有自己獨特的應(yīng)用場景。因此，作為框架來說，應(yīng)當(dāng)針對不同場景提供多種策略，供用戶進行選擇。
在Dubbo設(shè)計中，通過Cluster這個接口的抽象，把一組可供調(diào)用的Provider信息組合成為一個統(tǒng)一的`Invoker`供調(diào)用方進行調(diào)用。經(jīng)過路由規(guī)則過濾，負載均衡選址后，選中一個具體地址進行調(diào)用，如果調(diào)用失敗，則會按照集群配置的容錯策略進行容錯處理。
內(nèi)置集群容錯方式
Dubbo默認內(nèi)置了若干容錯策略，如果不能滿足用戶需求，則可以通過自定義容錯策略進行配置
Dubbo主要內(nèi)置了如下幾種策略：
- Failover(失敗自動切換)
- Failsafe(失敗安全)
- Failfast(快速失敗)
- Failback(失敗自動恢復(fù))
- Forking(并行調(diào)用)
- Broadcast(廣播調(diào)用)
這些名稱比較相似，概念也比較容易混淆，下面逐一進行解釋。
Failover(失敗自動切換)
`Failover`是高可用系統(tǒng)中的一個常用概念，服務(wù)器通常擁有主備兩套機器配置，如果主服務(wù)器出現(xiàn)故障，則自動切換到備服務(wù)器中，從而保證了整體的高可用性。
Dubbo也借鑒了這個思想，并且把它作為Dubbo`默認的容錯策略`。當(dāng)調(diào)用出現(xiàn)失敗的時候，根據(jù)配置的重試次數(shù)，會自動從其他可用地址中重新選擇一個可用的地址進行調(diào)用，直到調(diào)用成功，或者是達到重試的上限位置。
Dubbo里默認配置的重試次數(shù)是2，也就是說，算上第一次調(diào)用，最多會調(diào)用3次。
其配置方法，容錯策略既可以在服務(wù)提供方配置，也可以服務(wù)調(diào)用方進行配置。而重試次數(shù)的配置則更為靈活，既可以在服務(wù)級別進行配置，也可以在方法級別進行配置。具體優(yōu)先順序為：
```
服務(wù)調(diào)用方方法級配置 > 服務(wù)調(diào)用方服務(wù)級配置 > 服務(wù)提供方方法級配置 > 服務(wù)提供方服務(wù)級配置
```
以XML方式為例，具體配置方法如下：
服務(wù)提供方，服務(wù)級配置
<dubbo:service interface="org.apache.dubbo.demo.DemoService" ref="demoService" cluster="failover" retries="2" />
服務(wù)提供方，方法級配置
<dubbo:service interface="org.apache.dubbo.demo.DemoService" ref="demoService"cluster="failover"> <dubbo:method name="sayHello" retries="2" /> </dubbo:reference>
服務(wù)調(diào)用方，服務(wù)級配置
<dubbo:reference id="demoService" interface="org.apache.dubbo.demo.DemoService" cluster="failover" retries="1"/>
服務(wù)調(diào)用方，方法級配置：
<dubbo:reference id="demoService" interface="org.apache.dubbo.demo.DemoService" cluster="failover"> <dubbo:method name="sayHello" retries="3" /> </dubbo:reference>
Failover可以自動對失敗進行重試，對調(diào)用者屏蔽了失敗的細節(jié)，但是Failover策略也會帶來一些副作用：
- 重試會額外增加一下開銷，例如增加資源的使用，在高負載系統(tǒng)下，額外的重試可能讓系統(tǒng)雪上加霜。
- 重試會增加調(diào)用的響應(yīng)時間。
- 某些情況下，重試甚至?xí)斐少Y源的浪費?？紤]一個調(diào)用場景，A->B->C，如果A處設(shè)置了超時100ms，再B->C的第一次調(diào)用完成時已經(jīng)超過了100ms，但很不幸B->C失敗，這時候會進行重試，但其實這時候重試已經(jīng)沒有意義，因此在A看來這次調(diào)用已經(jīng)超時，A可能已經(jīng)開始執(zhí)行其他邏輯。
Failsafe(失敗安全)
失敗安全策略的核心是即使失敗了也不會影響整個調(diào)用流程。通常情況下用于旁路系統(tǒng)或流程中，它的失敗不影響核心業(yè)務(wù)的正確性。在實現(xiàn)上，當(dāng)出現(xiàn)調(diào)用失敗時，會忽略此錯誤，并記錄一條日志，同時返回一個空結(jié)果，在上游看來調(diào)用是成功的。
應(yīng)用場景，可以用于寫入審計日志等操作。
具體配置方法：
服務(wù)提供方，服務(wù)級配置
<dubbo:service interface="org.apache.dubbo.demo.DemoService" ref="demoService" cluster="failsafe" />
服務(wù)調(diào)用方，服務(wù)級配置
<dubbo:reference id="demoService" interface="org.apache.dubbo.demo.DemoService" cluster="failsafe"/>
其中服務(wù)調(diào)用方配置優(yōu)先于服務(wù)提供方配置。
Failfast(快速失敗)
某些業(yè)務(wù)場景中，某些操作可能是非冪等的，如果重復(fù)發(fā)起調(diào)用，可能會導(dǎo)致出現(xiàn)臟數(shù)據(jù)等。例如調(diào)用某個服務(wù)，其中包含一個數(shù)據(jù)庫的寫操作，如果寫操作完成，但是在發(fā)送結(jié)果給調(diào)用方的過程中出錯了，那么在調(diào)用發(fā)看來這次調(diào)用失敗了，但其實數(shù)據(jù)寫入已經(jīng)完成。這種情況下，重試可能并不是一個好策略，這時候就需要使用到`Failfast`策略，調(diào)用失敗立即報錯。讓調(diào)用方來決定下一步的操作并保證業(yè)務(wù)的冪等性。
具體配置方法：
服務(wù)提供方，服務(wù)級配置
<dubbo:service interface="org.apache.dubbo.demo.DemoService" ref="demoService" cluster="failfast" />
服務(wù)調(diào)用方，服務(wù)級配置
<dubbo:reference id="demoService" interface="org.apache.dubbo.demo.DemoService" cluster="failfast"/>
其中服務(wù)調(diào)用方配置優(yōu)先于服務(wù)提供方配置。
Failback(失敗自動恢復(fù))
`Failback`通常和`Failover`兩個概念聯(lián)系在一起。在高可用系統(tǒng)中，當(dāng)主機發(fā)生故障，通過`Failover`進行主備切換后，待故障恢復(fù)后，系統(tǒng)應(yīng)該具備自動恢復(fù)原始配置的能力。
Dubbo中的`Failback`策略中，如果調(diào)用失敗，則此次失敗相當(dāng)于`Failsafe`，將返回一個空結(jié)果。而與`Failsafe`不同的是，F(xiàn)ailback策略會將這次調(diào)用加入內(nèi)存中的失敗列表中，對于這個列表中的失敗調(diào)用，會在另一個線程中進行異步重試，重試如果再發(fā)生失敗，則會忽略，即使重試調(diào)用成功，原來的調(diào)用方也感知不到了。因此它通常適合于，對于實時性要求不高，且不需要返回值的一些異步操作。
具體配置方法：
服務(wù)提供方，服務(wù)級配置
<dubbo:service interface="org.apache.dubbo.demo.DemoService" ref="demoService" cluster="failsafe" />
服務(wù)調(diào)用方，服務(wù)級配置
<dubbo:reference id="demoService" interface="org.apache.dubbo.demo.DemoService" cluster="failsafe"/>
其中服務(wù)調(diào)用方配置優(yōu)先于服務(wù)提供方配置。
按照目前的實現(xiàn)，F(xiàn)ailback策略還有一些局限，例如內(nèi)存中的失敗調(diào)用列表沒有上限，可能導(dǎo)致堆積，異步重試的執(zhí)行間隔無法調(diào)整，默認是5秒。
Forking(并行調(diào)用)
上述幾種策略中，主要都是針對調(diào)用失敗發(fā)生后如何進行彌補的角度去考慮的，而`Forking`策略則跟上述幾種策略不同，是一種典型的用成本換時間的思路。即第一次調(diào)用的時候就同時發(fā)起多個調(diào)用，只要其中一個調(diào)用成功，就認為成功。在資源充足，且對于失敗的容忍度較低的場景下，可以采用此策略。
具體配置方法：
服務(wù)提供方，服務(wù)級配置
<dubbo:service interface="org.apache.dubbo.demo.DemoService" ref="demoService" cluster="forking" />
服務(wù)調(diào)用方，服務(wù)級配置
<dubbo:reference id="demoService" interface="org.apache.dubbo.demo.DemoService" cluster="forking"/>
其中服務(wù)調(diào)用方配置優(yōu)先于服務(wù)提供方配置。
Broadcast(廣播調(diào)用)
在某些場景下，可能需要對服務(wù)的所有提供者進行操作，此時可以使用廣播調(diào)用策略。此策略會逐個調(diào)用所有提供者，只要任意有一個提供者出錯，則認為此次調(diào)用出錯。通常用于通知所有提供者更新緩存或日志等本地資源信息。
具體配置方法：
服務(wù)提供方，服務(wù)級配置
<dubbo:service interface="org.apache.dubbo.demo.DemoService" ref="demoService" cluster="broadcast" />
服務(wù)調(diào)用方，服務(wù)級配置
<dubbo:reference id="demoService" interface="org.apache.dubbo.demo.DemoService" cluster="broadcast"/>
其中服務(wù)調(diào)用方配置優(yōu)先于服務(wù)提供方配置。
集群容錯調(diào)優(yōu)
下表對各種策略做一個簡單對比，
| 策略名稱 | 優(yōu)點 | 缺點 |
| --------- | -------------------------------- | -------------------------------------- |
| Failover | 對調(diào)用者屏蔽調(diào)用失敗的信息 | 增加RT，額外資源開銷，資源浪費 |
| Failfast | 業(yè)務(wù)快速感知失敗狀態(tài)進行自主決策 | 產(chǎn)生較多報錯的信息 |
| Failsafe | 即使失敗了也不會影響核心流程 | 對于失敗的信息不敏感，需要額外的監(jiān)控 |
| Failback | 失敗自動異步重試 | 重試任務(wù)可能堆積 |
| Forking | 并行發(fā)起多個調(diào)用，降低失敗概率 | 消耗額外的機器資源，需要確保操作冪等性 |
| Broadcast | 支持對所有的服務(wù)提供者進行操作 | 資源消耗很大 |
綜上我們得知，不同的容錯策略往往對應(yīng)不同的業(yè)務(wù)處理，這里做一個總結(jié)如下：
Failover：通常用于對調(diào)用rt不敏感的場景，如讀操作;但重試會帶來更長延遲
Failfast：通常用于非冪等性操作，需要快速感知失敗的場景;比如新增記錄
Failsafe：通常用于旁路系統(tǒng)，失敗不影響核心流程正確性的場景;如日志記錄
Failback：通常用于對于實時性要求不高，且不需要返回值的一些異步操作的場景
Forking：通常用于資源充足，且對于失敗的容忍度較低，實時性要求高的讀操作，但需要浪費更多服務(wù)資源
Broadcast：如通知所有提供者更新緩存或日志等本地資源信息
源碼分析
我們在上一章看到了兩個概念，分別是集群接口 Cluster 和 Cluster Invoker，這兩者是不同的。Cluster 是接口，而 Cluster Invoker 是一種 Invoker。服務(wù)提供者的選擇邏輯，以及遠程調(diào)用失敗后的的處理邏輯均是封裝在 Cluster Invoker 中。那么 Cluster 接口和相關(guān)實現(xiàn)類有什么用呢?用途比較簡單，僅用于生成 Cluster Invoker。下面我們來看一下源碼。
JavaEE
鴻蒙應(yīng)用開發(fā)
集成電路應(yīng)用開發(fā)
人工智能開發(fā)
AI+Linux云計算運維
Python+大數(shù)據(jù)開發(fā)
AI+設(shè)計
軟件測試
新媒體+短視頻
HTML&JS+前端
public class FailoverCluster implements Cluster {
public final static String NAME = "failover";
@Override
public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
// 創(chuàng)建并返回 FailoverClusterInvoker 對象
return new FailoverClusterInvoker<T>(directory);
}
}如上，F(xiàn)ailoverCluster 總共就包含這幾行代碼，用于創(chuàng)建 FailoverClusterInvoker 對象，很簡單。下面再看一個。
public class FailbackCluster implements Cluster {
public final static String NAME = "failback";
@Override
public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
// 創(chuàng)建并返回 FailbackClusterInvoker 對象
return new FailbackClusterInvoker<T>(directory);
}
} 如上，F(xiàn)ailbackCluster 的邏輯也是很簡單，無需解釋了。所以接下來，我們把重點放在各種 Cluster Invoker 上
Cluster Invoker
我們首先從各種 Cluster Invoker 的父類 AbstractClusterInvoker 源碼開始說起。前面說過，集群工作過程可分為兩個階段，第一個階段是在服務(wù)消費者初始化期間，這個在服務(wù)引用那篇文章中分析過，就不贅述。第二個階段是在服務(wù)消費者進行遠程調(diào)用時，此時 AbstractClusterInvoker 的 invoke 方法會被調(diào)用。列舉 Invoker，負載均衡等操作均會在此階段被執(zhí)行。因此下面先來看一下 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);
// 調(diào)用 doInvoke 進行后續(xù)操作
return doInvoke(invocation, invokers, loadbalance);
}
// 抽象方法，由子類實現(xiàn)
protected abstract Result doInvoke(Invocation invocation, List<Invoker<T>> invokers,
LoadBalance loadbalance) throws RpcException;AbstractClusterInvoker 的 invoke 方法主要用于列舉 Invoker，以及加載 LoadBalance。最后再調(diào)用模板方法 doInvoke 進行后續(xù)操作。下面我們來看一下 Invoker 列舉方法 list(Invocation) 的邏輯，如下：
protected List<Invoker<T>> list(Invocation invocation) throws RpcException {
// 調(diào)用 Directory 的 list 方法列舉 Invoker
List<Invoker<T>> invokers = directory.list(invocation);
return invokers;
} 如上，AbstractClusterInvoker 中的 list 方法做的事情很簡單，只是簡單的調(diào)用了 Directory 的 list 方法，沒有其他更多的邏輯了。Directory 即相關(guān)實現(xiàn)類在前文已經(jīng)分析過，這里就不多說了。接下來，我們把目光轉(zhuǎn)移到 AbstractClusterInvoker 的各種實現(xiàn)類上，來看一下這些實現(xiàn)類是如何實現(xiàn) doInvoke 方法邏輯的。
FailoverClusterInvoker
FailoverClusterInvoker 在調(diào)用失敗時，會自動切換 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);
// 獲取重試次數(shù)
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);
// 循環(huán)調(diào)用，失敗重試
for (int i = 0; i < len; i++) {
if (i > 0) {
checkWhetherDestroyed();
// 在進行重試前重新列舉 Invoker，這樣做的好處是，如果某個服務(wù)掛了，
// 通過調(diào)用 list 可得到最新可用的 Invoker 列表
copyinvokers = list(invocation);
// 對 copyinvokers 進行判空檢查
checkInvokers(copyinvokers, invocation);
}
// 通過負載均衡選擇 Invoker
Invoker<T> 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ù)進行循環(huán)調(diào)用，失敗后進行重試。在 for 循環(huán)內(nèi)，首先是通過負載均衡組件選擇一個 Invoker，然后再通過這個 Invoker 的 invoke 方法進行遠程調(diào)用。如果失敗了，記錄下異常，并進行重試。重試時會再次調(diào)用父類的 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;
// 獲取調(diào)用方法名
String methodName = invocation == null ? "" : invocation.getMethodName();
// 獲取 sticky 配置，sticky 表示粘滯連接。所謂粘滯連接是指讓服務(wù)消費者盡可能的
// 調(diào)用同一個服務(wù)提供者，除非該提供者掛了再進行切換
boolean sticky = invokers.get(0).getUrl().getMethodParameter(methodName, Constants.CLUSTER_STICKY_KEY, Constants.DEFAULT_CLUSTER_STICKY);
{
// 檢測 invokers 列表是否包含 stickyInvoker，如果不包含，
// 說明 stickyInvoker 代表的服務(wù)提供者掛了，此時需要將其置空
if (stickyInvoker != null && !invokers.contains(stickyInvoker)) {
stickyInvoker = null;
}
// 在 sticky 為 true，且 stickyInvoker != null 的情況下。如果 selected 包含
// stickyInvoker，表明 stickyInvoker 對應(yīng)的服務(wù)提供者可能因網(wǎng)絡(luò)原因未能成功提供服務(wù)。
// 但是該提供者并沒掛，此時 invokers 列表中仍存在該服務(wù)提供者對應(yīng)的 Invoker。
if (sticky && stickyInvoker != null && (selected == null || !selected.contains(stickyInvoker))) {
// availablecheck 表示是否開啟了可用性檢查，如果開啟了，則調(diào)用 stickyInvoker 的
// isAvailable 方法進行檢查，如果檢查通過，則直接返回 stickyInvoker。
if (availablecheck && stickyInvoker.isAvailable()) {
return stickyInvoker;
}
}
}
// 如果線程走到當(dāng)前代碼處，說明前面的 stickyInvoker 為空，或者不可用。
// 此時繼續(xù)調(diào)用 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 列表可以看做是**存活著的服務(wù)提供者**列表，如果這個列表不包含 stickyInvoker，那自然而然的認為 stickyInvoker 掛了，所以置空。如果 stickyInvoker 存在于 invokers 列表中，此時要進行下一項檢測 — 檢測 selected 中是否包含 stickyInvoker。如果包含的話，說明 stickyInvoker 在此之前沒有成功提供服務(wù)(但其仍然處于存活狀態(tài))。此時我們認為這個服務(wù)不可靠，不應(yīng)該在重試期間內(nèi)再次被調(diào)用，因此這個時候不會返回該 stickyInvoker。如果 selected 不包含 stickyInvoker，此時還需要進行可用性檢測，比如檢測服務(wù)提供者網(wǎng)絡(luò)連通性等。當(dāng)可用性檢測通過，才可返回 stickyInvoker，否則調(diào)用 doSelect 方法選擇 Invoker。如果 sticky 為 true，此時會將 doSelect 方法選出的 Invoker 賦值給 stickyInvoker。
以上就是 select 方法的邏輯，這段邏輯看起來不是很復(fù)雜，但是信息量比較大。不搞懂 invokers 和 selected 兩個入?yún)⒌暮x，以及粘滯連接特性，這段代碼是不容易看懂的。所以大家在閱讀這段代碼時，不要忽略了對背景知識的理解。關(guān)于 select 方法先分析這么多，繼續(xù)向下分析。
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 無法經(jīng)過可用性檢查，此時進行重選
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 不穩(wěn)定，或不可用，此時需要調(diào)用 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());
// 下面的 if-else 分支邏輯有些冗余，pull request #2826 對這段代碼進行了簡化，可以參考一下
// 根據(jù) availablecheck 進行不同的處理
if (availablecheck) {
// 遍歷 invokers 列表
for (Invoker<T> invoker : invokers) {
// 檢測可用性
if (invoker.isAvailable()) {
// 如果 selected 列表不包含當(dāng)前 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 列表不包含當(dāng)前 invoker，則將其添加到 reselectInvokers 中
if (selected == null || !selected.contains(invoker)) {
reselectInvokers.add(invoker);
}
}
if (!reselectInvokers.isEmpty()) {
// 通過負載均衡組件進行選擇
return loadbalance.select(reselectInvokers, getUrl(), invocation);
}
}
{
// 若線程走到此處，說明 reselectInvokers 集合為空，此時不會調(diào)用負載均衡組件進行篩選。
// 這里從 selected 列表中查找可用的 Invoker，并將其添加到 reselectInvokers 集合中
if (selected != null) {
for (Invoker<T> invoker : selected) {
if ((invoker.isAvailable())
&& !reselectInvokers.contains(invoker)) {
reselectInvokers.add(invoker);
}
}
}
if (!reselectInvokers.isEmpty()) {
// 再次進行選擇，并返回選擇結(jié)果
return loadbalance.select(reselectInvokers, getUrl(), invocation);
}
}
return null;
} reselect 方法總結(jié)下來其實只做了兩件事情，第一是查找可用的 Invoker，并將其添加到 reselectInvokers 集合中。第二，如果 reselectInvokers 不為空，則通過負載均衡組件再次進行選擇。其中第一件事情又可進行細分，一開始，reselect 從 invokers 列表中查找有效可用的 Invoker，若未能找到，此時再到 selected 列表中繼續(xù)查找。關(guān)于 reselect 方法就先分析到這，繼續(xù)分析其他的 Cluster Invoker。
FailbackClusterInvoker
FailbackClusterInvoker 會在調(diào)用失敗后，返回一個空結(jié)果給服務(wù)消費者。并通過定時任務(wù)對失敗的調(diào)用進行重傳，適合執(zhí)行消息通知等操作。下面來看一下它的實現(xiàn)邏輯。
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);
// 進行調(diào)用
return invoker.invoke(invocation);
} catch (Throwable e) {
// 如果調(diào)用過程中發(fā)生異常，此時僅打印錯誤日志，不拋出異常
logger.error("Failback to invoke method ...");
// 記錄調(diào)用信息
addFailed(invocation, this);
// 返回一個空結(jié)果給服務(wù)消費者
return new RpcResult();
}
}
private void addFailed(Invocation invocation, AbstractClusterInvoker<?> router) {
if (retryFuture == null) {
synchronized (this) {
if (retryFuture == null) {
// 創(chuàng)建定時任務(wù)，每隔5秒執(zhí)行一次
retryFuture = scheduledExecutorService.scheduleWithFixedDelay(new Runnable() {
@Override
public void run() {
try {
// 對失敗的調(diào)用進行重試
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 中
failed.put(invocation, router);
}
void retryFailed() {
if (failed.size() == 0) {
return;
}
// 遍歷 failed，對失敗的調(diào)用進行重試
for (Map.Entry<Invocation, AbstractClusterInvoker<?>> entry : new HashMap<Invocation, AbstractClusterInvoker<?>>(failed).entrySet()) {
Invocation invocation = entry.getKey();
Invoker<?> invoker = entry.getValue();
try {
// 再次進行調(diào)用
invoker.invoke(invocation);
// 調(diào)用成功后，從 failed 中移除 invoker
failed.remove(invocation);
} catch (Throwable e) {
// 僅打印異常，不拋出
logger.error("Failed retry to invoke method ...");
}
}
}
} 這個類主要由3個方法組成，首先是 doInvoker，該方法負責(zé)初次的遠程調(diào)用。若遠程調(diào)用失敗，則通過 addFailed 方法將調(diào)用信息存入到 failed 中，等待定時重試。addFailed 在開始階段會根據(jù) retryFuture 為空與否，來決定是否開啟定時任務(wù)。retryFailed 方法則是包含了失敗重試的邏輯，該方法會對 failed 進行遍歷，然后依次對 Invoker 進行調(diào)用。調(diào)用成功則將 Invoker 從 failed 中移除，調(diào)用失敗則忽略失敗原因。
以上就是 FailbackClusterInvoker 的執(zhí)行邏輯，不是很復(fù)雜，繼續(xù)往下看。
FailfastClusterInvoker
FailfastClusterInvoker 只會進行一次調(diào)用，失敗后立即拋出異常。適用于冪等操作，比如新增記錄。源碼如下：
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 {
// 調(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 ...");
}
}
} 如上，首先是通過 select 方法選擇 Invoker，然后進行遠程調(diào)用。如果調(diào)用失敗，則立即拋出異常。FailfastClusterInvoker 就先分析到這，下面分析 FailsafeClusterInvoker。
FailsafeClusterInvoker
FailsafeClusterInvoker 是一種失敗安全的 Cluster Invoker。所謂的失敗安全是指，當(dāng)調(diào)用過程中出現(xiàn)異常時，F(xiàn)ailsafeClusterInvoker 僅會打印異常，而不會拋出異常。適用于寫入審計日志等操作。下面分析源碼。
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);
// 進行遠程調(diào)用
return invoker.invoke(invocation);
} catch (Throwable e) {
// 打印錯誤日志，但不拋出
logger.error("Failsafe ignore exception: " + e.getMessage(), e);
// 返回空結(jié)果忽略錯誤
return new RpcResult();
}
}
} FailsafeClusterInvoker 的邏輯和 FailfastClusterInvoker 的邏輯一樣簡單，無需過多說明。繼續(xù)向下分析。
ForkingClusterInvoker
ForkingClusterInvoker 會在運行時通過線程池創(chuàng)建多個線程，并發(fā)調(diào)用多個服務(wù)提供者。只要有一個服務(wù)提供者成功返回了結(jié)果，doInvoke 方法就會立即結(jié)束運行。ForkingClusterInvoker 的應(yīng)用場景是在一些對實時性要求比較高**讀操作**(注意是讀操作，并行寫操作可能不安全)下使用，但這將會耗費更多的資源。下面來看該類的實現(xiàn)。
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>>();
// 循環(huán)選出 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 創(chuàng)建一個執(zhí)行線程
executor.execute(new Runnable() {
@Override
public void run() {
try {
// 進行遠程調(diào)用
Result result = invoker.invoke(invocation);
// 將結(jié)果存到阻塞隊列中
ref.offer(result);
} catch (Throwable e) {
int value = count.incrementAndGet();
// 僅在 value 大于等于 selected.size() 時，才將異常對象
// 放入阻塞隊列中，請大家思考一下為什么要這樣做。
if (value >= selected.size()) {
// 將異常對象存入到阻塞隊列中
ref.offer(e);
}
}
}
});
}
// ----------------------? 分割線2 ?---------------------- //
try {
// 從阻塞隊列中取出遠程調(diào)用結(jié)果
Object ret = ref.poll(timeout, TimeUnit.MILLISECONDS);
// 如果結(jié)果類型為 Throwable，則拋出異常
if (ret instanceof Throwable) {
Throwable e = (Throwable) ret;
throw new RpcException(..., "Failed to forking invoke provider ...");
}
// 返回結(jié)果
return (Result) ret;
} catch (InterruptedException e) {
throw new RpcException("Failed to forking invoke provider ...");
}
} finally {
RpcContext.getContext().clearAttachments();
}
}
} ForkingClusterInvoker 的 doInvoker 方法比較長，這里通過兩個分割線將整個方法劃分為三個邏輯塊。從方法開始到分割線1之間的代碼主要是用于選出 forks 個 Invoker，為接下來的并發(fā)調(diào)用提供輸入。分割線1和分割線2之間的邏輯通過線程池并發(fā)調(diào)用多個 Invoker，并將結(jié)果存儲在阻塞隊列中。分割線2到方法結(jié)尾之間的邏輯主要用于從阻塞隊列中獲取返回結(jié)果，并對返回結(jié)果類型進行判斷。如果為異常類型，則直接拋出，否則返回。
以上就是ForkingClusterInvoker 的 doInvoker 方法大致過程。我們在分割線1和分割線2之間的代碼上留了一個問題，問題是這樣的：為什么要在`value >= selected.size()`的情況下，才將異常對象添加到阻塞隊列中?這里來解答一下。原因是這樣的，在并行調(diào)用多個服務(wù)提供者的情況下，只要有一個服務(wù)提供者能夠成功返回結(jié)果，而其他全部失敗。此時 ForkingClusterInvoker 仍應(yīng)該返回成功的結(jié)果，而非拋出異常。在`value >= selected.size()`時將異常對象放入阻塞隊列中，可以保證異常對象不會出現(xiàn)在正常結(jié)果的前面，這樣可從阻塞隊列中優(yōu)先取出正常的結(jié)果。
關(guān)于 ForkingClusterInvoker 就先分析到這，接下來分析最后一個 Cluster Invoker。
BroadcastClusterInvoker
本章的最后，我們再來看一下 BroadcastClusterInvoker。BroadcastClusterInvoker 會逐個調(diào)用每個服務(wù)提供者，如果其中一臺報錯，在循環(huán)調(diào)用結(jié)束后，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 列表，逐個調(diào)用
for (Invoker<T> invoker : invokers) {
try {
// 進行遠程調(diào)用
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 的代碼，比較簡單，就不多說了。
這里分析了集群容錯的幾種實現(xiàn)方式。集群容錯對于 Dubbo 框架來說，是很重要的邏輯。集群模塊處于服務(wù)提供者和消費者之間，對于服務(wù)消費者來說，集群可向其屏蔽服務(wù)提供者集群的情況，使其能夠?qū)Ｐ倪M行遠程調(diào)用。除此之外，通過集群模塊，我們還可以對服務(wù)之間的調(diào)用鏈路進行編排優(yōu)化，治理服務(wù)?？偟膩碚f，對于 Dubbo 而言，集群容錯相關(guān)邏輯是非常重要的。想要對 Dubbo 有比較深的理解，集群容錯是必須要掌握的。
負載均衡
在之前章節(jié)中，介紹了服務(wù)集群的調(diào)用方式。我們發(fā)現(xiàn)在多服務(wù)實例時，負載均衡調(diào)用是其中極其重要的一環(huán)。在本章節(jié)中，我們一起學(xué)習(xí)Dubbo中的各種負載均衡策略
負載均衡的主要作用
負載均衡(LoadBalance)，它的職責(zé)是將網(wǎng)絡(luò)請求，或者其他形式的負載“均攤”到不同的機器上。避免集群中部分服務(wù)器壓力過大，而另一些服務(wù)器比較空閑的情況。通過負載均衡，可以讓每臺服務(wù)器獲取到適合自己處理能力的負載。在為高負載服務(wù)器分流的同時，還可以避免資源浪費，一舉兩得。
在 Dubbo 中，也有負載均衡的概念和相應(yīng)的實現(xiàn)。Dubbo 需要對服務(wù)消費者的調(diào)用請求進行分配，避免少數(shù)服務(wù)提供者負載過大。服務(wù)提供者負載過大，會導(dǎo)致部分請求超時。因此將負載均衡到每個服務(wù)提供者上，是非常必要的。
內(nèi)置的負載均衡策略
Dubbo 提供了4種負載均衡實現(xiàn)，分別是基于權(quán)重隨機算法的 RandomLoadBalance、基于最少活躍調(diào)用數(shù)算法的 LeastActiveLoadBalance、基于 hash 一致性的 ConsistentHashLoadBalance，以及基于加權(quán)輪詢算法的 RoundRobinLoadBalance。這幾個負載均衡算法代碼不是很長，但是想看懂也不是很容易，需要大家對這幾個算法的原理有一定了解才行。如果不是很了解，也沒不用太擔(dān)心。我們會在分析每個算法的源碼之前，對算法原理進行簡單的講解，幫助大家建立初步的印象。
RandomLoadBalance
RandomLoadBalance 是加權(quán)隨機算法的具體實現(xiàn)，它的算法思想很簡單。假設(shè)我們有一組服務(wù)器 servers = [A, B, C]，他們對應(yīng)的權(quán)重為 weights = [5, 3, 2]，權(quán)重總和為10?，F(xiàn)在把這些權(quán)重值平鋪在一維坐標(biāo)值上，[0, 5) 區(qū)間屬于服務(wù)器 A，[5, 8) 區(qū)間屬于服務(wù)器 B，[8, 10) 區(qū)間屬于服務(wù)器 C。接下來通過隨機數(shù)生成器生成一個范圍在 [0, 10) 之間的隨機數(shù)，然后計算這個隨機數(shù)會落到哪個區(qū)間上。比如數(shù)字3會落到服務(wù)器 A 對應(yīng)的區(qū)間上，此時返回服務(wù)器 A 即可。權(quán)重越大的機器，在坐標(biāo)軸上對應(yīng)的區(qū)間范圍就越大，因此隨機數(shù)生成器生成的數(shù)字就會有更大的概率落到此區(qū)間內(nèi)。只要隨機數(shù)生成器產(chǎn)生的隨機數(shù)分布性很好，在經(jīng)過多次選擇后，每個服務(wù)器被選中的次數(shù)比例接近其權(quán)重比例。
以上就是 RandomLoadBalance 背后的算法思想，比較簡單。下面開始分析源碼。
RandomLoadBalance 的算法思想比較簡單，在經(jīng)過多次請求后，能夠?qū)⒄{(diào)用請求按照權(quán)重值進行“均勻”分配。當(dāng)然 RandomLoadBalance 也存在一定的缺點，當(dāng)調(diào)用次數(shù)比較少時，Random 產(chǎn)生的隨機數(shù)可能會比較集中，此時多數(shù)請求會落到同一臺服務(wù)器上。這個缺點并不是很嚴重，多數(shù)情況下可以忽略。RandomLoadBalance 是一個簡單，高效的負載均衡實現(xiàn)，因此 Dubbo 選擇它作為缺省實現(xiàn)。
LeastActiveLoadBalance
LeastActiveLoadBalance 翻譯過來是最小活躍數(shù)負載均衡?；钴S調(diào)用數(shù)越小，表明該服務(wù)提供者效率越高，單位時間內(nèi)可處理更多的請求。此時應(yīng)優(yōu)先將請求分配給該服務(wù)提供者。在具體實現(xiàn)中，每個服務(wù)提供者對應(yīng)一個活躍數(shù) active。初始情況下，所有服務(wù)提供者活躍數(shù)均為0。每收到一個請求，活躍數(shù)加1，完成請求后則將活躍數(shù)減1。在服務(wù)運行一段時間后，性能好的服務(wù)提供者處理請求的速度更快，因此活躍數(shù)下降的也越快，此時這樣的服務(wù)提供者能夠優(yōu)先獲取到新的服務(wù)請求、這就是最小活躍數(shù)負載均衡算法的基本思想。關(guān)于 LeastActiveLoadBalance 的背景知識就先介紹到這里，下面開始分析源碼。
public class RandomLoadBalance extends AbstractLoadBalance {
public static final String NAME = "random";
private final Random random = new Random();
@Override
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
int length = invokers.size();
int totalWeight = 0;
boolean sameWeight = true;
// 下面這個循環(huán)有兩個作用，第一是計算總權(quán)重 totalWeight，
// 第二是檢測每個服務(wù)提供者的權(quán)重是否相同
for (int i = 0; i < length; i++) {
int weight = getWeight(invokers.get(i), invocation);
// 累加權(quán)重
totalWeight += weight;
// 檢測當(dāng)前服務(wù)提供者的權(quán)重與上一個服務(wù)提供者的權(quán)重是否相同，
// 不相同的話，則將 sameWeight 置為 false。
if (sameWeight && i > 0
&& weight != getWeight(invokers.get(i - 1), invocation)) {
sameWeight = false;
}
}
// 下面的 if 分支主要用于獲取隨機數(shù)，并計算隨機數(shù)落在哪個區(qū)間上
if (totalWeight > 0 && !sameWeight) {
// 隨機獲取一個 [0, totalWeight) 區(qū)間內(nèi)的數(shù)字
int offset = random.nextInt(totalWeight);
// 循環(huán)讓 offset 數(shù)減去服務(wù)提供者權(quán)重值，當(dāng) offset 小于0時，返回相應(yīng)的 Invoker。
// 舉例說明一下，我們有 servers = [A, B, C]，weights = [5, 3, 2]，offset = 7。
// 第一次循環(huán)，offset - 5 = 2 > 0，即 offset > 5，
// 表明其不會落在服務(wù)器 A 對應(yīng)的區(qū)間上。
// 第二次循環(huán)，offset - 3 = -1 < 0，即 5 < offset < 8，
// 表明其會落在服務(wù)器 B 對應(yīng)的區(qū)間上
for (int i = 0; i < length; i++) {
// 讓隨機值 offset 減去權(quán)重值
offset -= getWeight(invokers.get(i), invocation);
if (offset < 0) {
// 返回相應(yīng)的 Invoker
return invokers.get(i);
}
}
}
// 如果所有服務(wù)提供者權(quán)重值相同，此時直接隨機返回一個即可
return invokers.get(random.nextInt(length));
}
} 除了最小活躍數(shù)，LeastActiveLoadBalance 在實現(xiàn)上還引入了權(quán)重值。所以準(zhǔn)確的來說，LeastActiveLoadBalance 是基于加權(quán)最小活躍數(shù)算法實現(xiàn)的。舉個例子說明一下，在一個服務(wù)提供者集群中，有兩個性能優(yōu)異的服務(wù)提供者。某一時刻它們的活躍數(shù)相同，此時 Dubbo 會根據(jù)它們的權(quán)重去分配請求，權(quán)重越大，獲取到新請求的概率就越大。如果兩個服務(wù)提供者權(quán)重相同，此時隨機選擇一個即可。
ConsistentHashLoadBalance
一致性 hash 算法由麻省理工學(xué)院的 Karger 及其合作者于1997年提出的，算法提出之初是用于大規(guī)模緩存系統(tǒng)的負載均衡。它的工作過程是這樣的，首先根據(jù) ip 或者其他的信息為緩存節(jié)點生成一個 hash，并將這個 hash 投射到 [0, 2^32-1] 的圓環(huán)上。當(dāng)有查詢或?qū)懭胝埱髸r，則為緩存項的 key 生成一個 hash 值。然后查找第一個大于或等于該 hash 值的緩存節(jié)點，并到這個節(jié)點中查詢或?qū)懭刖彺骓棥Ｈ绻?dāng)前節(jié)點掛了，則在下一次查詢或?qū)懭刖彺鏁r，為緩存項查找另一個大于其 hash 值的緩存節(jié)點即可。大致效果如下圖所示，每個緩存節(jié)點在圓環(huán)上占據(jù)一個位置。如果緩存項的 key 的 hash 值小于緩存節(jié)點 hash 值，則到該緩存節(jié)點中存儲或讀取緩存項。比如下面綠色點對應(yīng)的緩存項將會被存儲到 cache-2 節(jié)點中。由于 cache-3 掛了，原本應(yīng)該存到該節(jié)點中的緩存項最終會存儲到 cache-4 節(jié)點中。
下面來看看一致性 hash 在 Dubbo 中的應(yīng)用。我們把上圖的緩存節(jié)點替換成 Dubbo 的服務(wù)提供者，于是得到了下圖：
這里相同顏色的節(jié)點均屬于同一個服務(wù)提供者，比如 Invoker1-1，Invoker1-2，……, Invoker1-160。這樣做的目的是通過引入虛擬節(jié)點，讓 Invoker 在圓環(huán)上分散開來，避免數(shù)據(jù)傾斜問題。所謂數(shù)據(jù)傾斜是指，由于節(jié)點不夠分散，導(dǎo)致大量請求落到了同一個節(jié)點上，而其他節(jié)點只會接收到了少量請求的情況。比如：
如上，由于 Invoker-1 和 Invoker-2 在圓環(huán)上分布不均，導(dǎo)致系統(tǒng)中75%的請求都會落到 Invoker-1 上，只有 25% 的請求會落到 Invoker-2 上。解決這個問題辦法是引入虛擬節(jié)點，通過虛擬節(jié)點均衡各個節(jié)點的請求量。
到這里背景知識就普及完了，接下來開始分析源碼。我們先從 ConsistentHashLoadBalance 的 doSelect 方法開始看起，如下：
public class ConsistentHashLoadBalance extends AbstractLoadBalance {
private final ConcurrentMap<String, ConsistentHashSelector<?>> selectors =
new ConcurrentHashMap<String, ConsistentHashSelector<?>>();
@Override
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
String methodName = RpcUtils.getMethodName(invocation);
String key = invokers.get(0).getUrl().getServiceKey() + "." + methodName;
// 獲取 invokers 原始的 hashcode
int identityHashCode = System.identityHashCode(invokers);
ConsistentHashSelector<T> selector = (ConsistentHashSelector<T>) selectors.get(key);
// 如果 invokers 是一個新的 List 對象，意味著服務(wù)提供者數(shù)量發(fā)生了變化，可能新增也可能減少了。
// 此時 selector.identityHashCode != identityHashCode 條件成立
if (selector == null || selector.identityHashCode != identityHashCode) {
// 創(chuàng)建新的 ConsistentHashSelector
selectors.put(key, new ConsistentHashSelector<T>(invokers, methodName, identityHashCode));
selector = (ConsistentHashSelector<T>) selectors.get(key);
}
// 調(diào)用 ConsistentHashSelector 的 select 方法選擇 Invoker
return selector.select(invocation);
}
private static final class ConsistentHashSelector<T> {...}
}如上，doSelect 方法主要做了一些前置工作，比如檢測 invokers 列表是不是變動過，以及創(chuàng)建 ConsistentHashSelector。這些工作做完后，接下來開始調(diào)用 ConsistentHashSelector 的 select 方法執(zhí)行負載均衡邏輯。在分析 select 方法之前，我們先來看一下一致性 hash 選擇器 ConsistentHashSelector 的初始化過程，如下：
private static final class ConsistentHashSelector<T> {
// 使用 TreeMap 存儲 Invoker 虛擬節(jié)點
private final TreeMap<Long, Invoker<T>> virtualInvokers;
private final int replicaNumber;
private final int identityHashCode;
private final int[] argumentIndex;
ConsistentHashSelector(List<Invoker<T>> invokers, String methodName, int identityHashCode) {
this.virtualInvokers = new TreeMap<Long, Invoker<T>>();
this.identityHashCode = identityHashCode;
URL url = invokers.get(0).getUrl();
// 獲取虛擬節(jié)點數(shù)，默認為160
this.replicaNumber = url.getMethodParameter(methodName, "hash.nodes", 160);
// 獲取參與 hash 計算的參數(shù)下標(biāo)值，默認對第一個參數(shù)進行 hash 運算
String[] index = Constants.COMMA_SPLIT_PATTERN.split(url.getMethodParameter(methodName, "hash.arguments", "0"));
argumentIndex = new int[index.length];
for (int i = 0; i < index.length; i++) {
argumentIndex[i] = Integer.parseInt(index[i]);
}
for (Invoker<T> invoker : invokers) {
String address = invoker.getUrl().getAddress();
for (int i = 0; i < replicaNumber / 4; i++) {
// 對 address + i 進行 md5 運算，得到一個長度為16的字節(jié)數(shù)組
byte[] digest = md5(address + i);
// 對 digest 部分字節(jié)進行4次 hash 運算，得到四個不同的 long 型正整數(shù)
for (int h = 0; h < 4; h++) {
// h = 0 時，取 digest 中下標(biāo)為 0 ~ 3 的4個字節(jié)進行位運算
// h = 1 時，取 digest 中下標(biāo)為 4 ~ 7 的4個字節(jié)進行位運算
// h = 2, h = 3 時過程同上
long m = hash(digest, h);
// 將 hash 到 invoker 的映射關(guān)系存儲到 virtualInvokers 中，
// virtualInvokers 需要提供高效的查詢操作，因此選用 TreeMap 作為存儲結(jié)構(gòu)
virtualInvokers.put(m, invoker);
}
}
}
}
} ConsistentHashSelector 的構(gòu)造方法執(zhí)行了一系列的初始化邏輯，比如從配置中獲取虛擬節(jié)點數(shù)以及參與 hash 計算的參數(shù)下標(biāo)，默認情況下只使用第一個參數(shù)進行 hash。需要特別說明的是，ConsistentHashLoadBalance 的負載均衡邏輯只受參數(shù)值影響，具有相同參數(shù)值的請求將會被分配給同一個服務(wù)提供者。ConsistentHashLoadBalance 不 關(guān)系權(quán)重，因此使用時需要注意一下。
在獲取虛擬節(jié)點數(shù)和參數(shù)下標(biāo)配置后，接下來要做的事情是計算虛擬節(jié)點 hash 值，并將虛擬節(jié)點存儲到 TreeMap 中。到此，ConsistentHashSelector 初始化工作就完成了。接下來，我們來看看 select 方法的邏輯。
public Invoker<T> select(Invocation invocation) {
// 將參數(shù)轉(zhuǎn)為 key
String key = toKey(invocation.getArguments());
// 對參數(shù) key 進行 md5 運算
byte[] digest = md5(key);
// 取 digest 數(shù)組的前四個字節(jié)進行 hash 運算，再將 hash 值傳給 selectForKey 方法，
// 尋找合適的 Invoker
return selectForKey(hash(digest, 0));
}
private Invoker<T> selectForKey(long hash) {
// 到 TreeMap 中查找第一個節(jié)點值大于或等于當(dāng)前 hash 的 Invoker
Map.Entry<Long, Invoker<T>> entry = virtualInvokers.tailMap(hash, true).firstEntry();
// 如果 hash 大于 Invoker 在圓環(huán)上最大的位置，此時 entry = null，
// 需要將 TreeMap 的頭節(jié)點賦值給 entry
if (entry == null) {
entry = virtualInvokers.firstEntry();
}
// 返回 Invoker
return entry.getValue();
} 如上，選擇的過程相對比較簡單了。首先是對參數(shù)進行 md5 以及 hash 運算，得到一個 hash 值。然后再拿這個值到 TreeMap 中查找目標(biāo) Invoker 即可。
RoundRobinLoadBalance
LeastActiveLoadBalance 即加權(quán)輪詢負載均衡，我們先來了解一下什么是加權(quán)輪詢。這里從最簡單的輪詢開始講起，所謂輪詢是指將請求輪流分配給每臺服務(wù)器。舉個例子，我們有三臺服務(wù)器 A、B、C。我們將第一個請求分配給服務(wù)器 A，第二個請求分配給服務(wù)器 B，第三個請求分配給服務(wù)器 C，第四個請求再次分配給服務(wù)器 A。這個過程就叫做輪詢。輪詢是一種無狀態(tài)負載均衡算法，實現(xiàn)簡單，適用于每臺服務(wù)器性能相近的場景下。但現(xiàn)實情況下，我們并不能保證每臺服務(wù)器性能均相近。如果我們將等量的請求分配給性能較差的服務(wù)器，這顯然是不合理的。因此，這個時候我們需要對輪詢過程進行加權(quán)，以調(diào)控每臺服務(wù)器的負載。經(jīng)過加權(quán)后，每臺服務(wù)器能夠得到的請求數(shù)比例，接近或等于他們的權(quán)重比。比如服務(wù)器 A、B、C 權(quán)重比為 5:2:1。那么在8次請求中，服務(wù)器 A 將收到其中的5次請求，服務(wù)器 B 會收到其中的2次請求，服務(wù)器 C 則收到其中的1次請求。
public class RoundRobinLoadBalance extends AbstractLoadBalance {
public static final String NAME = "roundrobin";
private static int RECYCLE_PERIOD = 60000;
protected static class WeightedRoundRobin {
// 服務(wù)提供者權(quán)重
private int weight;
// 當(dāng)前權(quán)重
private AtomicLong current = new AtomicLong(0);
// 最后一次更新時間
private long lastUpdate;
public void setWeight(int weight) {
this.weight = weight;
// 初始情況下，current = 0
current.set(0);
}
public long increaseCurrent() {
// current = current + weight；
return current.addAndGet(weight);
}
public void sel(int total) {
// current = current - total;
current.addAndGet(-1 * total);
}
}
// 嵌套 Map 結(jié)構(gòu)，存儲的數(shù)據(jù)結(jié)構(gòu)示例如下：
// {
// "UserService.query": {
// "url1": WeightedRoundRobin@123,
// "url2": WeightedRoundRobin@456,
// },
// "UserService.update": {
// "url1": WeightedRoundRobin@123,
// "url2": WeightedRoundRobin@456,
// }
// }
// 最外層為服務(wù)類名 + 方法名，第二層為 url 到 WeightedRoundRobin 的映射關(guān)系。
// 這里我們可以將 url 看成是服務(wù)提供者的 id
private ConcurrentMap<String, ConcurrentMap<String, WeightedRoundRobin>> methodWeightMap = new ConcurrentHashMap<String, ConcurrentMap<String, WeightedRoundRobin>>();
// 原子更新鎖
private AtomicBoolean updateLock = new AtomicBoolean();
@Override
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();
// 獲取 url 到 WeightedRoundRobin 映射表，如果為空，則創(chuàng)建一個新的
ConcurrentMap<String, WeightedRoundRobin> map = methodWeightMap.get(key);
if (map == null) {
methodWeightMap.putIfAbsent(key, new ConcurrentHashMap<String, WeightedRoundRobin>());
map = methodWeightMap.get(key);
}
int totalWeight = 0;
long maxCurrent = Long.MIN_VALUE;
// 獲取當(dāng)前時間
long now = System.currentTimeMillis();
Invoker<T> selectedInvoker = null;
WeightedRoundRobin selectedWRR = null;
// 下面這個循環(huán)主要做了這樣幾件事情：
// 1. 遍歷 Invoker 列表，檢測當(dāng)前 Invoker 是否有
// 相應(yīng)的 WeightedRoundRobin，沒有則創(chuàng)建
// 2. 檢測 Invoker 權(quán)重是否發(fā)生了變化，若變化了，
// 則更新 WeightedRoundRobin 的 weight 字段
// 3. 讓 current 字段加上自身權(quán)重，等價于 current += weight
// 4. 設(shè)置 lastUpdate 字段，即 lastUpdate = now
// 5. 尋找具有最大 current 的 Invoker，以及 Invoker 對應(yīng)的 WeightedRoundRobin，
// 暫存起來，留作后用
// 6. 計算權(quán)重總和
for (Invoker<T> invoker : invokers) {
String identifyString = invoker.getUrl().toIdentityString();
WeightedRoundRobin weightedRoundRobin = map.get(identifyString);
int weight = getWeight(invoker, invocation);
if (weight < 0) {
weight = 0;
}
// 檢測當(dāng)前 Invoker 是否有對應(yīng)的 WeightedRoundRobin，沒有則創(chuàng)建
if (weightedRoundRobin == null) {
weightedRoundRobin = new WeightedRoundRobin();
// 設(shè)置 Invoker 權(quán)重
weightedRoundRobin.setWeight(weight);
// 存儲 url 唯一標(biāo)識 identifyString 到 weightedRoundRobin 的映射關(guān)系
map.putIfAbsent(identifyString, weightedRoundRobin);
weightedRoundRobin = map.get(identifyString);
}
// Invoker 權(quán)重不等于 WeightedRoundRobin 中保存的權(quán)重，說明權(quán)重變化了，此時進行更新
if (weight != weightedRoundRobin.getWeight()) {
weightedRoundRobin.setWeight(weight);
}
// 讓 current 加上自身權(quán)重，等價于 current += weight
long cur = weightedRoundRobin.increaseCurrent();
// 設(shè)置 lastUpdate，表示近期更新過
weightedRoundRobin.setLastUpdate(now);
// 找出最大的 current
if (cur > maxCurrent) {
maxCurrent = cur;
// 將具有最大 current 權(quán)重的 Invoker 賦值給 selectedInvoker
selectedInvoker = invoker;
// 將 Invoker 對應(yīng)的 weightedRoundRobin 賦值給 selectedWRR，留作后用
selectedWRR = weightedRoundRobin;
}
// 計算權(quán)重總和
totalWeight += weight;
}
// 對 <identifyString, WeightedRoundRobin> 進行檢查，過濾掉長時間未被更新的節(jié)點。
// 該節(jié)點可能掛了，invokers 中不包含該節(jié)點，所以該節(jié)點的 lastUpdate 長時間無法被更新。
// 若未更新時長超過閾值后，就會被移除掉，默認閾值為60秒。
if (!updateLock.get() && invokers.size() != map.size()) {
if (updateLock.compareAndSet(false, true)) {
try {
ConcurrentMap<String, WeightedRoundRobin> newMap = new ConcurrentHashMap<String, WeightedRoundRobin>();
// 拷貝
newMap.putAll(map);
// 遍歷修改，即移除過期記錄
Iterator<Entry<String, WeightedRoundRobin>> it = newMap.entrySet().iterator();
while (it.hasNext()) {
Entry<String, WeightedRoundRobin> item = it.next();
if (now - item.getValue().getLastUpdate() > RECYCLE_PERIOD) {
it.remove();
}
}
// 更新引用
methodWeightMap.put(key, newMap);
} finally {
updateLock.set(false);
}
}
}
if (selectedInvoker != null) {
// 讓 current 減去權(quán)重總和，等價于 current -= totalWeight
selectedWRR.sel(totalWeight);
// 返回具有最大 current 的 Invoker
return selectedInvoker;
}
// should not happen here
return invokers.get(0);
}
} 輪詢調(diào)用并不是簡單的一個接著一個依次調(diào)用，它是根據(jù)權(quán)重的值進行循環(huán)的。
負載均衡總結(jié)
Dubbo 負載均衡策略提供下列四種方式：
Random LoadBalance 隨機，按權(quán)重設(shè)置隨機概率。Dubbo的默認負載均衡策略
在一個截面上碰撞的概率高，但調(diào)用量越大分布越均勻，而且按概率使用權(quán)重后也比較均勻，有利于動態(tài)調(diào)整提供者權(quán)重。
RoundRobin LoadBalance 輪循，按公約后的權(quán)重設(shè)置輪循比率。
存在慢的提供者累積請求問題，比如：第二臺機器很慢，但沒掛，當(dāng)請求調(diào)到第二臺時就卡在那，久而久之，所有請求都卡在調(diào)到第二臺上。
LeastActive LoadBalance 最少活躍調(diào)用數(shù)，相同活躍數(shù)的隨機，活躍數(shù)指調(diào)用前后計數(shù)差。
使慢的提供者收到更少請求，因為越慢的提供者的調(diào)用前后計數(shù)差會越大。
ConsistentHash LoadBalance 一致性Hash，相同參數(shù)的請求總是發(fā)到同一提供者。
當(dāng)某一臺提供者掛時，原本發(fā)往該提供者的請求，基于虛擬節(jié)點，平攤到其它提供者，不會引起劇烈變動。
服務(wù)治理
服務(wù)治理的概述
服務(wù)治理主要作用是改變運行時服務(wù)的行為和選址邏輯，達到限流，權(quán)重配置等目的，主要有：標(biāo)簽路由，條件路由，黑白名單，動態(tài)配置，權(quán)重調(diào)節(jié)，負載均衡等功能。
執(zhí)行過程
1、消費者，提供者啟動成功，訂閱zookeeper節(jié)點
2、管理平臺對服務(wù)進行治理處理，向zookeeper寫入節(jié)點數(shù)據(jù)
3、寫入成功，通知消費者，提供者
4、根據(jù)不同的業(yè)務(wù)處理，在invoker調(diào)用時做出響應(yīng)的處理
相關(guān)案例
服務(wù)禁用
服務(wù)禁用：通常用于臨時踢除某臺提供者機器
configVersion: v2.7 scope: application key: demo-provider enabled: true configs: - addresses: ["192.168.191.2:20883"] side: provider parameters: disabled: true
服務(wù)降級屏蔽
服務(wù)降級，當(dāng)服務(wù)器壓力劇增的情況下，根據(jù)當(dāng)前業(yè)務(wù)情況及流量對一些服務(wù)和頁面有策略的降級，以此釋放服務(wù)器資源以保證核心任務(wù)的正常運行。
容錯：當(dāng)系統(tǒng)出現(xiàn)非業(yè)務(wù)異常(比如并發(fā)數(shù)太高導(dǎo)致超時，網(wǎng)絡(luò)異常等)時，不對該接口進行處理。
屏蔽：在大促，促銷活動的可預(yù)知情況下，例如雙11活動。采用直接屏蔽接口訪問
configVersion: v2.7 scope: service key: org.apache.dubbo.samples.governance.api.DemoService enabled: true configs: - side: consumer parameters: force: return 12345
科列表頁面右側(cè)廣告圖.jpg)
科列表頁面右側(cè)廣告圖230x148.jpg)
據(jù)學(xué)科列表頁右側(cè).png)
科列表頁面右側(cè)廣告圖.jpg)
.jpg)