大家好，我是小?，一個漂泊江湖多年的 985 非科班程序員，曾混跡于國企、互聯(lián)網(wǎng)大廠和創(chuàng)業(yè)公司的后臺開發(fā)攻城獅。

1. 引言

在今天的文章中，我們來聊一聊 RabbitMQ，這是小 ? 在工作中用的最早的消息中間件，主要用于大量數(shù)據(jù)的異步消費。

2. RabbitMQ

2.1 核心組件

RabbitMQ 是一個開源的消息中間件，它實現(xiàn)了高級消息隊列協(xié)議（AMQP），同時提供了各種重要組件來支持消息的生產(chǎn)、傳輸和消費。

圖片

• Producer（生產(chǎn)者）： 生產(chǎn)者是消息的發(fā)送方，負責將消息發(fā)布到 RabbitMQ 服務器。消息可以包含任何內容，例如任務、日志、通知等。
• Channel（信道）：消息推送與接收時使用的通道。
• Exchange（交換機）： 交換機是消息的中轉站，它接收來自生產(chǎn)者的消息并將其路由到一個或多個隊列。不同類型的交換機，如 fanout，direct，topic，headers，支持不同的路由規(guī)則。
• Queue（隊列）： 隊列是消息的緩沖區(qū)，消息在發(fā)送到消費者之前存儲在隊列中，消費者從隊列中獲取消息并進行處理。
• Consumer（消費者）： 消費者是消息的接收方，它從隊列中獲取消息并進行處理。消費者可以是多個，它們可以在不同的應用程序或服務器上運行。

2.2 工作流程

RabbitMQ 的工作方式是基于生產(chǎn)者、交換機和隊列之間的協(xié)作。這是一個簡單的消息傳遞過程：

• 將隊列與交換機綁定（Binding）起來，定義了消息的路由規(guī)則；
• 生產(chǎn)者將消息發(fā)布到交換機，交換機根據(jù)綁定規(guī)則將消息路由到一個或多個隊列；
• 消費者從隊列中獲取消息并進行處理。

這種模型具有高度的靈活性，可以輕松處理大量消息，同時確保消息的可靠傳遞。

2.3 特性

說到消息中間件，很多人首先想到的就是 Kafka，但 RabbitMQ 也是許多金融或互聯(lián)網(wǎng)公司構建可靠、可伸縮和高性能系統(tǒng)的首選。

這是為什么呢？

主要得從 RabbitMQ 的特性說起，主要有二：一個是功能強大，另一個是可靠性！

RabbitMQ 注重消息的可靠性和靈活性，適合任務排隊和消息傳遞。而 Kafka 是分布式流式平臺，注重日志存儲和數(shù)據(jù)分發(fā)。

順序消費也是可靠性的一種，RabbitMQ 可以使用單一隊列或多個單一隊列來確保順序消費。

除此之外，RabbitMQ 還提供持久性隊列和消息，以確保消息在 RabbitMQ 服務器宕機后不會丟失。另外，生產(chǎn)者可以使用發(fā)布確認機制來確認消息是否被接收。

RabbitMQ 相對 kafka 可靠性更好，數(shù)據(jù)更不易丟失，這對于一些數(shù)據(jù)敏感型的業(yè)務來說，顯然更適合用前者。

并且，RabbitMQ 中原生支持死信隊列，可以更好地處理未完成的業(yè)務消息，以及實現(xiàn)延時隊列等特性，接下來我們一一介紹。

3. 保證順序消費

RabbitMQ 提供了多個隊列模型來保證消息的順序消費。這對于某些應用程序非常重要，例如處理訂單、支付和庫存管理。

消息錯亂消費的場景

圖片

如上圖所示，有三條業(yè)務消息分別是刪除、增加和修改操作，但是 Consumer 沒有按順序消費，最終存儲的順序是增加、修改和刪除，就會發(fā)生數(shù)據(jù)錯亂。

針對消息有序性的問題，RabbitMQ 的解決方法是分三個階段來保證。

• 發(fā)送消息：入隊列

消息發(fā)送時，需要業(yè)務來保證順序性，就是保證生產(chǎn)者入隊的順序是有序的。

在分布式的場景下如果難以保證各個服務器的入隊順序，則可以加分布式鎖的方式來解決?；蛘咴跇I(yè)務生產(chǎn)方的消息里帶上消息遞增 ID，以及消息產(chǎn)生的時間戳。

• 隊列中的消息

在 RabbitMQ 的消息會保存在隊列（Queue）中，在同一個隊列里的消息是先進先出（FIFO）的，這個由 RabbitMQ 來幫我們保證順序。

而不同隊列中的消息，RabbitMQ 無法保證其順序性，就像我們在食堂打飯一樣，站在不同的排隊隊列，我們也無法保證會比其他隊列的人先打上飯。

• 消費消息：出隊

一般來說，出隊后的順序消費交給消費者去保證。我們說的保證消費順序，通常也是指消費者消費消息的順序。

有多個消費者的情況下，通常是無法保證消息順序的。

這就相當于我們在排隊打飯時，有多個打飯阿姨，但是每個阿姨打飯的速度不一致，對應我們消費者的消費能力也不同。

所以，為了保證消息的順序性，我們可以只使用一個消費者來接收業(yè)務消息。

