46      技術社區[雲棲]

RabbitMQ高級指南：從配置、使用到高可用集群搭

本文大綱：

1. RabbitMQ簡介

2. RabbitMQ安裝與配置

3. C# 如何使用RabbitMQ

4. 幾種Exchange模式

5. RPC 遠程過程調用

6. RabbitMQ高可用集群搭建

一、RabbitMQ簡介

1、介紹

RabbitMQ是一個由erlang開發的基於AMQP（Advanced Message Queue ）協議的開源實現。用於在分布式係統中存儲轉發消息，在易用性、擴展性、高可用性等方麵都非常的優秀。是當前最主流的消息中間件之一。

RabbitMQ官網：https://www.rabbitmq.com

2、AMQP

AMQP是應用層協議的一個開放標準，為麵向消息的中間件設計。消息中間件主要用於組件之間的解耦，消息的發送者無需知道消息使用者的存在，同樣，消息使用者也不用知道發送者的存在。AMQP的主要特征是麵向消息、隊列、路由（包括點對點和發布/訂閱）、可靠性、安全。

3、係統架構 

消息隊列的使用過程大概如下：

（1）客戶端連接到消息隊列服務器，打開一個channel。

（2）客戶端聲明一個exchange，並設置相關屬性。

（3）客戶端聲明一個queue，並設置相關屬性。

（4）客戶端使用routing key，在exchange和queue之間建立好綁定關係。

（5） 客戶端投遞消息到exchange。exchange接收到消息後，就根據消息的

key和已經設置的binding，進行消息路由，將消息投遞到一個或多個隊列裏。

如上圖所示：AMQP裏主要說兩個組件：Exchange和Queue。綠色的X就是Exchange ，紅色的是Queue ，這兩者都在Server端，又稱作Broker，這部分是RabbitMQ實現的，而藍色的則是客戶端，通常有Producer和Consumer兩種類型。

4、幾個概念

• P： 為Producer，數據的發送方。

• C：為Consumer，數據的接收方。

• Exchange：消息交換機，它指定消息按什麼規則，路由到哪個隊列。

• Queue：消息隊列載體，每個消息都會被投入到一個或多個隊列。

• Binding：綁定，它的作用就是把exchange和queue按照路由規則綁定起來。

• Routing Key：路由關鍵字，exchange根據這個關鍵字進行消息投遞。

• vhost：虛擬主機，一個broker裏可以開設多個vhost，用作不同用戶的權限分離。

• channel：消息通道，在客戶端的每個連接裏，可建立多個channel，每個channel代表一個會話任務。

二、RabbitMQ安裝與配置

1、安裝

RabbitMQ是建立在強大的Erlang OTP平台上，因此安裝RabbitMQ之前要先安裝Erlang。

erlang：https://www.erlang.org/download.html

RabbitMQ：https://www.rabbitmq.com/download.html

注意：

• 現在先別裝最新的 3.6.3 ，本人在安裝完最新的版本，queue 隊列有問題，降到了 3.6.2 就解決了。

• 默認安裝的RabbitMQ監聽端口是：5672

2、配置

（1）安裝完以後erlang需要手動設置ERLANG_HOME 的係統變量。

輸入：set ERLANG_HOME=C:\Program Files\erl8.0

（2）激活RabbitMQ's Management Plugin

使用Rabbit MQ 管理插件，可以更好的可視化方式查看Rabbit MQ 服務器實例的狀態，你可以在命令行中使用下麵的命令激活。

輸入：rabbitmq-plugins.bat  enable  rabbitmq_management　

同時，我們也使用rabbitmqctl控製台命令（位於rabbitmq_server-3.6.3\sbin>）來創建用戶、密碼、綁定權限等。

（3）創建管理用戶

輸入：rabbitmqctl.bat add_user zhangweizhong weizhong1988　　　

（4）設置管理員

輸入：rabbitmqctl.bat set_user_tags zhangweizhong administrator　　

（5）設置權限

輸入：rabbitmqctl.bat set_permissions -p / zhangweizhong ".*" ".*" ".*"　

（6）其它命令

• 查詢用戶： rabbitmqctl.bat list_users

