RabbitMQ分布式部署有3種方式：

• 集群
• Federation
• Shovel

Federation與Shovel都是以插件的形式來實現(xiàn)，復(fù)雜性相對高，而集群是RabbitMQ的自帶屬性，相對簡單。

這三種方式并不是互斥的，可以根據(jù)需求選擇相互組合來達(dá)到目的。

1、集群

1.1 基本概念

RabbitMQ本身是基于Erlang編寫，Erlang語言天生具備分布式特性（通過同步Erlang集群各節(jié)點(diǎn)的magic cookie來實現(xiàn)）。

因此，RabbitMQ天然支持Clustering。這使得RabbitMQ本身不需要像ActiveMQ、Kafka那樣通過ZooKeeper分別來實現(xiàn)HA方案和保存集群的元數(shù)據(jù)。集群是保證可靠性的一種方式，同時可以通過水平擴(kuò)展以達(dá)到增加消息吞吐量能力的目的。

我們把部署RabbitMQ的機(jī)器稱為節(jié)點(diǎn)，也就是broker。broker有2種類型節(jié)點(diǎn)：磁盤節(jié)點(diǎn)和內(nèi)存節(jié)點(diǎn)。顧名思義，磁盤節(jié)點(diǎn)的broker把元數(shù)據(jù)存儲在磁盤中，內(nèi)存節(jié)點(diǎn)把元數(shù)據(jù)存儲在內(nèi)存中，很明顯，磁盤節(jié)點(diǎn)的broker在重啟后元數(shù)據(jù)可以通過讀取磁盤進(jìn)行重建，保證了元數(shù)據(jù)不丟失，內(nèi)存節(jié)點(diǎn)的broker可以獲得更高的性能，但在重啟后元數(shù)據(jù)就都丟了。

元數(shù)據(jù)包含以下內(nèi)容：

• queue元數(shù)據(jù)：queue名稱、屬性
• exchange：exchange名稱、屬性
• binding元數(shù)據(jù)：exchange和queue之間、exchange和exchange之間的綁定關(guān)系
• vhost元數(shù)據(jù)：vhost內(nèi)部的命名空間、安全屬性數(shù)據(jù)等

單節(jié)點(diǎn)系統(tǒng)必須是磁盤節(jié)點(diǎn)，否則每次你重啟RabbitMQ之后所有的系統(tǒng)配置信息都會丟失。

集群中至少有一個磁盤節(jié)點(diǎn)，當(dāng)節(jié)點(diǎn)加入和離開集群時，必須通知磁盤 節(jié)點(diǎn)。

如果集群中的唯一一個磁盤節(jié)點(diǎn)，結(jié)果這個磁盤節(jié)點(diǎn)還崩潰了，那會發(fā)生什么情況？集群依然可以繼續(xù)路由消息（因為其他節(jié)點(diǎn)元數(shù)據(jù)在還存在），但無法做以下操作：

• 創(chuàng)建隊列、交換器、綁定
• 添加用戶
• 更改權(quán)限
• 添加、刪除集群節(jié)點(diǎn)

也就是說，如果唯一磁盤的磁盤節(jié)點(diǎn)崩潰，集群是可以保持運(yùn)行的，但不能更改任何東西。為了增加可靠性，一般會在集群中設(shè)置兩個磁盤節(jié)點(diǎn)，只要任何一個處于工作狀態(tài)，就可以保障集群的正常服務(wù)。

RabbitMQ的集群模式分為兩種：普通模式與鏡像模式。

1.2 普通模式

普通模式，也是默認(rèn)的集群模式。

對于Queue來說，消息實體只存在于其中一個節(jié)點(diǎn)，A、B兩個節(jié)點(diǎn)僅有相同的元數(shù)據(jù)，即隊列結(jié)構(gòu)。當(dāng)消息進(jìn)入A節(jié)點(diǎn)的Queue中后，consumer從B節(jié)點(diǎn)拉取時，RabbitMQ會臨時在A、B間進(jìn)行消息傳輸，把A中的消息實體取出并經(jīng)過B發(fā)送給consumer。所以consumer應(yīng)盡量連接每一個節(jié)點(diǎn)，從中取消息。即對于同一個邏輯隊列，要在多個節(jié)點(diǎn)建立物理Queue。否則無論consumer連A或B，出口總在A，會產(chǎn)生瓶頸。