就好比只有一個阿姨在打飯，來得早就一定能早點打上飯。但很明顯，這樣效率不是很高，所以在使用時我們需要權衡利弊：看業(yè)務更需要順序性，還是更需要消費效率。

優(yōu)先級隊列

在保證順序消費時，另一個迂回策略是可以使用優(yōu)先級隊列（Priority Queue）。

在 RabbitMQ3.5 之后，當消費者數(shù)量較少，如果服務器檢測到消費者不能及時消費消息的情況下，優(yōu)先級隊列就會生效。

具體有兩種優(yōu)先級策略：

• 設置隊列的優(yōu)先級
• 設置消息的優(yōu)先級

在聲明隊列時，我們可以通過 x-max-priority 屬性來設置隊列的最大優(yōu)先級，或通過 Priority 屬性來設置消息的優(yōu)先級，從 1~10。

Golang 實現(xiàn)代碼如下：

// 隊列屬性
props := make(map[string]interface{})
// 設置隊列最大優(yōu)先級
props["x-max-priority"] = 10

ch.Publish(
   "tizi365",     // 交換機
   "", // 路由參數(shù)
   false,
   false,
   amqp.Publishing{
       Priority:5, // 設置消息優(yōu)先級
       DeliveryMode:2,  // 消息投遞模式，1代表非持久化，2代表持久化，
       ContentType: "text/plain",
       Body:       []byte(body),
  })

當優(yōu)先級隊列消費生效時，會首先消費高優(yōu)先級隊列中的優(yōu)先級高的消息，以此來實現(xiàn)順序消費。

但需要注意的是，優(yōu)先級隊列觸發(fā)的條件比較苛刻，在需要嚴格保證業(yè)務消息順序的情況下最好不要使用！

4. 死信隊列

RabbitMQ 里，當消息在隊列中變成死信（消費者無法正常處理的消息）之后，它會被重新投遞到一個交換機上（即死信交換機），死信交換機上綁定的消費隊列就是死信隊列。

圖片

死信的產(chǎn)生

死信產(chǎn)生需要滿足如下條件：

• 消息被消費者手動拒絕接收，并且 requeue（重新加入隊列）策略為 False；
• 消息已經(jīng)過期（TTL）；
• 隊列達到最大長度，消息裝不下了。

死信的處理步驟

當死信產(chǎn)生時，如果我們定義了一個死信交換機（其實就是一個普通的交換機，只是用于處理死信，所以叫死信交換機），然后在死信交換機上綁定了一個隊列（稱作死信隊列）。

最后，如果死信隊列有消費者監(jiān)聽時，死信消息的處理就會和正常業(yè)務消息一樣，從交換機到隊列，再由死信消費者（監(jiān)聽死信隊列的消費者）正常消費。

5. 延時隊列

RabbitMQ 本身不支持延時隊列，但是我們可以通過 RabbitMQ 的插件 rabbitmq-delayed-message-exchange，或者使用 死信隊列 + 消息過期 的方式來實現(xiàn)。

5.1 應用場景

當我們在電商里購物，或者在 12306 買票時，大概都會遇到這樣一個場景：每次下訂單后，到支付訂單中間有一段商品鎖定時間，超過時間后未支付訂單就會關閉。

狀態(tài)轉換圖如下：

圖片

5.2 插件實現(xiàn)

• 安裝插件

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 設置自定義交換機屬性，支持發(fā)送延遲消息：

props := make(map[string]interface{})
   //關鍵參數(shù)，支持發(fā)送延遲消息
   props["x-delayed-type"] = "direct"

   // 聲明交換機
   err = ch.ExchangeDeclare(
       "delay.queue",   // 交換機名字
       "fanout", // 交換機類型
       true,     // 是否持久化
       false,    
       false,
       false,
       props,      // 設置屬性
  )

• 發(fā)送延遲消息

通過消息頭（x-delay），設置消息延遲時間。

msgHeaders := make(map[string]interface{})
       // 通過消息頭，設置消息延遲時間，單位毫秒
       msgHeaders["x-delay"] = 6000

       err = ch.Publish(
           "delay.queue",     // 交換機名字
           "", // 路由參數(shù)
           false,
           false,
           amqp.Publishing{
               Headers:msgHeaders, // 設置消息頭
               ContentType: "text/plain",
               Body:       []byte(body),
          })

5.3 死信隊列 + 消息過期方案

該方案的核心思想是，先創(chuàng)建死信交換機、隊列和消費者，來監(jiān)聽死信消息。

然后創(chuàng)建定時過期的消息，比如訂單支付的時間為 30min，則將消息的 TTL（最大存活時間）設置為 30min，將消息放到一個沒有消費者消費的隊列中，當消息過期后就會成為死信。

死信消息被重新發(fā)送到死信交換機，然后我們在死信隊列中消費該消息，根據(jù)商品 ID 判斷該商品是否被支付。

如果沒有支付，就取消訂單，修改訂單狀態(tài)為待下單。如果已經(jīng)支付，就將商品狀態(tài)修改為已完成，并丟掉這條死信消息。

5. 小結

RabbitMQ 是一個功能強大的消息中間件，它在許多互聯(lián)網(wǎng)應用中扮演了關鍵角色，比如華為攝像機 SDK 的監(jiān)控圖像數(shù)據(jù)上報，大部分電商系統(tǒng)的異步消費等等。