玩轉不同業務場景，這些RabbitMQ特性會是得力助攻

RabbitMQ · 發表 2018-12-07 11:13:58

摘要：我是在解決分散式事務的一致性問題時瞭解到的RabbitMQ，當時主要是要基於RabbitMQ來實現我們分散式系統之間對有事務可靠性要求的系統間通訊。提到RabbitMQ，不難想到的幾個關鍵字：訊息中介軟體、訊息佇列。當時在大學學習作業系統這門課，訊息佇列不難想到生產者消費者模式。...

我是在解決分散式事務的一致性問題時瞭解到的RabbitMQ，當時主要是要基於RabbitMQ來實現我們分散式系統之間對有事務可靠性要求的系統間通訊。

提到RabbitMQ，不難想到的幾個關鍵字：訊息中介軟體、訊息佇列。當時在大學學習作業系統這門課，訊息佇列不難想到生產者消費者模式。

(PS：作業系統這門課程真的很好也很重要，其中的一些思想在我工作的很長一段時間內給了我很大幫助和啟發，給我提供了許多解決問題的思路。強烈建議每一個程式設計師都去學一學作業系統。)

一、訊息中介軟體

1、簡介

訊息中介軟體也可以稱訊息佇列，是指用高效可靠的訊息傳遞機制進行與平臺無關的資料交流，並基於資料通訊來進行分散式系統的整合。通過提供訊息傳遞和訊息佇列模型，可以在分散式環境下擴充套件程序的通訊。

當下主流的訊息中介軟體有RabbitMQ、Kafka、ActiveMQ、RocketMQ等。其能在不同平臺之間進行通訊，常用來遮蔽各種平臺協議之間的特性，實現應用程式之間的協同。優點在於能夠在客戶端和伺服器之間進行同步和非同步的連線，並且在任何時刻都可以將訊息進行傳送和轉發，是分散式系統中非常重要的元件，主要用來解決應用耦合、非同步通訊、流量削峰等問題。

2、作用

訊息中介軟體幾大主要作用如下：

解耦
冗餘(儲存)
擴充套件性
削峰
可恢復性
順序保證
緩衝
非同步通訊

3、訊息中介軟體的兩種模式

P2P模式

P2P模式包含三個角色：訊息佇列（Queue）、傳送者(Sender)、接收者(Receiver)。每個訊息都被髮送到一個特定的佇列，接收者從佇列中獲取訊息。佇列保留著訊息，直到它們被消費或超時。

P2P的特點：

每個訊息只有一個消費者（Consumer），即一旦被消費，訊息就不再在訊息佇列中
傳送者和接收者之間在時間上沒有依賴性，也就是說當傳送者傳送了訊息之後，不管接收者有沒有正在執行它不會影響到訊息被髮送到佇列
接收者在成功接收訊息之後需向佇列應答成功
如果希望傳送的每個訊息都會被成功處理的話，那麼需要P2P模式

Pub/Sub模式

Pub/Sub模式包含三個角色：主題（Topic）、釋出者（Publisher）、訂閱者（Subscriber）。多個釋出者將訊息傳送到Topic，系統將這些訊息傳遞給多個訂閱者。

Pub/Sub的特點：

每個訊息可以有多個消費者
釋出者和訂閱者之間有時間上的依賴性。針對某個主題（Topic）的訂閱者，它必須建立一個訂閱者之後，才能消費釋出者的訊息
為了消費訊息，訂閱者必須保持執行的狀態
如果希望傳送的訊息可以不被做任何處理、或者只被一個訊息者處理、或者可以被多個消費者處理的話，那麼可以採用Pub/Sub模型

4、常用中介軟體介紹與對比

Kafka

Kafka是LinkedIn開源的分散式釋出-訂閱訊息系統，目前歸屬於Apache頂級專案。Kafka主要特點是基於Pull的模式來處理訊息消費，追求高吞吐量，一開始的目的就是用於日誌收集和傳輸。0.8版本開始支援複製，不支援事務，對訊息的重複、丟失、錯誤沒有嚴格要求，適合產生大量資料的網際網路服務的資料收集業務。

RabbitMQ