1.png

隊列所在的節(jié)點(diǎn)稱為宿主節(jié)點(diǎn)。

隊列創(chuàng)建時，只會在宿主節(jié)點(diǎn)創(chuàng)建隊列的進(jìn)程，宿主節(jié)點(diǎn)包含完整的隊列信息，包括元數(shù)據(jù)、狀態(tài)、內(nèi)容等等。因此，只有隊列的宿主節(jié)點(diǎn)才能知道隊列的所有信息。

隊列創(chuàng)建后，集群只會同步隊列和交換器的元數(shù)據(jù)到集群中的其他節(jié)點(diǎn)，并不會同步隊列本身，因此非宿主節(jié)點(diǎn)就只知道隊列的元數(shù)據(jù)和指向該隊列宿主節(jié)點(diǎn)的指針。

假如現(xiàn)在一個客戶端需要對Queue A進(jìn)行發(fā)布或者訂閱，發(fā)起與集群的連接，有兩種可能的場景：

• 如果客戶端連接至Broker A，Broker A是Queue A的宿主節(jié)點(diǎn)，那么此時的集群中的消息收發(fā)只與Broker A相關(guān)。
• 如果客戶端連接至Broker B或Broker C，不是Queue A的宿主節(jié)點(diǎn)，那么此時的Broker主要起了一個路由轉(zhuǎn)發(fā)作用，根據(jù)這兩個節(jié)點(diǎn)上的元數(shù)據(jù)轉(zhuǎn)發(fā)至Broker A上。

由于節(jié)點(diǎn)之間存在路由轉(zhuǎn)發(fā)的情況，對延遲非常敏感，應(yīng)當(dāng)只在本地局域網(wǎng)內(nèi)使用，在廣域網(wǎng)中不應(yīng)該使用集群，而應(yīng)該用Federation或者Shovel代替。

這樣的設(shè)計，保證了不論從哪個broker中均可以消費(fèi)所有隊列的數(shù)據(jù)，并分擔(dān)了負(fù)載，因此，增加broker可以線性提高服務(wù)的性能和吞吐量。

但該方案也有顯著的缺陷，那就是不能保證消息不會丟失。當(dāng)集群中某一節(jié)點(diǎn)崩潰時，崩潰節(jié)點(diǎn)所在的隊列進(jìn)程和關(guān)聯(lián)的綁定都會消失，附加在那些隊列上的消費(fèi)者也會丟失其訂閱信息，匹配該隊列的新消息也會丟失。比如A為宿主節(jié)點(diǎn)，當(dāng)A節(jié)點(diǎn)故障后，B節(jié)點(diǎn)無法取到A節(jié)點(diǎn)中還未消費(fèi)的消息實體。如果做了消息持久化，那么得等A節(jié)點(diǎn)恢復(fù)，然后才可被消費(fèi)；如果沒有持久化的話，然后就沒有然后了……

肯定有不少同學(xué)會問，想要實現(xiàn)HA方案，那將RabbitMQ集群中的所有Queue的完整數(shù)據(jù)在所有節(jié)點(diǎn)上都保存一份不就可以了嗎？比如類似MySQL的主主模式，任何一個節(jié)點(diǎn)出現(xiàn)故障或者宕機(jī)不可用時，那么使用者的客戶端只要能連接至其他節(jié)點(diǎn)，不就能夠照常完成消息的發(fā)布和訂閱嗎？

RabbitMQ這么設(shè)計是基于性能和存儲空間上來考慮：

• 存儲空間，如果每個集群節(jié)點(diǎn)都擁有所有Queue的完全數(shù)據(jù)拷貝，那么每個節(jié)點(diǎn)的存儲空間會非常大，集群的消息積壓能力會非常弱，無法通過集群節(jié)點(diǎn)的擴(kuò)容提高消息積壓能力。

