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