• 查詢vhosts： rabbitmqctl.bat list_vhosts

• 啟動RabbitMQ服務: net stop RabbitMQ && net start RabbitMQ

以上這些，賬號、vhost、權限、作用域等基本就設置完了。

3、Rabbit MQ管理後台

使用瀏覽器打開https://localhost:15672 訪問RabbitMQ的管理控製台，使用剛才創建的賬號登陸係統即可。

RabbitMQ管理後台，可以更好的可視化方式查看RabbitMQ服務器實例的狀態。　　

4、創建vhosts

創建vhosts，在admin頁麵，點擊右側Virtual Hosts：

將剛創建的OrderQueue分配給相關用戶。

其它創建exchange 、queue大家自己在後台創建吧，這裏不再贅述。

三、C# 如何使用RabbitMQ

1、客戶端

RabbitMQ.Client 是RabbitMQ官方提供的的客戶端，net版本地址 ：https://www.rabbitmq.com/dotnet.html　

EasyNetQ 是基於RabbitMQ.Client 基礎上封裝的開源客戶端。使用非常方便。地址：https://easynetq.com/ 。 

本篇使用示例代碼下載地址：

demo示例下載（https://files.cnblogs.com/files/zhangweizhong/Weiz.RabbitMQ.RPC.rar）

RabbitMQ 還有很多其它客戶端API，都非常好用。我們一直用的都是 EasyNetQ，所以這裏的demo隻介紹EasyNetQ客戶端實現。

2、項目結構

說明：前麵我們提到過，RabbitMQ由Producer（生成者） 和Consumer（消費者）兩部分組成。Weiz.Consumer就是Consumer（消費者），Weiz. Producer為Producer（生成者），Weiz.MQ為消息隊列的通用處理類庫。

3、項目搭建

（1）Weiz.MQ 項目，消息隊列的通用處理類庫，用於正在的訂閱和發布消息。

• 通過nuget安裝項目EasyNetQ 相關組件（略）

• 增加BusBuilder.cs管道創建類，主要負責鏈接RabbitMQ。

• 增加IProcessMessage類，定義了一個消息方法，用於消息傳遞

• 增加Message類，定義了消息傳遞的實體屬性字段等信息

• 增加MQHelper類，用於正在的訂閱和發布消息。 

（2）RabbitMQ由Producer（生成者）

創建一個aspx 頁麵，增加如下代碼：

（3）Weiz.Consumer 就是Consumer（消費者）

• 新增OrderProcessMessage.cs

• Program 增加如下代碼：

4、運行

（1）啟動 Weiz.Consumer （消費者），啟動消費者，會自動在RabbitMQ 服務器上創建相關的exchange 和 queue 。　

Consumer 消費者，使用的是Subscribe (訂閱)的模式，所以，Weiz.Consumer客戶端啟動後，會自動創建connection，生成相關的exchange 和queue。

（2）啟動Weiz. Producer 裏的TestMQ.aspx 頁麵，往隊列裏麵寫一條消息。訂閱的消費者立馬就能拿到這條消息。

至此，C#向RabbitMQ消息隊列發送消息已經簡單完成。

四、幾種Exchange模式

本節說些理論的東西——Exchange 的幾種模式。

AMQP協議中的核心思想就是生產者和消費者隔離，生產者從不直接將消息發送給隊列。生產者通常不知道是否一個消息會被發送到隊列中，隻是將消息發送到一個交換機。先由Exchange來接收，然後Exchange按照特定的策略轉發到Queue進行存儲。同理，消費者也是如此。Exchange 就類似於一個交換機，轉發各個消息分發到相應的隊列中。

RabbitMQ提供了四種Exchange模式：fanout、direct、topic、header 。 header模式在實際使用中較少，本文隻對前三種模式進行比較。

1、Fanout Exchange

所有發送到Fanout Exchange的消息都會被轉發到與該Exchange 綁定(Binding)的所有Queue上。

Fanout Exchange  不需要處理RouteKey 。隻需要簡單的將隊列綁定到exchange 上。這樣發送到exchange的消息都會被轉發到與該交換機綁定的所有隊列上。類似子網廣播，每台子網內的主機都獲得了一份複製的消息。