• 性能，消息的發(fā)布者需要將消息復(fù)制到每一個集群節(jié)點(diǎn)，對于持久化消息，網(wǎng)絡(luò)和磁盤同步復(fù)制的開銷都會明顯增，無法提升性能。（此處可以引申思考一下kafka中replica的分配方式） 。

1.3 鏡像模式

引入鏡像隊列（Mirror Queue）的機(jī)制，可以將隊列鏡像到集群中的其他Broker節(jié)點(diǎn)之上，如果集群中的一個節(jié)點(diǎn)失效了，隊列能夠自動切換到鏡像中的另一個節(jié)點(diǎn)上以保證服務(wù)的可用性。

一個鏡像隊列中包含有1個主節(jié)點(diǎn)master和若干個從節(jié)點(diǎn)slave。其主從節(jié)點(diǎn)包含如下幾個特點(diǎn)：

• 消息的讀寫都是在master上進(jìn)行，并不是讀寫分離

• master接收命令后會向salve進(jìn)行組播，salve會命令執(zhí)行順序執(zhí)行

• master失效，根據(jù)節(jié)點(diǎn)加入的時間，最老的slave會被提升為master

• 互為鏡像的是隊列，并非節(jié)點(diǎn)，集群中可以不同節(jié)點(diǎn)可以互為鏡像隊列，也就是說隊列的master可以分布在不同的節(jié)點(diǎn)上

該模式和普通模式不同之處在于，消息實體會主動在鏡像節(jié)點(diǎn)間同步，而不是在consumer取數(shù)據(jù)時臨時拉取。該模式帶來的副作用也很明顯，除了降低系統(tǒng)性能外，如果鏡像隊列數(shù)量過多，加之大量的消息進(jìn)入，集群內(nèi)部的網(wǎng)絡(luò)帶寬將會被這種同步通訊大大消耗掉。所以在對可靠性要求較高的場合中適用。

1.3.1 鏡像隊列的設(shè)置

一個隊列想做成鏡像隊列，需要先設(shè)置policy，然后客戶端創(chuàng)建隊列的時候，rabbitmq集群根據(jù)隊列名稱自動設(shè)置為普通隊列還是鏡像隊列。

鏡像隊列的配置通過添加policy完成，policy添加的命令為：

rabbitmqctl set_policy [-p Vhost] Name Pattern Definition [Priority]

• -p Vhost: 可選參數(shù)，針對指定vhost下的queue進(jìn)行設(shè)置
• Name: policy的名稱
• Pattern: queue的匹配模式（正則表達(dá)式）
• Definition: 鏡像定義，包括三個部分 ha-mode，ha-params，ha-sync-mode
  • ha-mode: 指明鏡像隊列的模式，有效值為 all/exactly/nodes
  • all表示在集群所有的節(jié)點(diǎn)上進(jìn)行鏡像
  • exactly表示在指定個數(shù)的節(jié)點(diǎn)上進(jìn)行鏡像，節(jié)點(diǎn)的個數(shù)由ha-params指定
  • nodes表示在指定的節(jié)點(diǎn)上進(jìn)行鏡像，節(jié)點(diǎn)名稱通過ha-params指定
  • ha-params: ha-mode模式需要用到的參數(shù)
  • ha-sync-mode: 鏡像隊列中消息的同步方式，有效值為automatic，manually
• Priority: 可選參數(shù)， policy的優(yōu)先級

例如，對隊列名稱以hello開頭的所有隊列進(jìn)行鏡像，并在集群的兩個節(jié)點(diǎn)上完成鏡像，policy的設(shè)置命令為：

rabbitmqctl set_policy hello-ha "^hello" '{"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"}'

1.3.2 鏡像隊列的實現(xiàn)原理

1.3.2.1 綜述

