首頁技術(shù)文章正文

Java培訓(xùn):dubbo源碼解析-高可用集群

更新時間:2022-09-16 來源:黑馬程序員 瀏覽量:

  服務(wù)集群的概述

  概述

  為了避免單點(diǎn)故障,現(xiàn)在的應(yīng)用通常至少會部署在兩臺服務(wù)器上,這樣就組成了集群。集群就是單機(jī)的多實(shí)例,在多個服務(wù)器上部署多個服務(wù),每個服務(wù)就是一個節(jié)點(diǎn),部署N個節(jié)點(diǎn),處理業(yè)務(wù)的能力就提升 N倍(大約),這些節(jié)點(diǎn)的集合就叫做集群。

1663316185243_1.jpg

  管理控制臺

  目前的管理控制臺已經(jīng)發(fā)布0.1版本,結(jié)構(gòu)上采取了前后端分離的方式,前端使用Vue和Vuetify分別作為Javascript框架和UI框架,后端采用Spring Boot框架。既可以按照標(biāo)準(zhǔn)的Maven方式進(jìn)行打包,部署,也可以采用前后端分離的部署方式,方便開發(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/)

  - 前后端分離模式下,前端的修改可以實(shí)時生效

  環(huán)境搭建

  略

  集群調(diào)用存在的問題

  負(fù)載均衡

  集群容錯

  服務(wù)治理

  集群的調(diào)用過程

  調(diào)用過程

  在對集群相關(guān)代碼進(jìn)行分析之前,這里有必要先來介紹一下集群容錯的所有組件。包含 Cluster、Cluster Invoker、Directory、Router 和 LoadBalance 等。

  