所以，Fanout Exchange 轉發消息是最快的。

2、Direct Exchange　

所有發送到Direct Exchange的消息被轉發到RouteKey中指定的Queue。

Direct模式可以使用RabbitMQ自帶的Exchange：default Exchange 。所以不需要將Exchange進行任何綁定(binding)操作 。消息傳遞時，RouteKey必須完全匹配，才會被隊列接收，否則該消息會被拋棄。

3、Topic Exchange

所有發送到Topic Exchange的消息被轉發到所有關心RouteKey中指定Topic的Queue上，Exchange 將RouteKey 和某Topic 進行模煳匹配。此時隊列需要綁定一個Topic。可以使用通配符進行模煳匹配，符號“#”匹配一個或多個詞，符號“*”匹配不多不少一個詞。因此“log.#”能夠匹配到“log.info.oa”，但是“log.*” 隻會匹配到“log.error”。

所以，Topic Exchange 使用非常靈活。

這個是RabbitMQ 的實際使用的幾個場景，熟悉了這個，基本上RabbitMQ也就了解了。https://www.rabbitmq.com/tutorials/tutorial-one-dotnet.html

至此，RabbitMQ幾種Exchange 模式已經介紹完了，實際使用過程中，我們會根據不同的場景，來使用不同的exchange模式。

五、RPC 遠程過程調用

最近有朋友問我，RabbitMQ RPC 是幹嘛的、有什麼用。其實，RabbitMQ RPC就是通過消息隊列（Message Queue）來實現RPC的功能，就是客戶端向服務端發送定義好的Queue消息，其中攜帶的消息就應該是服務端將要調用的方法的參數 ，並使用Propertis告訴服務端將結果返回到指定的Queue。

1、RabbitMQ RPC的特點

• Message Queue把所有的請求消息存儲起來，然後處理，和客戶端解耦。

• Message Queue引入新的結點，係統的可靠性會受Message Queue結點的影響。

• Message Queue是異步單向的消息。發送消息設計成是不需要等待消息處理的完成。

所以對於有同步返回需求的，Message Queue是個不錯的方向。

2、普通PRC的特點

• 同步調用，對於要等待返回結果/處理結果的場景，RPC是可以非常自然直覺的使用方式。當然RPC也可以是異步調用。

• 由於等待結果，客戶端會有線程消耗。

如果以異步RPC的方式使用，客戶端線程消耗可以去掉。但不能做到像消息一樣暫存消息請求，壓力會直接傳導到服務端。

3、適用場合說明

• 希望同步得到結果的場合，RPC合適。

• 希望使用簡單，則RPC；RPC操作基於接口，使用簡單，使用方式模擬本地調用。異步的方式編程比較複雜。

• 不希望客戶端受限於服務端的速度等，可以使用Message Queue。

4、RabbitMQ RPC工作流程

基本概念：

Callback queue 回調隊列，客戶端向服務器發送請求，服務器端處理請求後，將其處理結果保存在一個存儲體中。而客戶端為了獲得處理結果，那麼客戶在向服務器發送請求時，同時發送一個回調隊列地址reply_to。

Correlation id 關聯標識，客戶端可能會發送多個請求給服務器，當服務器處理完後，客戶端無法辨別在回調隊列中的響應具體和那個請求時對應的。為了處理這種情況，客戶端在發送每個請求時，同時會附帶一個獨有correlation_id屬性，這樣客戶端在回調隊列中根據correlation_id字段的值就可以分辨此響應屬於哪個請求。

流程說明：

• 當客戶端啟動的時候，它創建一個匿名獨享的回調隊列。

• 在 RPC 請求中，客戶端發送帶有兩個屬性的消息：一個是設置回調隊列的 reply_to 屬性，另一個是設置唯一值的 correlation_id 屬性。

• 將請求發送到一個 rpc_queue 隊列中。

• 服務器等待請求發送到這個隊列中來。當請求出現的時候，它執行他的工作並且將帶有執行結果的消息發送給 reply_to 字段指定的隊列。

• 客戶端等待回調隊列裏的數據。當有消息出現的時候，它會檢查 correlation_id 屬性。如果此屬性的值與請求匹配，將它返回給應用