RabbitMQ是使用Erlang語言開發的開源訊息佇列系統，基於AMQP協議來實現。AMQP的主要特徵是面向訊息、佇列、路由（包括點對點和釋出/訂閱）、可靠性、安全。AMQP協議更多用在企業系統內對資料一致性、穩定性和可靠性要求很高的場景，對效能和吞吐量的要求還在其次。

RocketMQ

RocketMQ是阿里開源的訊息中介軟體，它是純Java開發，具有高吞吐量、高可用性、適合大規模分散式系統應用的特點。RocketMQ思路起源於Kafka，但並不是Kafka的一個Copy，它對訊息的可靠傳輸及事務性做了優化，目前在阿里集團被廣泛應用於交易、充值、流計算、訊息推送、日誌流式處理、binglog分發等場景。

RabbitMQ比Kafka可靠，Kafka更適合IO高吞吐的處理，一般應用在大資料日誌處理或對實時性（少量延遲），可靠性（少量丟資料）要求稍低的場景使用，比如ELK日誌收集。

二、RabbitMQ瞭解

1、簡介

RabbitMQ是流行的開源訊息佇列系統，是AMQP（高階訊息佇列協議）的標準實現。它支援多種客戶端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支援AJAX，持久化，用於在分散式系統中儲存轉發訊息，在易用性、擴充套件性、高可用性等方面表現不俗。

RabbitMQ是使用Erlang編寫的一個開源的訊息佇列，本身支援很多的協議：AMQP，XMPP, SMTP, STOMP，也正是如此，使的它變的非常重量級，更適合於企業級的開發。它同時實現了一個Broker構架，這意味著訊息在傳送給客戶端時先在中心佇列排隊，對路由(Routing)、負載均衡(Load balance)或者資料持久化都有很好的支援。

其主要特點如下：

可靠性
靈活的路由
擴充套件性
高可用性
多種協議
多語言客戶端
管理介面
外掛機制

2、概念

RabbitMQ從整體上來看是一個典型的生產者消費者模型，主要負責接收、儲存和轉發訊息。其整體模型架構如下圖所示：

我們先來看一個RabbitMQ的運轉流程，稍後會對這個流程中所涉及到的一些概念進行詳細的解釋。

生產者：

生產者連線到RabbitMQ Broker，建立一個連線(Connection)開啟一個通道(Channel)
生產者宣告一個交換器，並設定相關屬性，比如交換機型別、是否持久化等
生產者宣告一個佇列井設定相關屬性，比如是否排他、是否持久化、是否自動刪除等
生產者通過路由鍵將交換器和佇列繫結起來
生產者傳送訊息至RabbitMQ Broker，其中包含路由鍵、交換器等資訊
相應的交換器根據接收到的路由鍵查詢相匹配的佇列
如果找到，則將從生產者傳送過來的訊息存入相應的佇列中
如果沒有找到，則根據生產者配置的屬性選擇丟棄還是回退給生產者
關閉通道
關閉連線

消費者：

消費者連線到RabbitMQ Broker ，建立一個連線(Connection)，開啟一個通道(Channel)
消費者向RabbitMQ Broker請求消費相應佇列中的訊息，可能會設定相應的回撥函式
等待RabbitMQ Broker迴應並投遞相應佇列中的訊息，消費者接收訊息
消費者確認(ack) 接收到的訊息
RabbitMQ從佇列中刪除相應己經被確認的訊息
關閉通道
關閉連線

通道

這裡我們主要討論兩個問題：

為何要有通道?

主要原因還是在於TCP連線的“昂貴”性。無論是生產者還是消費者，都需要和RabbitMQ Broker 建立連線，這個連線就是一條TCP連線。而作業系統對於TCP連線的建立與銷燬是非常昂貴的開銷。

假設消費者要消費訊息，並根據服務需求合理排程執行緒，若只進行TCP連線，那麼當高併發的時候，每秒可能都有成千上萬的TCP連線，不僅僅是對TCP連線的浪費，也很快會超過作業系統每秒所能建立連線的數量。如果能在一條TCP連線上操作，又能保證各個執行緒之間的私密性就完美了，於是通道的概念出現了。

通道為何?