通常隊列由兩部分組成：一部分是AMQQueue，負(fù)責(zé)AMQP協(xié)議相關(guān)的消息處理，即接收生產(chǎn)者發(fā)布的消息、向消費(fèi)者投遞消息、處理消息confirm、acknowledge等等；另一部分是BackingQueue，它提供了相關(guān)的接口供AMQQueue調(diào)用，完成消息的存儲以及可能的持久化工作等。

鏡像隊列基本上就是一個特殊的BackingQueue，它內(nèi)部包裹了一個普通的BackingQueue做本地消息持久化處理，在此基礎(chǔ)上增加了將消息和ack復(fù)制到所有鏡像的功能。所有對mirror_queue_master的操作，會通過組播GM（下面會講到）的方式同步到各slave節(jié)點(diǎn)。GM負(fù)責(zé)消息的廣播，mirror_queue_slave負(fù)責(zé)回調(diào)處理，而master上的回調(diào)處理是由coordinator負(fù)責(zé)完成。mirror_queue_slave中包含了普通的BackingQueue進(jìn)行消息的存儲，master節(jié)點(diǎn)中BackingQueue包含在mirror_queue_master中由AMQQueue進(jìn)行調(diào)用。

消息的發(fā)布（除了Basic.Publish之外）與消費(fèi)都是通過master節(jié)點(diǎn)完成。master節(jié)點(diǎn)對消息進(jìn)行處理的同時將消息的處理動作通過GM廣播給所有的slave節(jié)點(diǎn)，slave節(jié)點(diǎn)的GM收到消息后，通過回調(diào)交由mirror_queue_slave進(jìn)行實際的處理。

3.jpeg

GM(Guarenteed Multicast)是一種可靠的組播通訊協(xié)議，該協(xié)議能夠保證組播消息的原子性，即保證組中活著的節(jié)點(diǎn)要么都收到消息要么都收不到。它的實現(xiàn)大致如下：

將所有的節(jié)點(diǎn)形成一個循環(huán)鏈表，每個節(jié)點(diǎn)都會監(jiān)控位于自己左右兩邊的節(jié)點(diǎn)，當(dāng)有節(jié)點(diǎn)新增時，相鄰的節(jié)點(diǎn)保證當(dāng)前廣播的消息會復(fù)制到新的節(jié)點(diǎn)上；當(dāng)有節(jié)點(diǎn)失效時，相鄰的節(jié)點(diǎn)會接管保證本次廣播的消息會復(fù)制到所有的節(jié)點(diǎn)。在master節(jié)點(diǎn)和slave節(jié)點(diǎn)上的這些gm形成一個group，group（gm_group）的信息會記錄在mnesia中。不同的鏡像隊列形成不同的group。消息從master節(jié)點(diǎn)對于的gm發(fā)出后，順著鏈表依次傳送到所有的節(jié)點(diǎn)，由于所有節(jié)點(diǎn)組成一個循環(huán)鏈表，master節(jié)點(diǎn)對應(yīng)的gm最終會收到自己發(fā)送的消息，這個時候master節(jié)點(diǎn)就知道消息已經(jīng)復(fù)制到所有的slave節(jié)點(diǎn)了。

1.3.2.2 新增節(jié)點(diǎn)

slave節(jié)點(diǎn)先從gm_group中獲取對應(yīng)group的所有成員信息，然后隨機(jī)選擇一個節(jié)點(diǎn)并向這個節(jié)點(diǎn)發(fā)送請求，這個節(jié)點(diǎn)收到請求后，更新gm_group對應(yīng)的信息，同時通知左右節(jié)點(diǎn)更新鄰居信息（調(diào)整對左右節(jié)點(diǎn)的監(jiān)控）及當(dāng)前正在廣播的消息，然后回復(fù)通知請求節(jié)點(diǎn)成功加入group。請求加入group的節(jié)點(diǎn)收到回復(fù)后再更新rabbit_queue中的相關(guān)信息，并根據(jù)需要進(jìn)行消息的同步。

4.jpg

5.jpg

1.3.2.3 刪除節(jié)點(diǎn)