5、完整代碼

（1）創建兩個控製台程序，作為RPC Server和RPC Client，引用RabbitMQ.Client

（2） RPC Server

（3）RPC Client

（4）分別運行Server和Client

六、RabbitMQ高可用集群

RabbitMQ是用erlang開發的，集群非常方便，因為erlang天生就是一門分布式語言，但其本身並不支持負載均衡。Rabbit模式大概分為以下三種：單一模式、普通模式、鏡像模式。

• 單一模式：最簡單的情況，非集群模式。

沒什麼好說的。

• 普通模式：默認的集群模式。

對於Queue來說，消息實體隻存在於其中一個節點，A、B兩個節點僅有相同的元數據，即隊列結構。

當消息進入A節點的Queue中後，consumer從B節點拉取時，RabbitMQ會臨時在A、B間進行消息傳輸，把A中的消息實體取出並經過B發送給consumer。

所以consumer應盡量連接每一個節點，從中取消息。即對於同一個邏輯隊列，要在多個節點建立物理Queue。否則無論consumer連A或B，出口總在A，會產生瓶頸。

該模式存在一個問題就是當A節點故障後，B節點無法取到A節點中還未消費的消息實體。

如果做了消息持久化，那麼得等A節點恢複，然後才可被消費；如果沒有持久化的話，然後就沒有然後了……

• 鏡像模式：把需要的隊列做成鏡像隊列，存在於多個節點，屬於RabbitMQ的HA方案。

該模式解決了上述問題，其實質和普通模式不同之處在於，消息實體會主動在鏡像節點間同步，而不是在consumer取數據時臨時拉取。

該模式帶來的副作用也很明顯，除了降低係統性能外，如果鏡像隊列數量過多，加之大量的消息進入，集群內部的網絡帶寬將會被這種同步通訊大大消耗掉。

所以在對可靠性要求較高的場合中適用（後麵會詳細介紹這種模式，目前我們搭建的環境屬於該模式）。

1、集群中的基本概念

RabbitMQ的集群節點包括內存節點、磁盤節點。顧名思義內存節點就是將所有數據放在內存，磁盤節點將數據放在磁盤。不過，如前文所述，如果在投遞消息時，打開了消息的持久化，那即使是內存節點，數據還是安全的放在磁盤。

一個RabbitMQ集群中可以共享user、vhost、queue、exchange等，所有的數據和狀態都是必須在所有節點上複製的，一個例外是那些當前隻屬於創建它的節點的消息隊列，盡管它們可見且可被所有節點讀取。RabbitMQ節點可以動態地加入到集群中，一個節點它可以加入到集群中，也可以從集群環集群進行一個基本的負載均衡。

集群中有兩種節點：

• 內存節點：隻保存狀態到內存（一個例外的情況是：持久的queue的持久內容將被保存到disk）

• 磁盤節點：保存狀態到內存和磁盤。
  內存節點雖然不寫入磁盤，但是它執行比磁盤節點要好。集群中，隻需要一個磁盤節點來保存狀態 就足夠了
  如果集群中隻有內存節點，那麼不能停止它們，否則所有的狀態，消息等都會丟失。

思路：

那麼具體如何實現RabbitMQ高可用，我們先搭建一個普通集群模式，在這個模式基礎上再配置鏡像模式實現高可用，Rabbit集群前增加一個反向代理，生產者、消費者通過反向代理訪問RabbitMQ集群。

架構圖如下：

圖片來自https://www.nsbeta.info

上述圖裏是3個RabbitMQ運行在同一主機上，分別用不同的服務端口。當然我們的生產實際裏，多個RabbitMQ肯定是運行在不同的物理服務器上，否則就失去了高可用的意義。

 2、集群模式配置

設計架構可以如下：在一個集群裏，有4台機器，其中1台使用磁盤模式，另2台使用內存模式。2台內存模式的節點，無疑速度更快，因此客戶端（consumer、producer）連接訪問它們。而磁盤模式的節點，由於磁盤IO相對較慢，因此僅作數據備份使用，另外一台作為反向代理。

四台服務器hostname分別為：queue 、panyuntao1、panyuntao2、panyuntao3（ip:172.16.3.110）

