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 的監控圖像數據上報,大部分電商系統的異步消費等等。