當(dāng)slave節(jié)點(diǎn)失效時，僅僅是相鄰節(jié)點(diǎn)感知，然后重新調(diào)整鄰居節(jié)點(diǎn)信息、更新rabbit_queue、gm_group的記錄等。如果是master節(jié)點(diǎn)失效，"資格最老"的slave節(jié)點(diǎn)被提升為master節(jié)點(diǎn)，slave節(jié)點(diǎn)會創(chuàng)建出新的coordinator，并告知gm修改回調(diào)處理為coordinator，原來的mirror_queue_slave充當(dāng)amqqueue_process處理生產(chǎn)者發(fā)布的消息，向消費(fèi)者投遞消息等。

6.jpg

上面提到如果是slave節(jié)點(diǎn)失效，只有相鄰的節(jié)點(diǎn)能感知到，那么master節(jié)點(diǎn)失效是不是也是只有相鄰的節(jié)點(diǎn)能感知到？假如是這樣的話，如果相鄰的節(jié)點(diǎn)不是"資格最老"的節(jié)點(diǎn)，怎么通知"資格最老"的節(jié)點(diǎn)提升為新的master節(jié)點(diǎn)呢?

實際上，所有的slave節(jié)點(diǎn)在加入group時，mirror_queue_slave進(jìn)程會對master節(jié)點(diǎn)的amqqueue_process進(jìn)程（也可能是mirror_queue_slave進(jìn)程）進(jìn)行監(jiān)控，如果master節(jié)點(diǎn)失效的話，mirror_queue_slave會感知，然后再通過gm進(jìn)行廣播，這樣所有的節(jié)點(diǎn)最終都會知道m(xù)aster節(jié)點(diǎn)失效。當(dāng)然，只有"資格最老"的節(jié)點(diǎn)會提升自己為新的master。

1.3.2.4 消息的廣播

消息從master節(jié)點(diǎn)發(fā)出，順著節(jié)點(diǎn)鏈表發(fā)送。在這期間，所有的slave節(jié)點(diǎn)都會對消息進(jìn)行緩存，當(dāng)master節(jié)點(diǎn)收到自己發(fā)送的消息后，會再次廣播ack消息，同樣ack消息會順著節(jié)點(diǎn)鏈表經(jīng)過所有的slave節(jié)點(diǎn)，其作用是通知slave節(jié)點(diǎn)可以清除緩存的消息，當(dāng)ack消息回到master節(jié)點(diǎn)時對應(yīng)廣播消息的生命周期結(jié)束。

下圖為一個簡單的示意圖，A節(jié)點(diǎn)為master節(jié)點(diǎn)，廣播一條內(nèi)容為"test"的消息。"1"表示消息為廣播的第一條消息；"id=A"表示消息的發(fā)送者為節(jié)點(diǎn)A。右邊是slave節(jié)點(diǎn)記錄的狀態(tài)信息。

7.jpg

為什么所有的節(jié)點(diǎn)都需要緩存一份發(fā)布的消息呢？

master發(fā)布的消息是依次經(jīng)過所有slave節(jié)點(diǎn)，在這期間的任何時刻，有可能有節(jié)點(diǎn)失效，那么相鄰的節(jié)點(diǎn)可能需要重新發(fā)送給新的節(jié)點(diǎn)。例如，A->B->C->D->A形成的循環(huán)鏈表，A為master節(jié)點(diǎn)，廣播消息發(fā)送給節(jié)點(diǎn)B，B再發(fā)送給C，如果節(jié)點(diǎn)C收到B發(fā)送的消息還未發(fā)送給D時異常結(jié)束了，那么節(jié)點(diǎn)B感知后節(jié)點(diǎn)C失效后需要重新將消息發(fā)送給D。同樣，如果B節(jié)點(diǎn)將消息發(fā)送給C后，B,C節(jié)點(diǎn)中新增了E節(jié)點(diǎn)，那么B節(jié)點(diǎn)需要再將消息發(fā)送給新增的E節(jié)點(diǎn)。

1.3.2.5 消息的同步

配置鏡像隊列的時候有個ha-sync-mode屬性，這個有什么用呢?