配置RabbitMQ集群非常簡單，隻需要幾個命令，配置步驟如下：

Step 1：

queue、panyuntao1、panyuntao2做為RabbitMQ集群節點，分別安裝RabbitMq-Server ，安裝後分別啟動RabbitMq-server。

啟動命令  # Rabbit-Server start ，安裝過程及啟動命令參見：https://www.cnblogs.com/flat_peach/archive/2013/03/04/2943574.html

Step 2：

在安裝好的三台節點服務器中，分別修改/etc/hosts文件，指定queue、panyuntao1、panyuntao2的hosts，如：

172.16.3.32 queue

172.16.3.107 panyuntao1

172.16.3.108 panyuntao2

還有hostname文件也要正確，分別是queue、panyuntao1、panyuntao2，如果修改hostname建議安裝rabbitmq前修改。

請注意RabbitMQ集群節點必須在同一個網段裏，如果是跨廣域網效果就差。

Step 3：設置每個節點Cookie

RabbitMQ的集群是依賴於erlang的集群來工作的，所以必須先構建起erlang的集群環境。Erlang的集群中各節點是通過一個magic cookie來實現的，這個cookie存放在 /var/lib/rabbitmq/.erlang.cookie 中，文件是400的權限。所以必須保證各節點cookie保持一致，否則節點之間就無法通信。

-r--------. 1 rabbitmq rabbitmq 20 3月 5 00:00 /var/lib/rabbitmq/.erlang.cookie

將其中一台節點上的.erlang.cookie值複製下來保存到其他節點上。或者使用scp的方法也可，但是要注意文件的權限和屬主屬組。

我們這裏將queue中的cookie 複製到 panyuntao1、panyuntao2中，先修改下panyuntao1、panyuntao2中的.erlang.cookie權限

將queue的/var/lib/rabbitmq/.erlang.cookie這個文件，拷貝到panyuntao1、panyuntao2的同一位置（反過來亦可），該文件是集群節點進行通信的驗證密鑰，所有節點必須一致。拷完後重啟下RabbitMQ。

複製好後別忘記還原.erlang.cookie的權限，否則可能會遇到錯誤

設置好cookie後先將三個節點的RabbitMQ重啟

# rabbitmqctl stop

# rabbitmq-server start

Step 4：

停止所有節點RabbitMQ服務，然後使用detached參數獨立運行，這步很關鍵，尤其增加節點停止節點後再次啟動遇到無法啟動都可以參照這個順序。

分別查看下每個節點

Step 5：

將panyuntao1、panyuntao2作為內存節點與queue連接起來，在panyuntao1上，執行如下命令：

 (上方已經將panyuntao1與queue連接，也可以直接將panyuntao2與panyuntao1連接，同樣而已加入集群中)

上述命令先停掉RabbitMQ應用，然後調用cluster命令，將panyuntao1連接到，使兩者成為一個集群，最後重啟RabbitMQ應用。在這個cluster命令下，panyuntao1、panyuntao2是內存節點，queue是磁盤節點（RabbitMQ啟動後，默認是磁盤節點）。

queue 如果要使panyuntao1或panyuntao2在集群裏也是磁盤節點，join_cluster 命令去掉--ram參數即可。

隻要在節點列表裏包含了自己，它就成為一個磁盤節點。在RabbitMQ集群裏，必須至少有一個磁盤節點存在。

Step 6：

在queue、panyuntao1、panyuntao2上，運行cluster_status命令查看集群狀態：

這時我們可以看到每個節點的集群信息，分別有兩個內存節點一個磁盤節點。

Step 7:

往任意一台集群節點裏寫入消息隊列，會複製到另一個節點上，我們看到兩個節點的消息隊列數一致：

-p參數為vhost名稱

這樣RabbitMQ集群就正常工作了。

 這種模式更適合非持久化隊列，隻有該隊列是非持久的，客戶端才能重新連接到集群裏的其他節點，並重新創建隊列。假如該隊列是持久化的，那麼唯一辦法是將故障節點恢複起來。  