通道是建立在Connection之上的虛擬連線。當應用程式與Rabbit Broker建立TCP連線的時候，客戶端緊接著可以建立一個AMQP通道(Channel) ，每個通道都會被指派一個唯一的ID。RabbitMQ 處理的每條AMQP指令都是通過通道完成的。通道就像電纜裡的光纖束，一條電纜內含有許多光纖束，允許所有的連線通過多條光線束進行傳輸和接收。

生產者消費者

關於生產者消費者我們需要了解幾個概念：

Producer：生產者，即訊息投遞者一方。
訊息：訊息一般分兩個部分——訊息體(payload)和標籤。標籤用來描述這條訊息，如：一個交換器的名稱或者一個路由Key，Rabbit通過解析標籤來確定訊息的去向，payload是訊息內容可以使一個json、陣列等等。
Consumer：消費者，就是接收訊息的一方。消費者訂閱RabbitMQ的佇列，當消費者消費一條訊息時，只是消費訊息的訊息體。在訊息路由的過程中，會丟棄標籤，存入到佇列中的只有訊息體。
Broker：訊息中介軟體的服務節點。

佇列、交換器、路由key、繫結

從RabbitMQ的運轉流程我們可以知道生產者的訊息是釋出到交換器上的。而消費者則是從佇列上獲取訊息的。那麼訊息到底是如何從交換器到佇列的呢？我們先具體瞭解一下這幾個概念：

Queue：佇列，是RabbitMQ的內部物件，用於儲存訊息。RabbitMQ中訊息只能儲存在佇列中，生產者投遞訊息到佇列，消費者從佇列中獲取訊息並消費。多個消費者可以訂閱同一個佇列，這時佇列中的訊息會被平均分攤(輪詢)給多個消費者進行消費，而不是每個消費者都收到所有的訊息進行消費。

注意：RabbitMQ不支援佇列層面的廣播消費，如果需要廣播消費，可以採用一個交換器通過路由Key繫結多個佇列，由多個消費者來訂閱這些佇列的方式。

Exchange：交換器。在RabbitMQ中，生產者並非直接將訊息投遞到佇列中。真實情況是，生產者將訊息傳送到Exchange(交換器)，由交換器將訊息路由到一個或多個佇列中。如果路由不到，或返回給生產者，或直接丟棄，或做其它處理。

RoutingKey：路由Key。生產者將訊息傳送給交換器的時候，一般會指定一個RoutingKey，用來指定這個訊息的路由規則。這個路由Key需要與交換器型別和繫結鍵(BindingKey)聯合使用才能最終生效。在交換器型別和繫結鍵固定的情況下，生產者可以在傳送訊息給交換器時通過指定RoutingKey來決定訊息流向哪裡。

Binding： RabbitMQ通過繫結將交換器和佇列關聯起來，在繫結的時候一般會指定一個繫結鍵，這樣RabbitMQ就可以指定如何正確的路由到隊列了。

從這裡我們可以看到在RabbitMQ中交換器和佇列實際上可以是一對多，也可以是多對多關係。交換器和佇列就像我們關係資料庫中的兩張表。它們同歸BindingKey做關聯(多對多關係表)。在我們投遞訊息時，可以通過Exchange和RoutingKey(對應BindingKey)就可以找到相對應的佇列。

RabbitMQ主要有四種類型的交換器：

fanout：扇形交換器，它會把傳送到該交換器的訊息路由到所有與該交換器繫結的佇列中。如果使用扇形交換器，則不會匹配路由Key。

direct： direct交換器，會把訊息路由到RoutingKey與BindingKey完全匹配的佇列中。

topic：完全匹配BindingKey和RoutingKey的direct交換器有些時候並不能滿足實際業務的需求。topic型別的交換器在匹配規則上進行了擴充套件，它與direct型別的交換器相似，也是將訊息路由到BindingKey和RoutingKey相匹配的佇列中，但這裡的匹配規則有些不同，它約定:

RoutingKey為一個點號"."分隔的字串(被點號"."分隔開的每一段獨立的字串稱為一個單詞)，如"hs.rabbitmq.client"，"com.rabbit.client"等。
BindingKey和RoutingKey一樣也是點號"."分隔的字串；
BindingKey中可以存在兩種特殊字串"*"和"#"，用於做模糊匹配，其中"*"用於匹配一個單詞，"#"用於匹配多規格單詞(可以是零個)。

如圖：