新節(jié)點(diǎn)加入到group后，最多能從左邊節(jié)點(diǎn)獲取到當(dāng)前正在廣播的消息內(nèi)容，加入group之前已經(jīng)廣播的消息則無法獲取到。如果此時master節(jié)點(diǎn)不幸失效，而新節(jié)點(diǎn)有恰好成為了新的master，那么加入group之前已經(jīng)廣播的消息則會全部丟失。

注意：這里的消息具體是指新節(jié)點(diǎn)加入前已經(jīng)發(fā)布并復(fù)制到所有slave節(jié)點(diǎn)的消息，并且這些消息還未被消費(fèi)者消費(fèi)或者未被消費(fèi)者確認(rèn)。如果新節(jié)點(diǎn)加入前，所有廣播的消息被消費(fèi)者消費(fèi)并確認(rèn)了，master節(jié)點(diǎn)刪除消息的同時會通知slave節(jié)點(diǎn)完成相應(yīng)動作。這種情況等同于新節(jié)點(diǎn)加入前沒有發(fā)布任何消息。

避免這種問題的解決辦法就是對新的slave節(jié)點(diǎn)進(jìn)行消息同步。當(dāng)ha-sync-mode配置為自動同步（automatic）時，新節(jié)點(diǎn)加入group時會自動進(jìn)行消息的同步；如果配置為manually則需要手動操作完成同步。

2、Federation

Federation直譯過來是聯(lián)邦，它的設(shè)計目標(biāo)是使 RabbitMQ 在不同的 Broker 節(jié)點(diǎn)之間進(jìn)行消息傳遞而無須建
立集群。具有以下特點(diǎn)：

• 支持不同管理域(不同的用戶和vhost、不同版本的RabbitMQ)中的Broker或集群間傳遞消息
• 基于AMQP 0-9-1協(xié)議在不同的Broker之間通信，能容忍不穩(wěn)定的網(wǎng)絡(luò)連接情況

那么它到底有什么用呢？我們可以從一個實際場景入手：

有兩個服務(wù)分別部署在國內(nèi)和海外，它們之間需要通過消息隊列來通訊。

很明顯無論RabbitMQ部署在海外還是國內(nèi)，另一方一定得忍受連接上的延遲。因此我們可以在海外和國內(nèi)各部署一個MQ，這樣一來海外連接海外的MQ,國內(nèi)連接國內(nèi)，就不會有連接上的延遲了。

但這樣還會有問題，假設(shè)某生產(chǎn)者將消息存入海外MQ中的某個隊列 queueB ， 在國內(nèi)的服務(wù)想要消費(fèi) queueB 消息，消息的流轉(zhuǎn)及確認(rèn)必然要忍受較大的網(wǎng)絡(luò)延遲 ，內(nèi)部編碼邏輯也會因這一因素變得更加復(fù)雜。

此外，服務(wù)可能得維護(hù)兩個MQ的配置，比如國內(nèi)服務(wù)在生產(chǎn)消息時得使用國內(nèi)MQ，消費(fèi)消息時得監(jiān)聽海外MQ的隊列，降低了系統(tǒng)的維護(hù)性。

可能有人想到可以用集群，但是RabbitMQ的集群對延遲非常敏感，一般部署在局域網(wǎng)內(nèi)，如果部署在廣域網(wǎng)可能會產(chǎn)生網(wǎng)絡(luò)分區(qū)等等問題。

這時候，F(xiàn)ederation就派上用場了。它被設(shè)計成能夠容忍不穩(wěn)定的網(wǎng)絡(luò)連接情況，完全能夠滿足這樣的場景。

2.1 聯(lián)邦交換器

那使用Federation之后是怎樣的業(yè)務(wù)流程呢?

首先我們在海外MQ上定義exchangeA，它通過路由鍵“rkA”綁定著queueA。然后用Federation在exchangeA上建立一條單向連接到國內(nèi)RabbitMQ，F(xiàn)ederation則自動會在國內(nèi)RabbitMQ建立一個exchangeA交換器（默認(rèn)同名）。