為什麼RabbitMQ不將隊列複製到集群裏每個節點呢？這與它的集群的設計本意相衝突，集群的設計目的就是增加更多節點時，能線性的增加性能（CPU、內存）和容量（內存、磁盤）。理由如下：

1. storage space: If every cluster node had a full copy of every queue, adding nodes wouldn’t give you more storage capacity. For example, if one node could store 1GB of messages, adding two more nodes would simply give you two more copies of the same 1GB of messages.

2. performance: Publishing messages would require replicating those messages to every cluster node. For durable messages that would require triggering disk activity on all nodes for every message. Your network and disk load would increase every time you added a node, keeping the performance of the cluster the same (or possibly worse).

當然RabbitMQ新版本集群也支持隊列複製（有個選項可以配置）。比如在有五個節點的集群裏，可以指定某個隊列的內容在2個節點上進行存儲，從而在性能與高可用性之間取得一個平衡。

3、鏡像模式配置

上麵配置RabbitMQ默認集群模式，但並不保證隊列的高可用性，盡管交換機、綁定這些可以複製到集群裏的任何一個節點，但是隊列內容不會複製，雖然該模式解決一部分節點壓力，但隊列節點宕機直接導致該隊列無法使用，隻能等待重啟，所以要想在隊列節點宕機或故障也能正常使用，就要複製隊列內容到集群裏的每個節點，需要創建鏡像隊列。

我們看看如何鏡像模式來解決複製的問題，從而提高可用性 。

 Step 1：增加負載均衡器 

關於負載均衡器，商業的比如F5的BIG-IP，Radware的AppDirector，是硬件架構的產品，可以實現很高的處理能力。但這些產品昂貴的價格會讓人止步，所以我們還有軟件負載均衡方案。互聯網公司常用的軟件LB一般有LVS、HAProxy、Nginx等。LVS是一個內核層的產品，主要在第四層負責數據包轉發，使用較複雜。HAProxy和Nginx是應用層的產品，但Nginx主要用於處理HTTP，所以這裏選擇HAProxy作為RabbitMQ前端的LB。

HAProxy的安裝使用非常簡單，在Centos下直接yum install haproxy，然後更改/etc/haproxy/haproxy.cfg 文件即可，文件內容大概如下：

負載均衡器會監聽5672端口，輪詢我們的兩個內存節點172.16.3.107、172.16.3.108的5672端口，172.16.3.32為磁盤節點，隻做備份不提供給生產者、消費者使用，當然如果我們服務器資源充足情況也可以配置多個磁盤節點

，這樣磁盤節點除了故障也不會影響，除非同時出故障。

Step 2：配置策略 

使用Rabbit鏡像功能，需要基於RabbitMQ策略來實現，政策是用來控製和修改群集範圍的某個vhost隊列行為和Exchange行為。

在cluster中任意節點啟用策略，策略會自動同步到集群節點

這行命令在vhost名稱為hrsystem創建了一個策略，策略名稱為ha-allqueue,策略模式為 all 即複製到所有節點，包含新增節點，策略正則表達式為 “^” 表示所有匹配所有隊列名稱。

注意："^message" 這個規則要根據自己修改，這個是指同步"message"開頭的隊列名稱，我們配置時使用的應用於所有隊列，所以表達式為"^"

官方set_policy說明參見

set_policy [-p vhostpath] {name} {pattern} {definition} [priority]

（https://www.rabbitmq.com/man/rabbitmqctl.1.man.html）

ha-mode：

Step 3：

創建隊列時需要指定ha 參數，如果不指定x-ha-prolicy 的話將無法複製。

下麵為C#代碼片段：

Step 4：

客戶端使用負載服務器172.16.3.110 （panyuntao3）發送消息，隊列會被複製到所有節點，當然策略也可以配置製定某幾個節點，這時任何節點故障 、或者重啟將不會影響我們正常使用某個隊列。到這裏我們完成了高可用配置（所有節點都宕機那沒有辦法了）。

使用RabbitMQ管理端可以看到集群鏡像模式中對列狀態

原文發布時間為：2017-04-27

本文來自雲棲社區合作夥伴DBAplus

最後更新：2017-05-17 13:35:25

  上一篇：  org.springframework.beans簡單解讀

  下一篇：  半年的讀書計劃有保證了