路由鍵為" apple.rabbit.client" 的訊息會同時路由到Queuel 和Queue2；
路由鍵為" orange.mq.client" 的訊息只會路由到Queue2 中；
路由鍵為" apple.mq.demo" 的訊息只會路由到Queue2中；
路由鍵為" banana.rabbit.demo" 的訊息只會路由到Queuel 中；
路由鍵為" orange.apple.banana" 的訊息將會被丟棄或者返回給生產者因為它沒有匹配任何路由鍵。

header： headers型別的交換器不依賴於路由鍵的匹配規則來路由訊息，而是根據傳送的訊息內容中的headers屬性進行匹配。在繫結佇列和交換器時制定一組鍵值對，當傳送訊息到交換器時，RabbitMQ會獲取到該訊息的headers (也是一個鍵值對的形式) ，對比其中的鍵值對是否完全匹配佇列和交換器繫結時指定的鍵值對，如果完全匹配則訊息會路由到該佇列，否則不會路由到該佇列。

注：該交換器型別效能較差且不實用，因此一般不會用到。

瞭解了上面的概念，我們再來思考訊息是如何從交換器到佇列的。首先Rabbit在接收到訊息時，會解析訊息的標籤從而得到訊息的交換器與路由Key資訊。然後根據交換器的型別、路由Key以及該交換器和佇列的繫結關係來決定訊息最終投遞到哪個佇列裡面。

參考

《RabbitMQ實戰指南》
《RabbitMQ實戰-高效部署分散式訊息佇列》

簡單總結了分散式系統中的訊息中介軟體以及RabbitMQ的基本使用後，我們接下來主要分享一下RabbitMQ在日常專案開發中比較常用的幾個特性。

三、mandatory與備份交換器

前文我們瞭解到，生產者將訊息傳送到RabbitMQ的交換器中通過RoutingKey與BindingKey的匹配將之路由到具體的佇列中，以供消費者消費。那麼當我們通過匹配規則找不到佇列的時候，訊息將何去何從呢？Rabbit給我們提供了兩種方式：mandatory與備份交換器。

1、mandatory引數

mandatory引數是channel.BasicPublish方法中的引數。其主要功能是訊息傳遞過程中不可達目的地時將訊息返回給生產者。當mandatory引數設為true 時，如果交換器無法根據自身的型別和路由鍵找到一個符合條件的佇列，那麼RabbitMQ會呼叫BasicReturn命令將訊息返回給生產者；當mandatory 引數設定為false 時，則訊息直接被丟棄。