8.png

這時候，如果部署在國內(nèi)的client C在國內(nèi)MQ上publish了一條消息，這條消息會通過 Federation link 轉(zhuǎn)發(fā)到海外MQ的交換器exchangeA中，最終消息會存入與 exchangeA 綁定的隊列 queueA 中，而client C也能立即得到返回。

實際上，F(xiàn)ederation插件還會在國內(nèi)MQ建立一個內(nèi)部的交換器：exchangeA→ broker3 B（broker3是集群名），并通過路由鍵 "rkA"將它和國內(nèi)MQ的exchangeA綁定起來。接下來還會在國內(nèi)MQ上建立一個內(nèi)部隊列federation: exchangeA->broker3 B，并與內(nèi)部exchange綁定。這些操作都是內(nèi)部的，對客戶端來說是透明的。

值得一提的是，F(xiàn)ederation的連接是單向的，如果是在海外MQ的exchangeA上發(fā)送消息是不會轉(zhuǎn)到國內(nèi)的。

這種在exchange上建立連接進(jìn)行聯(lián)邦的，就叫做聯(lián)邦交換器。一個聯(lián)邦交換器接收上游（upstream）的信息，這里的上游指的是其他的MQ節(jié)點(diǎn)。

對比前面舉的例子，國內(nèi)MQ就是上游，聯(lián)邦交換器能夠?qū)⒃景l(fā)送給上游交換器的消息路由到本地的某個隊列中。

2.2 聯(lián)邦隊列

有聯(lián)邦交換器自然也有聯(lián)播隊列，聯(lián)邦隊列則允許一個本地消費(fèi)者接收到來自上游隊列的消息 。

9.png

如圖，海外MQ有隊列A，給其設(shè)置一條鏈接，F(xiàn)ederation則自動會在國內(nèi)RabbitMQ建立一個隊列A（默認(rèn)同名）。

當(dāng)有消費(fèi)者 ClinetA連接海外MQ并消費(fèi) queueA 中的消息時，如果隊列 queueA中本身有若干消息堆積，那么 ClientA直接消費(fèi)這些消息，此時海外MQ中的queueA并不會拉取國內(nèi)中的 queueA 的消息；如果隊列 queueA中沒有消息堆積或者消息被消費(fèi)完了，那么它會通過 Federation link 拉取上游隊列 queueA 中的消息(如果有消息)，然后存儲到本地，之后再被消費(fèi)者 ClientA進(jìn)行消費(fèi) 。

2.3 Federation使用

首先開啟Federation 功能：

##啟用插件
rabbitmq-plugins enable rabbitmq_federation
##啟用管理插件
rabbitmq-plugins enable rabbitmq_federation_management

值得注意的是，當(dāng)需要在集群中使用 Federation 功能的時候，集群中所有的節(jié)點(diǎn)都應(yīng)該開啟 Federation 插件。

接下來我們要配置兩個東西：upstreams和Policies。

每個 upstream 用于定義與其他 Broker 建立連接的信息。

10.png

通用參數(shù)如下：

• name: 定義這個upstreams的名稱
• URI : 定義 upstreams的 AMQP 連接。例如amqp://username:password@server-name/my-vhost
• Prefetch count : 定義 Federation 內(nèi)部緩存的消息條數(shù)，即在收到上游消息之后且在發(fā)送到下游之前緩存的消息條數(shù)。
• Reconnect delay: Federation link 由于某種原因斷開之后，需要等待多少秒開始重新建立連接。
• Acknowledgement Mode: 定義 Federation link 的消息確認(rèn)方式 。其有 3 種: on-confirm、 on-publish 、 no-acko 默認(rèn)為 on-confirm，表示在接收到下游的確認(rèn)消息之后再向上游發(fā)送消息確認(rèn)，這個選項可以確保網(wǎng)絡(luò)失敗或者 Broker 密機(jī)時不會丟失消息，但也是處理速度最慢的選項。如果設(shè)置為 on-publish ，則表示消息發(fā)送到下游后(井需要等待下游的 Basic . Ack)再向上游發(fā)送消息確認(rèn)，這個選項可以確保在網(wǎng)絡(luò)失敗的情況下不會丟失消息，但不能確保 Broker 巖機(jī)時不會丟失消息。 no-ack 表示無須進(jìn)行消息確認(rèn)，這個選項處理速度最快，但也最容易丟失消息。
• Expires：連接斷開后，上游隊列的超時時間，默認(rèn)為none，表示不刪除，單位為ms。相當(dāng)于設(shè)置隊列的x-expires參數(shù)，設(shè)置該值可以避免連接斷開后，生產(chǎn)者一直向上游隊列發(fā)送消息，造成上游大量消息堆積。