1663316415014_2.jpg

  集群工作過程可分為兩個階段,第一個階段是在服務(wù)消費(fèi)者初始化期間,集群 Cluster 實(shí)現(xiàn)類為服務(wù)消費(fèi)者創(chuàng)建 Cluster Invoker 實(shí)例,即上圖中的 merge 操作。第二個階段是在服務(wù)消費(fèi)者進(jìn)行遠(yuǎn)程調(diào)用時。以 FailoverClusterInvoker 為例,該類型 Cluster Invoker 首先會調(diào)用 Directory 的 list 方法列舉 Invoker 列表(可將 Invoker 簡單理解為服務(wù)提供者)。Directory 的用途是保存 Invoker,可簡單類比為 List。其實(shí)現(xiàn)類 RegistryDirectory 是一個動態(tài)服務(wù)目錄,可感知注冊中心配置的變化,它所持有的 Invoker 列表會隨著注冊中心內(nèi)容的變化而變化。每次變化后,RegistryDirectory 會動態(tài)增刪 Invoker,并調(diào)用 Router 的 route 方法進(jìn)行路由,過濾掉不符合路由規(guī)則的 Invoker。當(dāng) FailoverClusterInvoker 拿到 Directory 返回的 Invoker 列表后,它會通過 LoadBalance 從 Invoker 列表中選擇一個 Invoker。最后 FailoverClusterInvoker 會將參數(shù)傳給 LoadBalance 選擇出的 Invoker 實(shí)例的 invoke 方法,進(jìn)行真正的遠(yuǎn)程調(diào)用。

  組件介紹

  Directory:它代表多個Invoker,從methodInvokerMap提取,但是他的值是動態(tài),例如注冊中心的變更。

  Router:負(fù)責(zé)從多個Invoker中按路由規(guī)則選出子集,例如應(yīng)用隔離或讀寫分離或灰度發(fā)布等等

  Cluster:將Directory中的多個Invoker偽裝成一個Invoker,來容錯,調(diào)用失敗重試。

  LoadBalance:從多個Invoker選取一個做本次調(diào)用,具體包含很多種負(fù)載均衡算法。

  Invoker:Provider中的一個可調(diào)用接口。例如DemoService

   集群容錯

  在分布式系統(tǒng)中,集群某個某些節(jié)點(diǎn)出現(xiàn)問題是大概率事件,因此在設(shè)計分布式RPC框架的過程中,必須要把失敗作為設(shè)計的一等公民來對待。一次調(diào)用失敗之后,應(yīng)該如何選擇對失敗的選擇策略,這是一個見仁見智的問題,每種策略可能都有自己獨(dú)特的應(yīng)用場景。因此,作為框架來說,應(yīng)當(dāng)針對不同場景提供多種策略,供用戶進(jìn)行選擇。

  在Dubbo設(shè)計中,通過Cluster這個接口的抽象,把一組可供調(diào)用的Provider信息組合成為一個統(tǒng)一的`Invoker`供調(diào)用方進(jìn)行調(diào)用。經(jīng)過路由規(guī)則過濾,負(fù)載均衡選址后,選中一個具體地址進(jìn)行調(diào)用,如果調(diào)用失敗,則會按照集群配置的容錯策略進(jìn)行容錯處理。

  內(nèi)置集群容錯方式

  Dubbo默認(rèn)內(nèi)置了若干容錯策略,如果不能滿足用戶需求,則可以通過自定義容錯策略進(jìn)行配置

  Dubbo主要內(nèi)置了如下幾種策略:

  - Failover(失敗自動切換)

  - Failsafe(失敗安全)

  - Failfast(快速失敗)

  - Failback(失敗自動恢復(fù))

  - Forking(并行調(diào)用)

  - Broadcast(廣播調(diào)用)

  這些名稱比較相似,概念也比較容易混淆,下面逐一進(jìn)行解釋。

  Failover(失敗自動切換)

  `Failover`是高可用系統(tǒng)中的一個常用概念,服務(wù)器通常擁有主備兩套機(jī)器配置,如果主服務(wù)器出現(xiàn)故障,則自動切換到備服務(wù)器中,從而保證了整體的高可用性。

  Dubbo也借鑒了這個思想,并且把它作為Dubbo`默認(rèn)的容錯策略`。當(dāng)調(diào)用出現(xiàn)失敗的時候,根據(jù)配置的重試次數(shù),會自動從其他可用地址中重新選擇一個可用的地址進(jìn)行調(diào)用,直到調(diào)用成功,或者是達(dá)到重試的上限位置。

  Dubbo里默認(rèn)配置的重試次數(shù)是2,也就是說,算上第一次調(diào)用,最多會調(diào)用3次。

  其配置方法,容錯策略既可以在服務(wù)提供方配置,也可以服務(wù)調(diào)用方進(jìn)行配置。而重試次數(shù)的配置則更為靈活,既可以在服務(wù)級別進(jìn)行配置,也可以在方法級別進(jìn)行配置。具體優(yōu)先順序?yàn)椋?/p>

  ```

  服務(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可以自動對失敗進(jìn)行重試,對調(diào)用者屏蔽了失敗的細(xì)節(jié),但是Failover策略也會帶來一些副作用:

  - 重試會額外增加一下開銷,例如增加資源的使用,在高負(fù)載系統(tǒng)下,額外的重試可能讓系統(tǒng)雪上加霜。

  - 重試會增加調(diào)用的響應(yīng)時間。

  - 某些情況下,重試甚至?xí)斐少Y源的浪費(fèi)??紤]一個調(diào)用場景,A->B->C,如果A處設(shè)置了超時100ms,再B->C的第一次調(diào)用完成時已經(jīng)超過了100ms,但很不幸B->C失敗,這時候會進(jìn)行重試,但其實(shí)這時候重試已經(jīng)沒有意義,因此在A看來這次調(diào)用已經(jīng)超時,A可能已經(jīng)開始執(zhí)行其他邏輯。

  Failsafe(失敗安全)

  失敗安全策略的核心是即使失敗了也不會影響整個調(diào)用流程。通常情況下用于旁路系統(tǒng)或流程中,它的失敗不影響核心業(yè)務(wù)的正確性。在實(shí)現(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í)數(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)主機(jī)發(fā)生故障,通過`Failover`進(jìn)行主備切換后,待故障恢復(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)用,會在另一個線程中進(jìn)行異步重試,重試如果再發(fā)生失敗,則會忽略,即使重試調(diào)用成功,原來的調(diào)用方也感知不到了。因此它通常適合于,對于實(shí)時性要求不高,且不需要返回值的一些異步操作。

  具體配置方法:

  服務(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ù)提供方配置。

  按照目前的實(shí)現(xiàn),F(xiàn)ailback策略還有一些局限,例如內(nèi)存中的失敗調(diào)用列表沒有上限,可能導(dǎo)致堆積,異步重試的執(zhí)行間隔無法調(diào)整,默認(rèn)是5秒。

  Forking(并行調(diào)用)

  上述幾種策略中,主要都是針對調(diào)用失敗發(fā)生后如何進(jìn)行彌補(bǔ)的角度去考慮的,而`Forking`策略則跟上述幾種策略不同,是一種典型的用成本換時間的思路。即第一次調(diào)用的時候就同時發(fā)起多個調(diào)用,只要其中一個調(diào)用成功,就認(rèn)為成功。在資源充足,且對于失敗的容忍度較低的場景下,可以采用此策略。

  具體配置方法:

  服務(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ù)的所有提供者進(jìn)行操作,此時可以使用廣播調(diào)用策略。此策略會逐個調(diào)用所有提供者,只要任意有一個提供者出錯,則認(rèn)為此次調(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)點(diǎn) | 缺點(diǎn) |

  | --------- | -------------------------------- | -------------------------------------- |

  | Failover | 對調(diào)用者屏蔽調(diào)用失敗的信息 | 增加RT,額外資源開銷,資源浪費(fèi) |

  | Failfast | 業(yè)務(wù)快速感知失敗狀態(tài)進(jìn)行自主決策 | 產(chǎn)生較多報錯的信息 |

  | Failsafe | 即使失敗了也不會影響核心流程 | 對于失敗的信息不敏感,需要額外的監(jiān)控 |

  | Failback | 失敗自動異步重試 | 重試任務(wù)可能堆積 |

  | Forking | 并行發(fā)起多個調(diào)用,降低失敗概率 | 消耗額外的機(jī)器資源,需要確保操作冪等性 |

  | Broadcast | 支持對所有的服務(wù)提供者進(jìn)行操作 | 資源消耗很大 |

  綜上我們得知,不同的容錯策略往往對應(yīng)不同的業(yè)務(wù)處理,這里做一個總結(jié)如下:

  Failover:通常用于對調(diào)用rt不敏感的場景,如讀操作;但重試會帶來更長延遲

  Failfast:通常用于非冪等性操作,需要快速感知失敗的場景;比如新增記錄

  Failsafe:通常用于旁路系統(tǒng),失敗不影響核心流程正確性的場景;如日志記錄

  Failback:通常用于對于實(shí)時性要求不高,且不需要返回值的一些異步操作的場景

  Forking:通常用于資源充足,且對于失敗的容忍度較低,實(shí)時性要求高的讀操作,但需要浪費(fèi)更多服務(wù)資源

  Broadcast:如通知所有提供者更新緩存或日志等本地資源信息

  源碼分析

  我們在上一章看到了兩個概念,分別是集群接口 Cluster 和 Cluster Invoker,這兩者是不同的。Cluster 是接口,而 Cluster Invoker 是一種 Invoker。服務(wù)提供者的選擇邏輯,以及遠(yuǎn)程調(diào)用失敗后的的處理邏輯均是封裝在 Cluster Invoker 中。那么 Cluster 接口和相關(guān)實(shí)現(xiàn)類有什么用呢?用途比較簡單,僅用于生成 Cluster Invoker。下面我們來看一下源碼。

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 的邏輯也是很簡單,無需解釋了。所以接下來,我們把重點(diǎn)放在各種 Cluster Invoker 上

  Cluster Invoker

  我們首先從各種 Cluster Invoker 的父類 AbstractClusterInvoker 源碼開始說起。前面說過,集群工作過程可分為兩個階段,第一個階段是在服務(wù)消費(fèi)者初始化期間,這個在服務(wù)引用那篇文章中分析過,就不贅述。第二個階段是在服務(wù)消費(fèi)者進(jìn)行遠(yuǎn)程調(diào)用時,此時 AbstractClusterInvoker 的 invoke 方法會被調(diào)用。列舉 Invoker,負(fù)載均衡等操作均會在此階段被執(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 進(jìn)行后續(xù)操作
    return doInvoke(invocation, invokers, loadbalance);
}

// 抽象方法,由子類實(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ù)操作。下面我們來看一下 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)實(shí)現(xiàn)類在前文已經(jīng)分析過,這里就不多說了。接下來,我們把目光轉(zhuǎn)移到 AbstractClusterInvoker 的各種實(shí)現(xiàn)類上,來看一下這些實(shí)現(xiàn)類是如何實(shí)現(xiàn) doInvoke 方法邏輯的。

  FailoverClusterInvoker

  FailoverClusterInvoker 在調(diào)用失敗時,會自動切換 Invoker 進(jìn)行重試。默認(rèn)配置下,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();
                // 在進(jìn)行重試前重新列舉 Invoker,這樣做的好處是,如果某個服務(wù)掛了,
                // 通過調(diào)用 list 可得到最新可用的 Invoker 列表
                copyinvokers = list(invocation);
                // 對 copyinvokers 進(jìn)行判空檢查
                checkInvokers(copyinvokers, invocation);
            }

            // 通過負(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),首先是通過負(fù)載均衡組件選擇一個 Invoker,然后再通過這個 Invoker 的 invoke 方法進(jìn)行遠(yuǎn)程調(diào)用。如果失敗了,記錄下異常,并進(jìn)行重試。重試時會再次調(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ù)消費(fèi)者盡可能的
    // 調(diào)用同一個服務(wù)提供者,除非該提供者掛了再進(jìn)行切換
    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 方法進(jìn)行檢查,如果檢查通過,則直接返回 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,則將負(fù)載均衡組件選出的 Invoker 賦值給 stickyInvoker
    if (sticky) {
        stickyInvoker = invoker;
    }
    return invoker;
}

  如上,select 方法的主要邏輯集中在了對粘滯連接特性的支持上。首先是獲取 sticky 配置,然后再檢測 invokers 列表中是否包含 stickyInvoker,如果不包含,則認(rèn)為該 stickyInvoker 不可用,此時將其置空。這里的 invokers 列表可以看做是**存活著的服務(wù)提供者**列表,如果這個列表不包含 stickyInvoker,那自然而然的認(rèn)為 stickyInvoker 掛了,所以置空。如果 stickyInvoker 存在于 invokers 列表中,此時要進(jìn)行下一項(xiàng)檢測 — 檢測 selected 中是否包含 stickyInvoker。如果包含的話,說明 stickyInvoker 在此之前沒有成功提供服務(wù)(但其仍然處于存活狀態(tài))。此時我們認(rèn)為這個服務(wù)不可靠,不應(yīng)該在重試期間內(nèi)再次被調(diào)用,因此這個時候不會返回該 stickyInvoker。如果 selected 不包含 stickyInvoker,此時還需要進(jìn)行可用性檢測,比如檢測服務(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,默認(rèn)為 RandomLoadBalance
        loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(Constants.DEFAULT_LOADBALANCE);
    }
   
    // 通過負(fù)載均衡組件選擇 Invoker
    Invoker<T> invoker = loadbalance.select(invokers, getUrl(), invocation);

    // 如果 selected 包含負(fù)載均衡選擇出的 Invoker,或者該 Invoker 無法經(jīng)過可用性檢查,此時進(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,以下代碼等價于:
                    //     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 主要做了兩件事,第一是通過負(fù)載均衡組件選擇 Invoker。第二是,如果選出來的 Invoker 不穩(wěn)定,或不可用,此時需要調(diào)用 reselect 方法進(jìn)行重選。若 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ìn)行了簡化,可以參考一下
    // 根據(jù) availablecheck 進(jìn)行不同的處理
    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 不為空,此時通過負(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()) {
            // 通過負(fù)載均衡組件進(jìn)行選擇
            return loadbalance.select(reselectInvokers, getUrl(), invocation);
        }
    }

    {
        // 若線程走到此處,說明 reselectInvokers 集合為空,此時不會調(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é)下來其實(shí)只做了兩件事情,第一是查找可用的 Invoker,并將其添加到 reselectInvokers 集合中。第二,如果 reselectInvokers 不為空,則通過負(fù)載均衡組件再次進(jìn)行選擇。其中第一件事情又可進(jìn)行細(xì)分,一開始,reselect 從 invokers 列表中查找有效可用的 Invoker,若未能找到,此時再到 selected 列表中繼續(xù)查找。關(guān)于 reselect 方法就先分析到這,繼續(xù)分析其他的 Cluster Invoker。

  FailbackClusterInvoker

  FailbackClusterInvoker 會在調(diào)用失敗后,返回一個空結(jié)果給服務(wù)消費(fèi)者。并通過定時任務(wù)對失敗的調(diào)用進(jìn)行重傳,適合執(zhí)行消息通知等操作。下面來看一下它的實(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)用過程中發(fā)生異常,此時僅打印錯誤日志,不拋出異常
            logger.error("Failback to invoke method ...");
           
            // 記錄調(diào)用信息
            addFailed(invocation, this);
            // 返回一個空結(jié)果給服務(wù)消費(fèi)者
            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)用進(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,對失敗的調(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 ...");
            }
        }
    }
}

  這個類主要由3個方法組成,首先是 doInvoker,該方法負(fù)責(zé)初次的遠(yuǎn)程調(diào)用。若遠(yuǎn)程調(diào)用失敗,則通過 addFailed 方法將調(diào)用信息存入到 failed 中,等待定時重試。addFailed 在開始階段會根據(jù) retryFuture 為空與否,來決定是否開啟定時任務(wù)。retryFailed 方法則是包含了失敗重試的邏輯,該方法會對 failed 進(jìn)行遍歷,然后依次對 Invoker 進(jìn)行調(diào)用。調(diào)用成功則將 Invoker 從 failed 中移除,調(diào)用失敗則忽略失敗原因。

  以上就是 FailbackClusterInvoker 的執(zhí)行邏輯,不是很復(fù)雜,繼續(xù)往下看。

  FailfastClusterInvoker

  FailfastClusterInvoker 只會進(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 ...");
        }
    }
}

  如上,首先是通過 select 方法選擇 Invoker,然后進(jìn)行遠(yuǎn)程調(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);
            // 進(jìn)行遠(yuǎn)程調(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 會在運(yùn)行時通過線程池創(chuàng)建多個線程,并發(fā)調(diào)用多個服務(wù)提供者。只要有一個服務(wù)提供者成功返回了結(jié)果,doInvoke 方法就會立即結(jié)束運(yùn)行。ForkingClusterInvoker 的應(yīng)用場景是在一些對實(shí)時性要求比較高**讀操作**(注意是讀操作,并行寫操作可能不安全)下使用,但這將會耗費(fè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);
            // 獲取超時配置
            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 {
                            // 進(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() 時,才將異常對象
                            // 放入阻塞隊(duì)列中,請大家思考一下為什么要這樣做。
                            if (value >= selected.size()) {
                                // 將異常對象存入到阻塞隊(duì)列中
                                ref.offer(e);
                            }
                        }
                    }
                });
            }
           
            // ----------------------? 分割線2 ?---------------------- //
           
            try {
                // 從阻塞隊(duì)列中取出遠(yuǎn)程調(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é)果存儲在阻塞隊(duì)列中。分割線2到方法結(jié)尾之間的邏輯主要用于從阻塞隊(duì)列中獲取返回結(jié)果,并對返回結(jié)果類型進(jìn)行判斷。如果為異常類型,則直接拋出,否則返回。

  以上就是ForkingClusterInvoker 的 doInvoker 方法大致過程。我們在分割線1和分割線2之間的代碼上留了一個問題,問題是這樣的:為什么要在`value >= selected.size()`的情況下,才將異常對象添加到阻塞隊(duì)列中?這里來解答一下。原因是這樣的,在并行調(diào)用多個服務(wù)提供者的情況下,只要有一個服務(wù)提供者能夠成功返回結(jié)果,而其他全部失敗。此時 ForkingClusterInvoker 仍應(yīng)該返回成功的結(jié)果,而非拋出異常。在`value >= selected.size()`時將異常對象放入阻塞隊(duì)列中,可以保證異常對象不會出現(xiàn)在正常結(jié)果的前面,這樣可從阻塞隊(duì)列中優(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 {
                // 進(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 的代碼,比較簡單,就不多說了。

  這里分析了集群容錯的幾種實(shí)現(xiàn)方式。集群容錯對于 Dubbo 框架來說,是很重要的邏輯。集群模塊處于服務(wù)提供者和消費(fèi)者之間,對于服務(wù)消費(fèi)者來說,集群可向其屏蔽服務(wù)提供者集群的情況,使其能夠?qū)P倪M(jìn)行遠(yuǎn)程調(diào)用。除此之外,通過集群模塊,我們還可以對服務(wù)之間的調(diào)用鏈路進(jìn)行編排優(yōu)化,治理服務(wù)。總的來說,對于 Dubbo 而言,集群容錯相關(guān)邏輯是非常重要的。想要對 Dubbo 有比較深的理解,集群容錯是必須要掌握的。

  負(fù)載均衡

  在之前章節(jié)中,介紹了服務(wù)集群的調(diào)用方式。我們發(fā)現(xiàn)在多服務(wù)實(shí)例時,負(fù)載均衡調(diào)用是其中極其重要的一環(huán)。在本章節(jié)中,我們一起學(xué)習(xí)Dubbo中的各種負(fù)載均衡策略

  負(fù)載均衡的主要作用

1663317456367_3.jpg

  負(fù)載均衡(LoadBalance),它的職責(zé)是將網(wǎng)絡(luò)請求,或者其他形式的負(fù)載“均攤”到不同的機(jī)器上。避免集群中部分服務(wù)器壓力過大,而另一些服務(wù)器比較空閑的情況。通過負(fù)載均衡,可以讓每臺服務(wù)器獲取到適合自己處理能力的負(fù)載。在為高負(fù)載服務(wù)器分流的同時,還可以避免資源浪費(fèi),一舉兩得。

  在 Dubbo 中,也有負(fù)載均衡的概念和相應(yīng)的實(shí)現(xiàn)。Dubbo 需要對服務(wù)消費(fèi)者的調(diào)用請求進(jìn)行分配,避免少數(shù)服務(wù)提供者負(fù)載過大。服務(wù)提供者負(fù)載過大,會導(dǎo)致部分請求超時。因此將負(fù)載均衡到每個服務(wù)提供者上,是非常必要的。

  內(nèi)置的負(fù)載均衡策略

  Dubbo 提供了4種負(fù)載均衡實(shí)現(xiàn),分別是基于權(quán)重隨機(jī)算法的 RandomLoadBalance、基于最少活躍調(diào)用數(shù)算法的 LeastActiveLoadBalance、基于 hash 一致性的 ConsistentHashLoadBalance,以及基于加權(quán)輪詢算法的 RoundRobinLoadBalance。這幾個負(fù)載均衡算法代碼不是很長,但是想看懂也不是很容易,需要大家對這幾個算法的原理有一定了解才行。如果不是很了解,也沒不用太擔(dān)心。我們會在分析每個算法的源碼之前,對算法原理進(jìn)行簡單的講解,幫助大家建立初步的印象。

  RandomLoadBalance

  RandomLoadBalance 是加權(quán)隨機(jī)算法的具體實(shí)現(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。接下來通過隨機(jī)數(shù)生成器生成一個范圍在 [0, 10) 之間的隨機(jī)數(shù),然后計算這個隨機(jī)數(shù)會落到哪個區(qū)間上。比如數(shù)字3會落到服務(wù)器 A 對應(yīng)的區(qū)間上,此時返回服務(wù)器 A 即可。權(quán)重越大的機(jī)器,在坐標(biāo)軸上對應(yīng)的區(qū)間范圍就越大,因此隨機(jī)數(shù)生成器生成的數(shù)字就會有更大的概率落到此區(qū)間內(nèi)。只要隨機(jī)數(shù)生成器產(chǎn)生的隨機(jī)數(shù)分布性很好,在經(jīng)過多次選擇后,每個服務(wù)器被選中的次數(shù)比例接近其權(quán)重比例。

1663317491965_4.jpg

  以上就是 RandomLoadBalance 背后的算法思想,比較簡單。下面開始分析源碼。


  RandomLoadBalance 的算法思想比較簡單,在經(jīng)過多次請求后,能夠?qū)⒄{(diào)用請求按照權(quán)重值進(jìn)行“均勻”分配。當(dāng)然 RandomLoadBalance 也存在一定的缺點(diǎn),當(dāng)調(diào)用次數(shù)比較少時,Random 產(chǎn)生的隨機(jī)數(shù)可能會比較集中,此時多數(shù)請求會落到同一臺服務(wù)器上。這個缺點(diǎn)并不是很嚴(yán)重,多數(shù)情況下可以忽略。RandomLoadBalance 是一個簡單,高效的負(fù)載均衡實(shí)現(xiàn),因此 Dubbo 選擇它作為缺省實(shí)現(xiàn)。

  LeastActiveLoadBalance

  LeastActiveLoadBalance 翻譯過來是最小活躍數(shù)負(fù)載均衡?;钴S調(diào)用數(shù)越小,表明該服務(wù)提供者效率越高,單位時間內(nèi)可處理更多的請求。此時應(yīng)優(yōu)先將請求分配給該服務(wù)提供者。在具體實(shí)現(xiàn)中,每個服務(wù)提供者對應(yīng)一個活躍數(shù) active。初始情況下,所有服務(wù)提供者活躍數(shù)均為0。每收到一個請求,活躍數(shù)加1,完成請求后則將活躍數(shù)減1。在服務(wù)運(yùn)行一段時間后,性能好的服務(wù)提供者處理請求的速度更快,因此活躍數(shù)下降的也越快,此時這樣的服務(wù)提供者能夠優(yōu)先獲取到新的服務(wù)請求、這就是最小活躍數(shù)負(fù)載均衡算法的基本思想。關(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 分支主要用于獲取隨機(jī)數(shù),并計算隨機(jī)數(shù)落在哪個區(qū)間上
        if (totalWeight > 0 && !sameWeight) {
            // 隨機(jī)獲取一個 [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++) {
                // 讓隨機(jī)值 offset 減去權(quán)重值
                offset -= getWeight(invokers.get(i), invocation);
                if (offset < 0) {
                    // 返回相應(yīng)的 Invoker
                    return invokers.get(i);
                }
            }
        }
       
        // 如果所有服務(wù)提供者權(quán)重值相同,此時直接隨機(jī)返回一個即可
        return invokers.get(random.nextInt(length));
    }
}

  除了最小活躍數(shù),LeastActiveLoadBalance 在實(shí)現(xiàn)上還引入了權(quán)重值。所以準(zhǔn)確的來說,LeastActiveLoadBalance 是基于加權(quán)最小活躍數(shù)算法實(shí)現(xiàn)的。舉個例子說明一下,在一個服務(wù)提供者集群中,有兩個性能優(yōu)異的服務(wù)提供者。某一時刻它們的活躍數(shù)相同,此時 Dubbo 會根據(jù)它們的權(quán)重去分配請求,權(quán)重越大,獲取到新請求的概率就越大。如果兩個服務(wù)提供者權(quán)重相同,此時隨機(jī)選擇一個即可。

  ConsistentHashLoadBalance

  一致性 hash 算法由麻省理工學(xué)院的 Karger 及其合作者于1997年提出的,算法提出之初是用于大規(guī)模緩存系統(tǒng)的負(fù)載均衡。它的工作過程是這樣的,首先根據(jù) ip 或者其他的信息為緩存節(jié)點(diǎn)生成一個 hash,并將這個 hash 投射到 [0, 2^32-1] 的圓環(huán)上。當(dāng)有查詢或?qū)懭胝埱髸r,則為緩存項(xiàng)的 key 生成一個 hash 值。然后查找第一個大于或等于該 hash 值的緩存節(jié)點(diǎn),并到這個節(jié)點(diǎn)中查詢或?qū)懭刖彺骓?xiàng)。如果當(dāng)前節(jié)點(diǎn)掛了,則在下一次查詢或?qū)懭刖彺鏁r,為緩存項(xiàng)查找另一個大于其 hash 值的緩存節(jié)點(diǎn)即可。大致效果如下圖所示,每個緩存節(jié)點(diǎn)在圓環(huán)上占據(jù)一個位置。如果緩存項(xiàng)的 key 的 hash 值小于緩存節(jié)點(diǎn) hash 值,則到該緩存節(jié)點(diǎn)中存儲或讀取緩存項(xiàng)。比如下面綠色點(diǎn)對應(yīng)的緩存項(xiàng)將會被存儲到 cache-2 節(jié)點(diǎn)中。由于 cache-3 掛了,原本應(yīng)該存到該節(jié)點(diǎn)中的緩存項(xiàng)最終會存儲到 cache-4 節(jié)點(diǎn)中。

1663317608890_5.jpg

  下面來看看一致性 hash 在 Dubbo 中的應(yīng)用。我們把上圖的緩存節(jié)點(diǎn)替換成 Dubbo 的服務(wù)提供者,于是得到了下圖:

1663317622912_6.jpg

  這里相同顏色的節(jié)點(diǎn)均屬于同一個服務(wù)提供者,比如 Invoker1-1,Invoker1-2,……, Invoker1-160。這樣做的目的是通過引入虛擬節(jié)點(diǎn),讓 Invoker 在圓環(huán)上分散開來,避免數(shù)據(jù)傾斜問題。所謂數(shù)據(jù)傾斜是指,由于節(jié)點(diǎn)不夠分散,導(dǎo)致大量請求落到了同一個節(jié)點(diǎn)上,而其他節(jié)點(diǎn)只會接收到了少量請求的情況。比如:

1663317635837_7.jpg

  如上,由于 Invoker-1 和 Invoker-2 在圓環(huán)上分布不均,導(dǎo)致系統(tǒng)中75%的請求都會落到 Invoker-1 上,只有 25% 的請求會落到 Invoker-2 上。解決這個問題辦法是引入虛擬節(jié)點(diǎn),通過虛擬節(jié)點(diǎn)均衡各個節(jié)點(diǎn)的請求量。

  到這里背景知識就普及完了,接下來開始分析源碼。我們先從 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í)行負(fù)載均衡邏輯。在分析 select 方法之前,我們先來看一下一致性 hash 選擇器 ConsistentHashSelector 的初始化過程,如下:

private static final class ConsistentHashSelector<T> {

    // 使用 TreeMap 存儲 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 計算的參數(shù)下標(biāo)值,默認(rèn)對第一個參數(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++) {
                // 對 address + i 進(jìn)行 md5 運(yùn)算,得到一個長度為16的字節(jié)數(shù)組
                byte[] digest = md5(address + i);
                // 對 digest 部分字節(jié)進(jìn)行4次 hash 運(yùn)算,得到四個不同的 long 型正整數(shù)
                for (int h = 0; h < 4; h++) {
                    // h = 0 時,取 digest 中下標(biāo)為 0 ~ 3 的4個字節(jié)進(jìn)行位運(yùn)算
                    // h = 1 時,取 digest 中下標(biāo)為 4 ~ 7 的4個字節(jié)進(jìn)行位運(yùn)算
                    // 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é)點(diǎn)數(shù)以及參與 hash 計算的參數(shù)下標(biāo),默認(rèn)情況下只使用第一個參數(shù)進(jìn)行 hash。需要特別說明的是,ConsistentHashLoadBalance 的負(fù)載均衡邏輯只受參數(shù)值影響,具有相同參數(shù)值的請求將會被分配給同一個服務(wù)提供者。ConsistentHashLoadBalance 不 關(guān)系權(quán)重,因此使用時需要注意一下。

  在獲取虛擬節(jié)點(diǎn)數(shù)和參數(shù)下標(biāo)配置后,接下來要做的事情是計算虛擬節(jié)點(diǎn) hash 值,并將虛擬節(jié)點(diǎn)存儲到 TreeMap 中。到此,ConsistentHashSelector 初始化工作就完成了。接下來,我們來看看 select 方法的邏輯。

public Invoker<T> select(Invocation invocation) {
    // 將參數(shù)轉(zhuǎn)為 key
    String key = toKey(invocation.getArguments());
    // 對參數(shù) key 進(jìn)行 md5 運(yùn)算
    byte[] digest = md5(key);
    // 取 digest 數(shù)組的前四個字節(jié)進(jìn)行 hash 運(yùn)算,再將 hash 值傳給 selectForKey 方法,
    // 尋找合適的 Invoker
    return selectForKey(hash(digest, 0));
}

private Invoker<T> selectForKey(long hash) {
    // 到 TreeMap 中查找第一個節(jié)點(diǎn)值大于或等于當(dāng)前 hash 的 Invoker
    Map.Entry<Long, Invoker<T>> entry = virtualInvokers.tailMap(hash, true).firstEntry();
    // 如果 hash 大于 Invoker 在圓環(huán)上最大的位置,此時 entry = null,
    // 需要將 TreeMap 的頭節(jié)點(diǎn)賦值給 entry
    if (entry == null) {
        entry = virtualInvokers.firstEntry();
    }

    // 返回 Invoker
    return entry.getValue();
}

  如上,選擇的過程相對比較簡單了。首先是對參數(shù)進(jìn)行 md5 以及 hash 運(yùn)算,得到一個 hash 值。然后再拿這個值到 TreeMap 中查找目標(biāo) Invoker 即可。

  RoundRobinLoadBalance

  LeastActiveLoadBalance 即加權(quán)輪詢負(fù)載均衡,我們先來了解一下什么是加權(quán)輪詢。這里從最簡單的輪詢開始講起,所謂輪詢是指將請求輪流分配給每臺服務(wù)器。舉個例子,我們有三臺服務(wù)器 A、B、C。我們將第一個請求分配給服務(wù)器 A,第二個請求分配給服務(wù)器 B,第三個請求分配給服務(wù)器 C,第四個請求再次分配給服務(wù)器 A。這個過程就叫做輪詢。輪詢是一種無狀態(tài)負(fù)載均衡算法,實(shí)現(xiàn)簡單,適用于每臺服務(wù)器性能相近的場景下。但現(xiàn)實(shí)情況下,我們并不能保證每臺服務(wù)器性能均相近。如果我們將等量的請求分配給性能較差的服務(wù)器,這顯然是不合理的。因此,這個時候我們需要對輪詢過程進(jìn)行加權(quán),以調(diào)控每臺服務(wù)器的負(fù)載。經(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)重變化了,此時進(jì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> 進(jìn)行檢查,過濾掉長時間未被更新的節(jié)點(diǎn)。
        // 該節(jié)點(diǎn)可能掛了,invokers 中不包含該節(jié)點(diǎn),所以該節(jié)點(diǎn)的 lastUpdate 長時間無法被更新。
        // 若未更新時長超過閾值后,就會被移除掉,默認(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);
                   
                    // 遍歷修改,即移除過期記錄
                    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)重的值進(jìn)行循環(huán)的。

  負(fù)載均衡總結(jié)

  Dubbo 負(fù)載均衡策略提供下列四種方式:

   Random LoadBalance 隨機(jī),按權(quán)重設(shè)置隨機(jī)概率。Dubbo的默認(rèn)負(fù)載均衡策略

   在一個截面上碰撞的概率高,但調(diào)用量越大分布越均勻,而且按概率使用權(quán)重后也比較均勻,有利于動態(tài)調(diào)整提供者權(quán)重。

   RoundRobin LoadBalance 輪循,按公約后的權(quán)重設(shè)置輪循比率。

   存在慢的提供者累積請求問題,比如:第二臺機(jī)器很慢,但沒掛,當(dāng)請求調(diào)到第二臺時就卡在那,久而久之,所有請求都卡在調(diào)到第二臺上。

   LeastActive LoadBalance 最少活躍調(diào)用數(shù),相同活躍數(shù)的隨機(jī),活躍數(shù)指調(diào)用前后計數(shù)差。

   使慢的提供者收到更少請求,因?yàn)樵铰奶峁┱叩恼{(diào)用前后計數(shù)差會越大。

   ConsistentHash LoadBalance 一致性Hash,相同參數(shù)的請求總是發(fā)到同一提供者。

   當(dāng)某一臺提供者掛時,原本發(fā)往該提供者的請求,基于虛擬節(jié)點(diǎn),平攤到其它提供者,不會引起劇烈變動。

  服務(wù)治理

  服務(wù)治理的概述

  服務(wù)治理主要作用是改變運(yùn)行時服務(wù)的行為和選址邏輯,達(dá)到限流,權(quán)重配置等目的,主要有:標(biāo)簽路由,條件路由,黑白名單,動態(tài)配置,權(quán)重調(diào)節(jié),負(fù)載均衡等功能。

1663317823091_8.jpg

  執(zhí)行過程

1663317835641_9.jpg

  1、消費(fèi)者,提供者啟動成功,訂閱zookeeper節(jié)點(diǎn)

  2、管理平臺對服務(wù)進(jìn)行治理處理,向zookeeper寫入節(jié)點(diǎn)數(shù)據(jù)

  3、寫入成功,通知消費(fèi)者,提供者

  4、根據(jù)不同的業(yè)務(wù)處理,在invoker調(diào)用時做出響應(yīng)的處理

  相關(guān)案例

  服務(wù)禁用

  服務(wù)禁用:通常用于臨時踢除某臺提供者機(jī)器

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ù)的正常運(yùn)行。

  容錯:當(dāng)系統(tǒng)出現(xiàn)非業(yè)務(wù)異常(比如并發(fā)數(shù)太高導(dǎo)致超時,網(wǎng)絡(luò)異常等)時,不對該接口進(jìn)行處理。

  屏蔽:在大促,促銷活動的可預(yù)知情況下,例如雙11活動。采用直接屏蔽接口訪問

configVersion: v2.7
scope: service
key: org.apache.dubbo.samples.governance.api.DemoService
enabled: true
configs:
  - side: consumer
    parameters:
     force: return 12345


分享到:
在線咨詢 我要報名
和我們在線交談!