其運轉流程與實現程式碼如下(以C# RabbitMQ.Client 3.6.9為例)：

運轉流程

2、備份交換器

當訊息不能路由到佇列時，通過mandatory設定引數，我們可以將訊息返回給生產者處理。但這樣會有一個問題，就是生產者需要開一個回撥的函式來處理不能路由到的訊息，這無疑會增加生產者的處理邏輯。

備份交換器(Altemate Exchange)則提供了另一種方式來處理不能路由的訊息。

備份交換器可以將未被路由的訊息儲存在RabbitMQ中，在需要的時候去處理這些訊息。其主要實現程式碼如下：

備份交換器其實和普通的交換器沒有太大的區別，為了方便使用，建議設定為fanout型別。

需要注意的是，訊息被重新發送到備份交換器時的路由鍵和從生產者發出的路由鍵是一樣的。

考慮這樣一種情況，如果備份交換器的型別是direct，並且有一個與其繫結的佇列，假設繫結的路由鍵是key1，當某條攜帶路由鍵為key2的訊息被轉發到這個備份交換器的時候，備份交換器沒有匹配到合適的佇列，則訊息丟失；如果訊息攜帶的路由鍵為key1，則可以儲存到佇列中。

對於備份交換器，有以下幾種特殊情況：

如果設定的備份交換器不存在，客戶端和RabbitMQ服務端都不會有異常出現，此時訊息會丟失；
如果備份交換器沒有繫結任何佇列，客戶端和RabbitMQ服務端都不會有異常出現，此時訊息會丟失；
如果備份交換器沒有任何匹配的佇列，客戶端和RabbitMQ服務端都不會有異常出現，此時訊息會丟失；
如果備份交換器和mandatory引數一起使用，那麼mandatory引數無效。

四、過期時間(TTL)

1、設定訊息的TTL

目前有兩種方法可以設定訊息的TTL：

第一種方法是通過佇列屬性設定，佇列中所有訊息都有相同的過期時間；
第二種方法是對訊息本身進行單獨設定，每條訊息的TTL可以不同。

如果兩種方法一起使用，則訊息的TTL以兩者之間較小的那個數值為準。訊息在佇列中的生存時間一旦超過設定的TTL值時，就會變成“死信 ” (Dead Message) ，消費者將無法再收到該訊息。(有關死信佇列請往下看)

通過佇列屬性設定訊息TTL的方法是在channel.QueueDeclare方法中加入x-message-ttl引數實現的，這個引數的單位是毫秒。示例程式碼下：

如果不設定TTL，則表示此訊息不會過期；如果將TTL設定為0 ，則表示除非此時可以直接將訊息投遞到消費者，否則該訊息會被立即丟棄(或由死信佇列來處理)。

針對每條訊息設定TTL的方法是在channel.BasicPublish方法中加入Expiration的屬性引數，單位為毫秒。關鍵程式碼如下：

注意：

對於第一種設定佇列TTL屬性的方法，一旦訊息過期，就會從佇列中抹去；而在第二種方法中，即使訊息過期，也不會馬上從佇列中抹去，因為每條訊息是否過期是在即將投遞到消費者之前判定的。

Why？

在第一種方法裡，佇列中己過期的訊息肯定在佇列頭部，RabbitMQ只要定期從隊頭開始掃描是否有過期的訊息即可；而第二種方法裡，每條訊息的過期時間不同，如果要刪除所有過期訊息勢必要掃描整個佇列，所以不如等到此訊息即將被消費時再判定是否過期，如果過期再進行刪除即可。

2、設定佇列的TTL

需要注意的是，這裡和上述通過佇列設定訊息的TTL不同。上面刪除的是訊息，而這裡刪除的是佇列。通過channel.QueueDeclare方法中的x-expires引數可以控制佇列被自動刪除前處於未使用狀態的時間。這個未使用的意思是佇列上沒有任何的消費者，佇列也沒有被重新宣告，並且在過期時間段內也未呼叫過channel.BasicGet命令。

設定佇列裡的TTL可以應用於類似RPC方式的回覆佇列，在RPC中，許多佇列會被創建出來，但是卻是未被使用的。RabbitMQ會確保在過期時間到達後將佇列刪除，但是不保障刪除的動作有多及時。

在RabbitMQ重啟後，持久化的佇列的過期時間會被重新計算。用於表示過期時間的x-expires引數以毫秒為單位，井且服從和x-message-ttl一樣的約束條件，只不過不同的是它不能設定為0(會報錯)。

示例程式碼如下：

五、死信佇列

DLX(Dead-Letter-Exchange)是指死信交換器。當訊息在一個佇列中變成死信之後，它能重新被髮送到另一個交換器中，這個交換器就是DLX ，繫結DLX的佇列就被稱為死信佇列。

訊息變成死信主要有以下幾種情況：

訊息被拒絕(BasicReject/BasicNack) ，井且設定requeue引數為false；(消費者確認機制我們將會在下一篇文章中涉及)
訊息過期；
佇列達到最大長度。

DLX也是一個正常的交換器，和一般的交換器沒有區別，它能在任何佇列上被指定，實際上就是設定某個佇列的屬性。

當這個佇列中存在死信時，RabbitMQ就會自動地將這個訊息重新發布到設定的DLX上去，進而被路由到另一個佇列，即死信佇列。可以監聽這個佇列中的訊息、以進行相應的處理。

通過在channel.QueueDeclare方法中設定x-dead-letter-exchange引數來為這個佇列新增DLX。其示例程式碼如下：

以下為死信佇列的運轉流程：

六、延遲佇列

RabbitMQ本身並未提供延遲佇列的功能。延遲佇列是一個邏輯上的概念，可以通過過期時間+死信佇列來模擬它的實現。延遲佇列的邏輯架構大致如下：

生產者將訊息傳送到過期時間為n的佇列中，這個佇列並未有消費者來消費訊息，當過期時間到達時，訊息會通過死信交換器被轉發到死信佇列中。而消費者從死信佇列中消費訊息。這個時候就達到了“生產者釋出了訊息，在過了n時間後消費者消費了訊息 ” ，起到了延遲消費的作用。

延遲佇列在我們的專案中可以應用於很多場景，如：下單後兩個訊息取消訂單，七天自動收貨，七天自動好評，密碼凍結後24小時解凍，以及在分散式系統中訊息補償機制(1s後補償，10s後補償，5m後補償......)。

七、優先順序佇列

就像我們生活中的“特殊”人士一樣，我們的業務上也存在一些“特殊”訊息，可能需要優先進行處理，在生活上我們可能會對這部分特殊人士開闢一套VIP通道，而Rabbit同樣也有這樣的VIP通道(前提是在3.5的版本以後)，即優先順序佇列。佇列中的訊息會有優先順序，優先順序高的訊息具備優先被消費的特權。針對這些VIP訊息，我們只需做兩件事：

將佇列宣告為優先順序佇列，即在建立佇列的時候新增引數x-max-priority指定最大的優先順序，值為0-255（整數）。
為優先順序訊息新增優先順序。

其示例程式碼如下：

注意：

沒有指定優先順序的訊息會將優先順序以0對待。
對於超過優先順序佇列所定最大優先順序的訊息，優先順序以最大優先順序對待。
對於相同優先順序的訊息，後進的排在前面。

如果在消費者的消費速度大於生產者的速度且Broker中沒有訊息堆積的情況下，對傳送的訊息設定優先順序也就沒有什麼實際意義。因為生產者剛傳送完一條訊息就被消費者消費了，那麼就相當於Broker中至多隻有一條訊息，對於單條訊息來說優先順序是沒有什麼意義的。

關於優先順序佇列，好像違背了佇列這種資料結構先進先出的原則，其具體是怎麼實現的在這裡就不過多討論。有興趣的可以自己研究研究。後續可能也會有相關的文章來分析其原理。

八、RPC實現

RPC是Remote Procedure Call的簡稱，即遠端過程呼叫。它是一種通過網路從遠端計算機上請求服務，不需要了解底層網路的技術。RPC的主要功用是讓構建分散式計算更容易，在提供強大的遠端呼叫能力時不損失本地呼叫的語義簡潔性。

有關RPC不多介紹，這裡我們主要介紹RabbitMQ如何實現RPC。

RabbitMQ可以實現很簡單的RPC。客戶端傳送請求訊息，服務端回覆響應的訊息。為了接收響應的訊息，我們需要在請求訊息中傳送一個回撥佇列(可以使用預設的佇列)。其伺服器端實現程式碼如下：

客戶端實現程式碼如下：

以上是Rabbit客戶端自己幫我們封裝好的Rpc客戶端與服務端的邏輯。當然我們也可以自己實現，主要是藉助於BasicProperties的兩個引數。

ReplyTo：通常用來設定一個回撥佇列。
CorrelationId：用來關聯請求(request) 和其呼叫RPC之後的回覆(response) 。

其處理流程如下：

當客戶端啟動時，建立一個匿名的回撥佇列。
客戶端為RPC請求設定2個屬性：ReplyTo用來告知RPC服務端回覆請求時的目的佇列，即回撥佇列；Correlationld用來標記一個請求。
請求被髮送到RpcQueue佇列中。
RPC服務端監聽RpcQueue佇列中的請求，當請求到來時，服務端會處理並把帶有結果的訊息傳送給客戶端。接收的佇列就是ReplyTo設定的回撥佇列。
客戶端監聽回撥佇列，當有訊息時，檢查Correlationld屬性，如果與請求匹配，那就是結果了。

九、結語

本文簡單介紹了RabbitMQ的基礎內容以及在我們專案開發中的幾種常用特性，這些特性可以幫助我們更好地將Rabbit用於不同的業務場景中。

對於這些特性與示例的內容，為了幫助理解，大家可以自己在程式中執行一下，然後通過檢視Rabbit提供的web管理介面來驗證其正確性。當然，關於Rabbit的使用，仍有許多地方在本文中沒有提及，如：RabbitMQ的特色——確認機制、持久化......我們下一篇文章再繼續詳細介紹。