深入淺出RabbitMQ：順序消費、死信隊列和延時隊列
大家好，我是小?，一個漂泊江湖多年的 985 非科班程序員，曾混跡于國企、互聯網大廠和創業公司的后臺開發攻城獅。
1. 引言
在今天的文章中，我們來聊一聊 RabbitMQ，這是小 ? 在工作中用的最早的消息中間件，主要用于大量數據的異步消費。
2. RabbitMQ
2.1 核心組件
RabbitMQ 是一個開源的消息中間件，它實現了高級消息隊列協議（AMQP），同時提供了各種重要組件來支持消息的生產、傳輸和消費。
圖片
- Producer（生產者）： 生產者是消息的發送方，負責將消息發布到 RabbitMQ 服務器。消息可以包含任何內容，例如任務、日志、通知等。
- Channel（信道）：消息推送與接收時使用的通道。
- Exchange（交換機）： 交換機是消息的中轉站，它接收來自生產者的消息并將其路由到一個或多個隊列。不同類型的交換機，如 fanout，direct，topic，headers，支持不同的路由規則。
- Queue（隊列）： 隊列是消息的緩沖區，消息在發送到消費者之前存儲在隊列中，消費者從隊列中獲取消息并進行處理。
- Consumer（消費者）： 消費者是消息的接收方，它從隊列中獲取消息并進行處理。消費者可以是多個，它們可以在不同的應用程序或服務器上運行。
2.2 工作流程
RabbitMQ 的工作方式是基于生產者、交換機和隊列之間的協作。這是一個簡單的消息傳遞過程：
- 將隊列與交換機綁定（Binding）起來，定義了消息的路由規則；
- 生產者將消息發布到交換機，交換機根據綁定規則將消息路由到一個或多個隊列；
- 消費者從隊列中獲取消息并進行處理。
這種模型具有高度的靈活性，可以輕松處理大量消息，同時確保消息的可靠傳遞。
2.3 特性
說到消息中間件，很多人首先想到的就是 Kafka，但 RabbitMQ 也是許多金融或互聯網公司構建可靠、可伸縮和高性能系統的首選。
這是為什么呢？
主要得從 RabbitMQ 的特性說起，主要有二：一個是功能強大，另一個是可靠性！
RabbitMQ 注重消息的可靠性和靈活性，適合任務排隊和消息傳遞。而 Kafka 是分布式流式平臺，注重日志存儲和數據分發。
順序消費也是可靠性的一種，RabbitMQ 可以使用單一隊列或多個單一隊列來確保順序消費。
除此之外，RabbitMQ 還提供持久性隊列和消息，以確保消息在 RabbitMQ 服務器宕機后不會丟失。另外，生產者可以使用發布確認機制來確認消息是否被接收。
RabbitMQ 相對 kafka 可靠性更好，數據更不易丟失，這對于一些數據敏感型的業務來說，顯然更適合用前者。
并且，RabbitMQ 中原生支持死信隊列，可以更好地處理未完成的業務消息，以及實現延時隊列等特性，接下來我們一一介紹。
3. 保證順序消費
RabbitMQ 提供了多個隊列模型來保證消息的順序消費。這對于某些應用程序非常重要，例如處理訂單、支付和庫存管理。
消息錯亂消費的場景
圖片
如上圖所示，有三條業務消息分別是刪除、增加和修改操作，但是 Consumer 沒有按順序消費，最終存儲的順序是增加、修改和刪除，就會發生數據錯亂。
針對消息有序性的問題，RabbitMQ 的解決方法是分三個階段來保證。
- 發送消息：入隊列
消息發送時，需要業務來保證順序性，就是保證生產者入隊的順序是有序的。
在分布式的場景下如果難以保證各個服務器的入隊順序，則可以加分布式鎖的方式來解決。或者在業務生產方的消息里帶上消息遞增 ID，以及消息產生的時間戳。
- 隊列中的消息
在 RabbitMQ 的消息會保存在隊列（Queue）中，在同一個隊列里的消息是先進先出（FIFO）的，這個由 RabbitMQ 來幫我們保證順序。
而不同隊列中的消息，RabbitMQ 無法保證其順序性，就像我們在食堂打飯一樣，站在不同的排隊隊列，我們也無法保證會比其他隊列的人先打上飯。
- 消費消息：出隊
一般來說，出隊后的順序消費交給消費者去保證。我們說的保證消費順序，通常也是指消費者消費消息的順序。
有多個消費者的情況下，通常是無法保證消息順序的。
這就相當于我們在排隊打飯時，有多個打飯阿姨，但是每個阿姨打飯的速度不一致，對應我們消費者的消費能力也不同。
所以，為了保證消息的順序性，我們可以只使用一個消費者來接收業務消息。
就好比只有一個阿姨在打飯，來得早就一定能早點打上飯。但很明顯，這樣效率不是很高，所以在使用時我們需要權衡利弊：看業務更需要順序性，還是更需要消費效率。
優先級隊列
在保證順序消費時，另一個迂回策略是可以使用優先級隊列（Priority Queue）。
在 RabbitMQ3.5 之后，當消費者數量較少，如果服務器檢測到消費者不能及時消費消息的情況下，優先級隊列就會生效。
具體有兩種優先級策略：
- 設置隊列的優先級
- 設置消息的優先級
在聲明隊列時，我們可以通過 x-max-priority 屬性來設置隊列的最大優先級，或通過 Priority 屬性來設置消息的優先級，從 1~10。
Golang 實現代碼如下：
// 隊列屬性
props := make(map[string]interface{})
// 設置隊列最大優先級
props["x-max-priority"] = 10
ch.Publish(
"tizi365", // 交換機
"", // 路由參數
false,
false,
amqp.Publishing{
Priority:5, // 設置消息優先級
DeliveryMode:2, // 消息投遞模式，1代表非持久化，2代表持久化，
ContentType: "text/plain",
Body: []byte(body),
})當優先級隊列消費生效時，會首先消費高優先級隊列中的優先級高的消息，以此來實現順序消費。
但需要注意的是，優先級隊列觸發的條件比較苛刻，在需要嚴格保證業務消息順序的情況下最好不要使用！
4. 死信隊列
RabbitMQ 里，當消息在隊列中變成死信（消費者無法正常處理的消息）之后，它會被重新投遞到一個交換機上（即死信交換機），死信交換機上綁定的消費隊列就是死信隊列。
圖片
死信的產生
死信產生需要滿足如下條件：
- 消息被消費者手動拒絕接收，并且 requeue（重新加入隊列）策略為 False；
- 消息已經過期（TTL）；
- 隊列達到最大長度，消息裝不下了。
死信的處理步驟
當死信產生時，如果我們定義了一個死信交換機（其實就是一個普通的交換機，只是用于處理死信，所以叫死信交換機），然后在死信交換機上綁定了一個隊列（稱作死信隊列）。
最后，如果死信隊列有消費者監聽時，死信消息的處理就會和正常業務消息一樣，從交換機到隊列，再由死信消費者（監聽死信隊列的消費者）正常消費。
5. 延時隊列
RabbitMQ 本身不支持延時隊列，但是我們可以通過 RabbitMQ 的插件 rabbitmq-delayed-message-exchange，或者使用 死信隊列 + 消息過期 的方式來實現。
5.1 應用場景
當我們在電商里購物，或者在 12306 買票時，大概都會遇到這樣一個場景：每次下訂單后，到支付訂單中間有一段商品鎖定時間，超過時間后未支付訂單就會關閉。
狀態轉換圖如下：
圖片
5.2 插件實現
- 安裝插件
Github 地址：
https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases從 github 的 release 頁面的 assets, 下載 rabbitmq_delayed_message_exchange-3.8.9-0199d11c.ez 文件，把文件放到 rabbitmq 插件目錄（plugins目錄）
提示：版本號可能跟本教程不一樣，如果你的 rabbitmq 就是最新版本，插件也選擇最新版本就行。
- 激活插件
rabbitmq-plugins enable rabbitmq_delayed_message_exchange- 定義交換機
通過 x-delayed-type 設置自定義交換機屬性，支持發送延遲消息：
props := make(map[string]interface{})
//關鍵參數，支持發送延遲消息
props["x-delayed-type"] = "direct"
// 聲明交換機
err = ch.ExchangeDeclare(
"delay.queue", // 交換機名字
"fanout", // 交換機類型
true, // 是否持久化
false,
false,
false,
props, // 設置屬性
)- 發送延遲消息
通過消息頭（x-delay），設置消息延遲時間。
msgHeaders := make(map[string]interface{})
// 通過消息頭，設置消息延遲時間，單位毫秒
msgHeaders["x-delay"] = 6000
err = ch.Publish(
"delay.queue", // 交換機名字
"", // 路由參數
false,
false,
amqp.Publishing{
Headers:msgHeaders, // 設置消息頭
ContentType: "text/plain",
Body: []byte(body),
})5.3 死信隊列 + 消息過期方案
該方案的核心思想是，先創建死信交換機、隊列和消費者，來監聽死信消息。
然后創建定時過期的消息，比如訂單支付的時間為 30min，則將消息的 TTL（最大存活時間）設置為 30min，將消息放到一個沒有消費者消費的隊列中，當消息過期后就會成為死信。
死信消息被重新發送到死信交換機，然后我們在死信隊列中消費該消息，根據商品 ID 判斷該商品是否被支付。
如果沒有支付，就取消訂單，修改訂單狀態為待下單。如果已經支付，就將商品狀態修改為已完成，并丟掉這條死信消息。
5. 小結
RabbitMQ 是一個功能強大的消息中間件，它在許多互聯網應用中扮演了關鍵角色，比如華為攝像機 SDK 的監控圖像數據上報，大部分電商系統的異步消費等等。