然后定義一個 Policy， 用于匹配交換器：

11.png

^exchange意思是將匹配所有以exchange名字開頭的交換器，為它們在上游創(chuàng)建連接。這樣就創(chuàng)建了一個 Federation link。

3、Shovel

Shovel是RabbitMQ的一個插件， 能夠可靠、持續(xù)地從一個Broker 中的隊列（作為源端，即source ）拉取數(shù)據(jù)并轉(zhuǎn)發(fā)至另一個Broker 中的交換器（作為目的端，即destination ）。作為源端的隊列和作為目的端的交換器可以同時位于同一個 Broker 上，也可以位于不同的 Broker 上。

使用Shovel有以下優(yōu)勢：

• 松耦合，解決不同Broker、集群、用戶、vhost、MQ和Erlang版本之間的消息移動
• 支持廣域網(wǎng)，基于 AMQP 0-9-1 協(xié)議實現(xiàn)，可以容忍糟糕的網(wǎng)絡(luò)，允許連接斷開的同時不丟失消息
• 高度定制，當(dāng)Shovel成功連接后，可以配置

使用Shovel時，通常源為隊列，目的為交換器：

12.jpeg

但是，也可以源為隊列，目的為隊列。實際也是由交換器轉(zhuǎn)發(fā)，只不過這個交換器是默認(rèn)交換器。配置交換器做為源也是可行的。實際上會在源端自動新建一個隊列，消息先存在這個隊列，再被Shovel移走。

使用Shovel插件命令：

##啟用插件
rabbitmq-plugins enable rabbitmq_shovel
##啟用管理插件
rabbitmq-plugins enable rabbitmq_shovel_management

Shovel 既可以部署在源端，也可以部署在目的端。有兩種方式可以部署 Shovel：

• 靜態(tài)方式：在 rabbitmq.config 配置文件中設(shè)置
• 動態(tài)方式：通過 Runtime Parameter 設(shè)置

其主要差異如下：

來看一個使用Shovel治理消息堆積的案例。

當(dāng)某個隊列中的消息堆積嚴(yán)重時，比如超過某個設(shè)定的閾值，就可以通過 Shovel 將隊列中的消息移交給另一個集群。

13.png

• 情形 1：當(dāng)檢測到當(dāng)前運(yùn)行集群 cluster1 中的隊列 queue1 中有嚴(yán)重消息堆積，比如超過2 千萬或者消息占用大小(messages bytes) 超過10GB 時，就啟用 shovel1 將隊列 queue1 中的消息轉(zhuǎn)發(fā)至備份集群 cluster2 中的隊列queue2 。
• 情形 2 ：緊隨情形1，當(dāng)檢測到隊列queue1 中的消息個數(shù)低于1 百萬或者消息占用大小低于1GB 時就停止shovel1 ，然后讓原本隊列 queue1 中的消費(fèi)者慢慢處理剩余的堆積。
• 情形 3：當(dāng)檢測到隊列 queue1 中的消息個數(shù)低于10 萬或者消息占用大小低于100MB時，就開啟 shovel2 將隊列 queue2 中暫存的消息返還給隊列queue1 。
• 情形 4：緊隨情形3 ，當(dāng)檢測到隊列queuel 中的消息個數(shù)超過 1百萬或者消息占用大小高于1GB 時就將shovel2 停掉。