更新時間:2022-09-16 來源:黑馬程序員 瀏覽量:
服務集群的概述
概述
為了避免單點故障，現在的應用通常至少會部署在兩臺服務器上，這樣就組成了集群。集群就是單機的多實例，在多個服務器上部署多個服務，每個服務就是一個節點，部署N個節點，處理業務的能力就提升 N倍(大約)，這些節點的集合就叫做集群。
管理控制臺
目前的管理控制臺已經發布0.1版本，結構上采取了前后端分離的方式，前端使用Vue和Vuetify分別作為Javascript框架和UI框架，后端采用Spring Boot框架。既可以按照標準的Maven方式進行打包，部署，也可以采用前后端分離的部署方式，方便開發，功能上，目前具備了服務查詢，服務治理(包括Dubbo2.7中新增的治理規則)以及服務測試三部分內容。
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/)
- 前后端分離模式下，前端的修改可以實時生效
環境搭建
略
集群調用存在的問題
負載均衡
集群容錯
服務治理
集群的調用過程
調用過程
在對集群相關代碼進行分析之前，這里有必要先來介紹一下集群容錯的所有組件。包含 Cluster、Cluster Invoker、Directory、Router 和 LoadBalance 等。
集群工作過程可分為兩個階段，第一個階段是在服務消費者初始化期間，集群 Cluster 實現類為服務消費者創建 Cluster Invoker 實例，即上圖中的 merge 操作。第二個階段是在服務消費者進行遠程調用時。以 FailoverClusterInvoker 為例，該類型 Cluster Invoker 首先會調用 Directory 的 list 方法列舉 Invoker 列表(可將 Invoker 簡單理解為服務提供者)。Directory 的用途是保存 Invoker，可簡單類比為 List。其實現類 RegistryDirectory 是一個動態服務目錄，可感知注冊中心配置的變化，它所持有的 Invoker 列表會隨著注冊中心內容的變化而變化。每次變化后，RegistryDirectory 會動態增刪 Invoker，并調用 Router 的 route 方法進行路由，過濾掉不符合路由規則的 Invoker。當 FailoverClusterInvoker 拿到 Directory 返回的 Invoker 列表后，它會通過 LoadBalance 從 Invoker 列表中選擇一個 Invoker。最后 FailoverClusterInvoker 會將參數傳給 LoadBalance 選擇出的 Invoker 實例的 invoke 方法，進行真正的遠程調用。
組件介紹
Directory：它代表多個Invoker，從methodInvokerMap提取，但是他的值是動態，例如注冊中心的變更。
Router：負責從多個Invoker中按路由規則選出子集，例如應用隔離或讀寫分離或灰度發布等等
Cluster：將Directory中的多個Invoker偽裝成一個Invoker,來容錯，調用失敗重試。
LoadBalance：從多個Invoker選取一個做本次調用，具體包含很多種負載均衡算法。
Invoker：Provider中的一個可調用接口。例如DemoService
集群容錯
在分布式系統中，集群某個某些節點出現問題是大概率事件，因此在設計分布式RPC框架的過程中，必須要把失敗作為設計的一等公民來對待。一次調用失敗之后，應該如何選擇對失敗的選擇策略，這是一個見仁見智的問題，每種策略可能都有自己獨特的應用場景。因此，作為框架來說，應當針對不同場景提供多種策略，供用戶進行選擇。
在Dubbo設計中，通過Cluster這個接口的抽象，把一組可供調用的Provider信息組合成為一個統一的`Invoker`供調用方進行調用。經過路由規則過濾，負載均衡選址后，選中一個具體地址進行調用，如果調用失敗，則會按照集群配置的容錯策略進行容錯處理。
內置集群容錯方式
Dubbo默認內置了若干容錯策略，如果不能滿足用戶需求，則可以通過自定義容錯策略進行配置
Dubbo主要內置了如下幾種策略：
- Failover(失敗自動切換)
- Failsafe(失敗安全)
- Failfast(快速失敗)
- Failback(失敗自動恢復)
- Forking(并行調用)
- Broadcast(廣播調用)
這些名稱比較相似，概念也比較容易混淆，下面逐一進行解釋。
Failover(失敗自動切換)
`Failover`是高可用系統中的一個常用概念，服務器通常擁有主備兩套機器配置，如果主服務器出現故障，則自動切換到備服務器中，從而保證了整體的高可用性。
Dubbo也借鑒了這個思想，并且把它作為Dubbo`默認的容錯策略`。當調用出現失敗的時候，根據配置的重試次數，會自動從其他可用地址中重新選擇一個可用的地址進行調用，直到調用成功，或者是達到重試的上限位置。
Dubbo里默認配置的重試次數是2，也就是說，算上第一次調用，最多會調用3次。
其配置方法，容錯策略既可以在服務提供方配置，也可以服務調用方進行配置。而重試次數的配置則更為靈活，既可以在服務級別進行配置，也可以在方法級別進行配置。具體優先順序為：
```
服務調用方方法級配置 > 服務調用方服務級配置 > 服務提供方方法級配置 > 服務提供方服務級配置
```
以XML方式為例，具體配置方法如下：
服務提供方，服務級配置
<dubbo:service interface="org.apache.dubbo.demo.DemoService" ref="demoService" cluster="failover" retries="2" />
服務提供方，方法級配置
<dubbo:service interface="org.apache.dubbo.demo.DemoService" ref="demoService"cluster="failover"> <dubbo:method name="sayHello" retries="2" /> </dubbo:reference>
服務調用方，服務級配置
<dubbo:reference id="demoService" interface="org.apache.dubbo.demo.DemoService" cluster="failover" retries="1"/>
服務調用方，方法級配置：
<dubbo:reference id="demoService" interface="org.apache.dubbo.demo.DemoService" cluster="failover"> <dubbo:method name="sayHello" retries="3" /> </dubbo:reference>
Failover可以自動對失敗進行重試，對調用者屏蔽了失敗的細節，但是Failover策略也會帶來一些副作用：
- 重試會額外增加一下開銷，例如增加資源的使用，在高負載系統下，額外的重試可能讓系統雪上加霜。
- 重試會增加調用的響應時間。
- 某些情況下，重試甚至會造成資源的浪費。考慮一個調用場景，A->B->C，如果A處設置了超時100ms，再B->C的第一次調用完成時已經超過了100ms，但很不幸B->C失敗，這時候會進行重試，但其實這時候重試已經沒有意義，因此在A看來這次調用已經超時，A可能已經開始執行其他邏輯。
Failsafe(失敗安全)
失敗安全策略的核心是即使失敗了也不會影響整個調用流程。通常情況下用于旁路系統或流程中，它的失敗不影響核心業務的正確性。在實現上，當出現調用失敗時，會忽略此錯誤，并記錄一條日志，同時返回一個空結果，在上游看來調用是成功的。
應用場景，可以用于寫入審計日志等操作。
具體配置方法：
服務提供方，服務級配置
<dubbo:service interface="org.apache.dubbo.demo.DemoService" ref="demoService" cluster="failsafe" />
服務調用方，服務級配置
<dubbo:reference id="demoService" interface="org.apache.dubbo.demo.DemoService" cluster="failsafe"/>
其中服務調用方配置優先于服務提供方配置。
Failfast(快速失敗)
某些業務場景中，某些操作可能是非冪等的，如果重復發起調用，可能會導致出現臟數據等。例如調用某個服務，其中包含一個數據庫的寫操作，如果寫操作完成，但是在發送結果給調用方的過程中出錯了，那么在調用發看來這次調用失敗了，但其實數據寫入已經完成。這種情況下，重試可能并不是一個好策略，這時候就需要使用到`Failfast`策略，調用失敗立即報錯。讓調用方來決定下一步的操作并保證業務的冪等性。
具體配置方法：
服務提供方，服務級配置
<dubbo:service interface="org.apache.dubbo.demo.DemoService" ref="demoService" cluster="failfast" />
服務調用方，服務級配置
<dubbo:reference id="demoService" interface="org.apache.dubbo.demo.DemoService" cluster="failfast"/>
其中服務調用方配置優先于服務提供方配置。
Failback(失敗自動恢復)
`Failback`通常和`Failover`兩個概念聯系在一起。在高可用系統中，當主機發生故障，通過`Failover`進行主備切換后，待故障恢復后，系統應該具備自動恢復原始配置的能力。
Dubbo中的`Failback`策略中，如果調用失敗，則此次失敗相當于`Failsafe`，將返回一個空結果。而與`Failsafe`不同的是，Failback策略會將這次調用加入內存中的失敗列表中，對于這個列表中的失敗調用，會在另一個線程中進行異步重試，重試如果再發生失敗，則會忽略，即使重試調用成功，原來的調用方也感知不到了。因此它通常適合于，對于實時性要求不高，且不需要返回值的一些異步操作。
具體配置方法：
服務提供方，服務級配置
<dubbo:service interface="org.apache.dubbo.demo.DemoService" ref="demoService" cluster="failsafe" />
服務調用方，服務級配置
<dubbo:reference id="demoService" interface="org.apache.dubbo.demo.DemoService" cluster="failsafe"/>
其中服務調用方配置優先于服務提供方配置。
按照目前的實現，Failback策略還有一些局限，例如內存中的失敗調用列表沒有上限，可能導致堆積，異步重試的執行間隔無法調整，默認是5秒。
Forking(并行調用)
上述幾種策略中，主要都是針對調用失敗發生后如何進行彌補的角度去考慮的，而`Forking`策略則跟上述幾種策略不同，是一種典型的用成本換時間的思路。即第一次調用的時候就同時發起多個調用，只要其中一個調用成功，就認為成功。在資源充足，且對于失敗的容忍度較低的場景下，可以采用此策略。
具體配置方法：
服務提供方，服務級配置
<dubbo:service interface="org.apache.dubbo.demo.DemoService" ref="demoService" cluster="forking" />
服務調用方，服務級配置
<dubbo:reference id="demoService" interface="org.apache.dubbo.demo.DemoService" cluster="forking"/>
其中服務調用方配置優先于服務提供方配置。
Broadcast(廣播調用)
在某些場景下，可能需要對服務的所有提供者進行操作，此時可以使用廣播調用策略。此策略會逐個調用所有提供者，只要任意有一個提供者出錯，則認為此次調用出錯。通常用于通知所有提供者更新緩存或日志等本地資源信息。
具體配置方法：
服務提供方，服務級配置
<dubbo:service interface="org.apache.dubbo.demo.DemoService" ref="demoService" cluster="broadcast" />
服務調用方，服務級配置
<dubbo:reference id="demoService" interface="org.apache.dubbo.demo.DemoService" cluster="broadcast"/>
其中服務調用方配置優先于服務提供方配置。
集群容錯調優
下表對各種策略做一個簡單對比，
| 策略名稱 | 優點 | 缺點 |
| --------- | -------------------------------- | -------------------------------------- |
| Failover | 對調用者屏蔽調用失敗的信息 | 增加RT，額外資源開銷，資源浪費 |
| Failfast | 業務快速感知失敗狀態進行自主決策 | 產生較多報錯的信息 |
| Failsafe | 即使失敗了也不會影響核心流程 | 對于失敗的信息不敏感，需要額外的監控 |
| Failback | 失敗自動異步重試 | 重試任務可能堆積 |
| Forking | 并行發起多個調用，降低失敗概率 | 消耗額外的機器資源，需要確保操作冪等性 |
| Broadcast | 支持對所有的服務提供者進行操作 | 資源消耗很大 |
綜上我們得知，不同的容錯策略往往對應不同的業務處理，這里做一個總結如下：
Failover：通常用于對調用rt不敏感的場景，如讀操作;但重試會帶來更長延遲
Failfast：通常用于非冪等性操作，需要快速感知失敗的場景;比如新增記錄
Failsafe：通常用于旁路系統，失敗不影響核心流程正確性的場景;如日志記錄
Failback：通常用于對于實時性要求不高，且不需要返回值的一些異步操作的場景
Forking：通常用于資源充足，且對于失敗的容忍度較低，實時性要求高的讀操作，但需要浪費更多服務資源
Broadcast：如通知所有提供者更新緩存或日志等本地資源信息
源碼分析
我們在上一章看到了兩個概念，分別是集群接口 Cluster 和 Cluster Invoker，這兩者是不同的。Cluster 是接口，而 Cluster Invoker 是一種 Invoker。服務提供者的選擇邏輯，以及遠程調用失敗后的的處理邏輯均是封裝在 Cluster Invoker 中。那么 Cluster 接口和相關實現類有什么用呢?用途比較簡單，僅用于生成 Cluster Invoker。下面我們來看一下源碼。
JavaEE
鴻蒙應用開發
集成電路應用開發
人工智能開發
AI+Linux云計算運維
Python+大數據開發
AI+設計
軟件測試
新媒體+短視頻
HTML&JS+前端
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 上
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 方法列舉 Invoker
List<Invoker<T>> invokers = directory.list(invocation);
return invokers;
} 如上，AbstractClusterInvoker 中的 list 方法做的事情很簡單，只是簡單的調用了 Directory 的 list 方法，沒有其他更多的邏輯了。Directory 即相關實現類在前文已經分析過，這里就不多說了。接下來，我們把目光轉移到 AbstractClusterInvoker 的各種實現類上，來看一下這些實現類是如何實現 doInvoke 方法邏輯的。
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 兩個入參的含義，以及粘滯連接特性，這段代碼是不容易看懂的。所以大家在閱讀這段代碼時，不要忽略了對背景知識的理解。關于 select 方法先分析這么多，繼續向下分析。
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());
// 下面的 if-else 分支邏輯有些冗余，pull request #2826 對這段代碼進行了簡化，可以參考一下
// 根據 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。
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 中
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 的執行邏輯，不是很復雜，繼續往下看。
FailfastClusterInvoker
FailfastClusterInvoker 只會進行一次調用，失敗后立即拋出異常。適用于冪等操作，比如新增記錄。源碼如下：
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。
FailsafeClusterInvoker
FailsafeClusterInvoker 是一種失敗安全的 Cluster Invoker。所謂的失敗安全是指，當調用過程中出現異常時，FailsafeClusterInvoker 僅會打印異常，而不會拋出異常。適用于寫入審計日志等操作。下面分析源碼。
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 的邏輯一樣簡單，無需過多說明。繼續向下分析。
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。
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 的代碼，比較簡單，就不多說了。
這里分析了集群容錯的幾種實現方式。集群容錯對于 Dubbo 框架來說，是很重要的邏輯。集群模塊處于服務提供者和消費者之間，對于服務消費者來說，集群可向其屏蔽服務提供者集群的情況，使其能夠專心進行遠程調用。除此之外，通過集群模塊，我們還可以對服務之間的調用鏈路進行編排優化，治理服務。總的來說，對于 Dubbo 而言，集群容錯相關邏輯是非常重要的。想要對 Dubbo 有比較深的理解，集群容錯是必須要掌握的。
負載均衡
在之前章節中，介紹了服務集群的調用方式。我們發現在多服務實例時，負載均衡調用是其中極其重要的一環。在本章節中，我們一起學習Dubbo中的各種負載均衡策略
負載均衡的主要作用
負載均衡(LoadBalance)，它的職責是將網絡請求，或者其他形式的負載“均攤”到不同的機器上。避免集群中部分服務器壓力過大，而另一些服務器比較空閑的情況。通過負載均衡，可以讓每臺服務器獲取到適合自己處理能力的負載。在為高負載服務器分流的同時，還可以避免資源浪費，一舉兩得。
在 Dubbo 中，也有負載均衡的概念和相應的實現。Dubbo 需要對服務消費者的調用請求進行分配，避免少數服務提供者負載過大。服務提供者負載過大，會導致部分請求超時。因此將負載均衡到每個服務提供者上，是非常必要的。
內置的負載均衡策略
Dubbo 提供了4種負載均衡實現，分別是基于權重隨機算法的 RandomLoadBalance、基于最少活躍調用數算法的 LeastActiveLoadBalance、基于 hash 一致性的 ConsistentHashLoadBalance，以及基于加權輪詢算法的 RoundRobinLoadBalance。這幾個負載均衡算法代碼不是很長，但是想看懂也不是很容易，需要大家對這幾個算法的原理有一定了解才行。如果不是很了解，也沒不用太擔心。我們會在分析每個算法的源碼之前，對算法原理進行簡單的講解，幫助大家建立初步的印象。
RandomLoadBalance
RandomLoadBalance 是加權隨機算法的具體實現，它的算法思想很簡單。假設我們有一組服務器 servers = [A, B, C]，他們對應的權重為 weights = [5, 3, 2]，權重總和為10。現在把這些權重值平鋪在一維坐標值上，[0, 5) 區間屬于服務器 A，[5, 8) 區間屬于服務器 B，[8, 10) 區間屬于服務器 C。接下來通過隨機數生成器生成一個范圍在 [0, 10) 之間的隨機數，然后計算這個隨機數會落到哪個區間上。比如數字3會落到服務器 A 對應的區間上，此時返回服務器 A 即可。權重越大的機器，在坐標軸上對應的區間范圍就越大，因此隨機數生成器生成的數字就會有更大的概率落到此區間內。只要隨機數生成器產生的隨機數分布性很好，在經過多次選擇后，每個服務器被選中的次數比例接近其權重比例。
以上就是 RandomLoadBalance 背后的算法思想，比較簡單。下面開始分析源碼。
RandomLoadBalance 的算法思想比較簡單，在經過多次請求后，能夠將調用請求按照權重值進行“均勻”分配。當然 RandomLoadBalance 也存在一定的缺點，當調用次數比較少時，Random 產生的隨機數可能會比較集中，此時多數請求會落到同一臺服務器上。這個缺點并不是很嚴重，多數情況下可以忽略。RandomLoadBalance 是一個簡單，高效的負載均衡實現，因此 Dubbo 選擇它作為缺省實現。
LeastActiveLoadBalance
LeastActiveLoadBalance 翻譯過來是最小活躍數負載均衡。活躍調用數越小，表明該服務提供者效率越高，單位時間內可處理更多的請求。此時應優先將請求分配給該服務提供者。在具體實現中，每個服務提供者對應一個活躍數 active。初始情況下，所有服務提供者活躍數均為0。每收到一個請求，活躍數加1，完成請求后則將活躍數減1。在服務運行一段時間后，性能好的服務提供者處理請求的速度更快，因此活躍數下降的也越快，此時這樣的服務提供者能夠優先獲取到新的服務請求、這就是最小活躍數負載均衡算法的基本思想。關于 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;
// 下面這個循環有兩個作用，第一是計算總權重 totalWeight，
// 第二是檢測每個服務提供者的權重是否相同
for (int i = 0; i < length; i++) {
int weight = getWeight(invokers.get(i), invocation);
// 累加權重
totalWeight += weight;
// 檢測當前服務提供者的權重與上一個服務提供者的權重是否相同，
// 不相同的話，則將 sameWeight 置為 false。
if (sameWeight && i > 0
&& weight != getWeight(invokers.get(i - 1), invocation)) {
sameWeight = false;
}
}
// 下面的 if 分支主要用于獲取隨機數，并計算隨機數落在哪個區間上
if (totalWeight > 0 && !sameWeight) {
// 隨機獲取一個 [0, totalWeight) 區間內的數字
int offset = random.nextInt(totalWeight);
// 循環讓 offset 數減去服務提供者權重值，當 offset 小于0時，返回相應的 Invoker。
// 舉例說明一下，我們有 servers = [A, B, C]，weights = [5, 3, 2]，offset = 7。
// 第一次循環，offset - 5 = 2 > 0，即 offset > 5，
// 表明其不會落在服務器 A 對應的區間上。
// 第二次循環，offset - 3 = -1 < 0，即 5 < offset < 8，
// 表明其會落在服務器 B 對應的區間上
for (int i = 0; i < length; i++) {
// 讓隨機值 offset 減去權重值
offset -= getWeight(invokers.get(i), invocation);
if (offset < 0) {
// 返回相應的 Invoker
return invokers.get(i);
}
}
}
// 如果所有服務提供者權重值相同，此時直接隨機返回一個即可
return invokers.get(random.nextInt(length));
}
} 除了最小活躍數，LeastActiveLoadBalance 在實現上還引入了權重值。所以準確的來說，LeastActiveLoadBalance 是基于加權最小活躍數算法實現的。舉個例子說明一下，在一個服務提供者集群中，有兩個性能優異的服務提供者。某一時刻它們的活躍數相同，此時 Dubbo 會根據它們的權重去分配請求，權重越大，獲取到新請求的概率就越大。如果兩個服務提供者權重相同，此時隨機選擇一個即可。
ConsistentHashLoadBalance
一致性 hash 算法由麻省理工學院的 Karger 及其合作者于1997年提出的，算法提出之初是用于大規模緩存系統的負載均衡。它的工作過程是這樣的，首先根據 ip 或者其他的信息為緩存節點生成一個 hash，并將這個 hash 投射到 [0, 2^32-1] 的圓環上。當有查詢或寫入請求時，則為緩存項的 key 生成一個 hash 值。然后查找第一個大于或等于該 hash 值的緩存節點，并到這個節點中查詢或寫入緩存項。如果當前節點掛了，則在下一次查詢或寫入緩存時，為緩存項查找另一個大于其 hash 值的緩存節點即可。大致效果如下圖所示，每個緩存節點在圓環上占據一個位置。如果緩存項的 key 的 hash 值小于緩存節點 hash 值，則到該緩存節點中存儲或讀取緩存項。比如下面綠色點對應的緩存項將會被存儲到 cache-2 節點中。由于 cache-3 掛了，原本應該存到該節點中的緩存項最終會存儲到 cache-4 節點中。
下面來看看一致性 hash 在 Dubbo 中的應用。我們把上圖的緩存節點替換成 Dubbo 的服務提供者，于是得到了下圖：
這里相同顏色的節點均屬于同一個服務提供者，比如 Invoker1-1，Invoker1-2，……, Invoker1-160。這樣做的目的是通過引入虛擬節點，讓 Invoker 在圓環上分散開來，避免數據傾斜問題。所謂數據傾斜是指，由于節點不夠分散，導致大量請求落到了同一個節點上，而其他節點只會接收到了少量請求的情況。比如：
如上，由于 Invoker-1 和 Invoker-2 在圓環上分布不均，導致系統中75%的請求都會落到 Invoker-1 上，只有 25% 的請求會落到 Invoker-2 上。解決這個問題辦法是引入虛擬節點，通過虛擬節點均衡各個節點的請求量。
到這里背景知識就普及完了，接下來開始分析源碼。我們先從 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 對象，意味著服務提供者數量發生了變化，可能新增也可能減少了。
// 此時 selector.identityHashCode != identityHashCode 條件成立
if (selector == null || selector.identityHashCode != identityHashCode) {
// 創建新的 ConsistentHashSelector
selectors.put(key, new ConsistentHashSelector<T>(invokers, methodName, identityHashCode));
selector = (ConsistentHashSelector<T>) selectors.get(key);
}
// 調用 ConsistentHashSelector 的 select 方法選擇 Invoker
return selector.select(invocation);
}
private static final class ConsistentHashSelector<T> {...}
}如上，doSelect 方法主要做了一些前置工作，比如檢測 invokers 列表是不是變動過，以及創建 ConsistentHashSelector。這些工作做完后，接下來開始調用 ConsistentHashSelector 的 select 方法執行負載均衡邏輯。在分析 select 方法之前，我們先來看一下一致性 hash 選擇器 ConsistentHashSelector 的初始化過程，如下：
private static final class ConsistentHashSelector<T> {
// 使用 TreeMap 存儲 Invoker 虛擬節點
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();
// 獲取虛擬節點數，默認為160
this.replicaNumber = url.getMethodParameter(methodName, "hash.nodes", 160);
// 獲取參與 hash 計算的參數下標值，默認對第一個參數進行 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的字節數組
byte[] digest = md5(address + i);
// 對 digest 部分字節進行4次 hash 運算，得到四個不同的 long 型正整數
for (int h = 0; h < 4; h++) {
// h = 0 時，取 digest 中下標為 0 ~ 3 的4個字節進行位運算
// h = 1 時，取 digest 中下標為 4 ~ 7 的4個字節進行位運算
// h = 2, h = 3 時過程同上
long m = hash(digest, h);
// 將 hash 到 invoker 的映射關系存儲到 virtualInvokers 中，
// virtualInvokers 需要提供高效的查詢操作，因此選用 TreeMap 作為存儲結構
virtualInvokers.put(m, invoker);
}
}
}
}
} ConsistentHashSelector 的構造方法執行了一系列的初始化邏輯，比如從配置中獲取虛擬節點數以及參與 hash 計算的參數下標，默認情況下只使用第一個參數進行 hash。需要特別說明的是，ConsistentHashLoadBalance 的負載均衡邏輯只受參數值影響，具有相同參數值的請求將會被分配給同一個服務提供者。ConsistentHashLoadBalance 不 關系權重，因此使用時需要注意一下。
在獲取虛擬節點數和參數下標配置后，接下來要做的事情是計算虛擬節點 hash 值，并將虛擬節點存儲到 TreeMap 中。到此，ConsistentHashSelector 初始化工作就完成了。接下來，我們來看看 select 方法的邏輯。
public Invoker<T> select(Invocation invocation) {
// 將參數轉為 key
String key = toKey(invocation.getArguments());
// 對參數 key 進行 md5 運算
byte[] digest = md5(key);
// 取 digest 數組的前四個字節進行 hash 運算，再將 hash 值傳給 selectForKey 方法，
// 尋找合適的 Invoker
return selectForKey(hash(digest, 0));
}
private Invoker<T> selectForKey(long hash) {
// 到 TreeMap 中查找第一個節點值大于或等于當前 hash 的 Invoker
Map.Entry<Long, Invoker<T>> entry = virtualInvokers.tailMap(hash, true).firstEntry();
// 如果 hash 大于 Invoker 在圓環上最大的位置，此時 entry = null，
// 需要將 TreeMap 的頭節點賦值給 entry
if (entry == null) {
entry = virtualInvokers.firstEntry();
}
// 返回 Invoker
return entry.getValue();
} 如上，選擇的過程相對比較簡單了。首先是對參數進行 md5 以及 hash 運算，得到一個 hash 值。然后再拿這個值到 TreeMap 中查找目標 Invoker 即可。
RoundRobinLoadBalance
LeastActiveLoadBalance 即加權輪詢負載均衡，我們先來了解一下什么是加權輪詢。這里從最簡單的輪詢開始講起，所謂輪詢是指將請求輪流分配給每臺服務器。舉個例子，我們有三臺服務器 A、B、C。我們將第一個請求分配給服務器 A，第二個請求分配給服務器 B，第三個請求分配給服務器 C，第四個請求再次分配給服務器 A。這個過程就叫做輪詢。輪詢是一種無狀態負載均衡算法，實現簡單，適用于每臺服務器性能相近的場景下。但現實情況下，我們并不能保證每臺服務器性能均相近。如果我們將等量的請求分配給性能較差的服務器，這顯然是不合理的。因此，這個時候我們需要對輪詢過程進行加權，以調控每臺服務器的負載。經過加權后，每臺服務器能夠得到的請求數比例，接近或等于他們的權重比。比如服務器 A、B、C 權重比為 5:2:1。那么在8次請求中，服務器 A 將收到其中的5次請求，服務器 B 會收到其中的2次請求，服務器 C 則收到其中的1次請求。
public class RoundRobinLoadBalance extends AbstractLoadBalance {
public static final String NAME = "roundrobin";
private static int RECYCLE_PERIOD = 60000;
protected static class WeightedRoundRobin {
// 服務提供者權重
private int weight;
// 當前權重
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 結構，存儲的數據結構示例如下：
// {
// "UserService.query": {
// "url1": WeightedRoundRobin@123,
// "url2": WeightedRoundRobin@456,
// },
// "UserService.update": {
// "url1": WeightedRoundRobin@123,
// "url2": WeightedRoundRobin@456,
// }
// }
// 最外層為服務類名 + 方法名，第二層為 url 到 WeightedRoundRobin 的映射關系。
// 這里我們可以將 url 看成是服務提供者的 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 映射表，如果為空，則創建一個新的
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;
// 獲取當前時間
long now = System.currentTimeMillis();
Invoker<T> selectedInvoker = null;
WeightedRoundRobin selectedWRR = null;
// 下面這個循環主要做了這樣幾件事情：
// 1. 遍歷 Invoker 列表，檢測當前 Invoker 是否有
// 相應的 WeightedRoundRobin，沒有則創建
// 2. 檢測 Invoker 權重是否發生了變化，若變化了，
// 則更新 WeightedRoundRobin 的 weight 字段
// 3. 讓 current 字段加上自身權重，等價于 current += weight
// 4. 設置 lastUpdate 字段，即 lastUpdate = now
// 5. 尋找具有最大 current 的 Invoker，以及 Invoker 對應的 WeightedRoundRobin，
// 暫存起來，留作后用
// 6. 計算權重總和
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;
}
// 檢測當前 Invoker 是否有對應的 WeightedRoundRobin，沒有則創建
if (weightedRoundRobin == null) {
weightedRoundRobin = new WeightedRoundRobin();
// 設置 Invoker 權重
weightedRoundRobin.setWeight(weight);
// 存儲 url 唯一標識 identifyString 到 weightedRoundRobin 的映射關系
map.putIfAbsent(identifyString, weightedRoundRobin);
weightedRoundRobin = map.get(identifyString);
}
// Invoker 權重不等于 WeightedRoundRobin 中保存的權重，說明權重變化了，此時進行更新
if (weight != weightedRoundRobin.getWeight()) {
weightedRoundRobin.setWeight(weight);
}
// 讓 current 加上自身權重，等價于 current += weight
long cur = weightedRoundRobin.increaseCurrent();
// 設置 lastUpdate，表示近期更新過
weightedRoundRobin.setLastUpdate(now);
// 找出最大的 current
if (cur > maxCurrent) {
maxCurrent = cur;
// 將具有最大 current 權重的 Invoker 賦值給 selectedInvoker
selectedInvoker = invoker;
// 將 Invoker 對應的 weightedRoundRobin 賦值給 selectedWRR，留作后用
selectedWRR = weightedRoundRobin;
}
// 計算權重總和
totalWeight += weight;
}
// 對 <identifyString, WeightedRoundRobin> 進行檢查，過濾掉長時間未被更新的節點。
// 該節點可能掛了，invokers 中不包含該節點，所以該節點的 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 減去權重總和，等價于 current -= totalWeight
selectedWRR.sel(totalWeight);
// 返回具有最大 current 的 Invoker
return selectedInvoker;
}
// should not happen here
return invokers.get(0);
}
} 輪詢調用并不是簡單的一個接著一個依次調用，它是根據權重的值進行循環的。
負載均衡總結
Dubbo 負載均衡策略提供下列四種方式：
Random LoadBalance 隨機，按權重設置隨機概率。Dubbo的默認負載均衡策略
在一個截面上碰撞的概率高，但調用量越大分布越均勻，而且按概率使用權重后也比較均勻，有利于動態調整提供者權重。
RoundRobin LoadBalance 輪循，按公約后的權重設置輪循比率。
存在慢的提供者累積請求問題，比如：第二臺機器很慢，但沒掛，當請求調到第二臺時就卡在那，久而久之，所有請求都卡在調到第二臺上。
LeastActive LoadBalance 最少活躍調用數，相同活躍數的隨機，活躍數指調用前后計數差。
使慢的提供者收到更少請求，因為越慢的提供者的調用前后計數差會越大。
ConsistentHash LoadBalance 一致性Hash，相同參數的請求總是發到同一提供者。
當某一臺提供者掛時，原本發往該提供者的請求，基于虛擬節點，平攤到其它提供者，不會引起劇烈變動。
服務治理
服務治理的概述
服務治理主要作用是改變運行時服務的行為和選址邏輯，達到限流，權重配置等目的，主要有：標簽路由，條件路由，黑白名單，動態配置，權重調節，負載均衡等功能。
執行過程
1、消費者，提供者啟動成功，訂閱zookeeper節點
2、管理平臺對服務進行治理處理，向zookeeper寫入節點數據
3、寫入成功，通知消費者，提供者
4、根據不同的業務處理，在invoker調用時做出響應的處理
相關案例
服務禁用
服務禁用：通常用于臨時踢除某臺提供者機器
configVersion: v2.7 scope: application key: demo-provider enabled: true configs: - addresses: ["192.168.191.2:20883"] side: provider parameters: disabled: true
服務降級屏蔽
服務降級，當服務器壓力劇增的情況下，根據當前業務情況及流量對一些服務和頁面有策略的降級，以此釋放服務器資源以保證核心任務的正常運行。
容錯：當系統出現非業務異常(比如并發數太高導致超時，網絡異常等)時，不對該接口進行處理。
屏蔽：在大促，促銷活動的可預知情況下，例如雙11活動。采用直接屏蔽接口訪問
configVersion: v2.7 scope: service key: org.apache.dubbo.samples.governance.api.DemoService enabled: true configs: - side: consumer parameters: force: return 12345
.jpg)