4K字詳解RabbitMQ延遲消息指南「.NET6+EasyNetQ」

背景

最近遇到一個比較特殊需求，需要修改一個的RabbitMQ消費者，以實作在消費某種特定的類型消息時，延遲1小時再處理，幾個需要注意的點：

• 延遲是以小時為機關
• 不是所有消息都延遲消費，隻延遲特定類型的消息
• 隻在第一次消費時延遲1小時，容錯機制産生的重新消費（也即消息消費失敗，多次進入延遲隊列重試），則不再延遲1小時
• 消費者消費過程中可能會重新開機

考慮到這幾點，我們需要一個辨別以及持久化，不能簡單使用Thread.Sleep或者Task.Delay；下面開始示範在不引入其它架構資源的前提下，利用現有的RabbitMQ來實作這個需求。

準備

如果沒有可用的RabbitMQ測試環境，推薦使用docker本地搭建

docker run -it --rm --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3.10-management           

項目搭建

建立解決方案RabbitMQDemo， 并添加一個.Net6控制台程式Producer作為生産者，

mkdir RabbitMQDemocd RabbitMQDemo dotnet new sln -n RabbitMQDemo mkdir srccd srcdotnet new console -n Producer cd Producerdotnet add package EasyNetQ  -s https://api.nuget.org/v3/index.json dotnet add package Newtonsoft.Json  -s https://api.nuget.org/v3/index.json  cd ../..dotnet sln add ./src/Producer/Producer.csproj           

我們給Producer項目添加了兩個包 ——EasyNetQ是用來簡便RabbitMQ操作，添加Newtonsoft.Json則是因為EasyNetQ從v7版本開始移除了對前者的依賴，需要使用者自行添加。

接下來定義消息的資料結構，添加一個類庫Core到解決方案，

cd srcdotnet new classlib --name Core cd ..dotnet sln add ./src/Core/Core.csproj           

添加如下OrderNotification類，後面我們根據消息的 Type的值來确定是正常消費還是延遲消費。

namespace Core{    public class OrderNotification    {        public string OrderId { get; set; }        public int Type { get; set; }        public DateTime DateCreation { get; set; }    }}           

生産者

在Producer項目裡，聲明隊列orders.notification，綁定到同名交換機，然後向該交換機發送OrderNotification類型的資料，

實際項目中，我們很少直接發消息到隊列，都是發送到交換機，這個項目雖然隻是demo，但也遵循這個原則

完整代碼如下：

using Core; using EasyNetQ;using EasyNetQ.Topology; var bus = RabbitHutch.CreateBus("host=localhost;port=5672;virtualHost=/;username=guest;password=guest;requestedHeartbeat=10");//聲明交換機var sourceExchange = await bus.Advanced.ExchangeDeclareAsync(name: "orders.notification", ExchangeType.Direct);//聲明隊列var sourceQueue = await bus.Advanced.QueueDeclareAsync(name: "orders.notification");//綁定await bus.Advanced.BindAsync(sourceExchange, sourceQueue, ""); Console.WriteLine("按Ctrl + C 暫停發送，任意鍵恢複發送");Console.TreatControlCAsInput = true; while (true){    Random random = new();    var orderId = Guid.NewGuid().ToString();    var type = random.Next(1, 3);    await bus.Advanced.PublishAsync(sourceExchange, "", true, new Message<OrderNotification>(new OrderNotification { OrderId = orderId, Type = type, DateCreation = DateTime.Now }));    Console.WriteLine(#34;{DateTime.Now}:消息（OrderId:{orderId}，Type：{type}） 已發送");    Thread.Sleep(1000);}           

運作Producer項目，可以看到消息正在不停的發送

打開RabbitMQ背景，名orders.notification的隊列和交換機已經建立好且互相綁定，隊列裡已經有我們剛剛發送的消息

下面我們要做的就是将隊列orders.notification裡Type為1的消息延遲消費，其它則正常消費。

延遲消費

使用死信交換機實作

原理就是在聲明一個隊列時，給它配置死信交換機（Dead Letter Exchanges，簡稱DLX）政策，對應參數為x-dead-letter-exchange，這種隊列處理帶設定了過期時間屬性（Properties.expiration）的消息時，在消息到期時，會自動将消息投遞到事先配置好的死信交換機上。

我們解決方案增加一個控制台類型的消費者項目DLXConsumer

cd srcdotnet new console -n DLXConsumer cd DLXConsumerdotnet add package EasyNetQ  -s https://api.nuget.org/v3/index.json dotnet add package Newtonsoft.Json  -s https://api.nuget.org/v3/index.json  cd ../..dotnet sln add ./src/DLXConsumer/DLXConsumer.csproj           

和生産者類似，實作消費者我們也建立一對同名的交換機和隊列orders.notification_dlx，用于接收轉發過來延遲消息，同時将該隊列的死信交換機設定為orders.notification；

消費消息時，為了消息是不是已經延遲過，可以在消息頭裡添加一個自定義參數biz-delayed，在将需要延遲處理的消息發送到orders.notification_dlx交換機之前，除了設定過期時間，也同時将biz-delayed設定為1，後續再消費該消息時，讀取該值，不至于陷入死循環。完整代碼如下

using Core; using EasyNetQ;using EasyNetQ.Topology; var bus = RabbitHutch.CreateBus("host=localhost;port=5672;virtualHost=/;username=guest;password=guest;requestedHeartbeat=10");var sourceExchange = await bus.Advanced.ExchangeDeclareAsync(name: "orders.notification", ExchangeType.Direct);var sourceQueue = await bus.Advanced.QueueDeclareAsync(name: "orders.notification"); var dlxExchange = await bus.Advanced.ExchangeDeclareAsync(name: "orders.notification_dlx", ExchangeType.Direct);var dlxQueue = await bus.Advanced.QueueDeclareAsync(name: "orders.notification_dlx", configure => configure.WithDeadLetterExchange(sourceExchange));await bus.Advanced.BindAsync(dlxExchange, dlxQueue, ""); bus.Advanced.Consume<OrderNotification>(sourceQueue, OrderNotificationHandler);Console.ReadLine(); async Task OrderNotificationHandler(IMessage<OrderNotification> message, MessageReceivedInfo msgInfo){    Console.WriteLine(#34;{DateTime.Now}: 開始消費 OrderId:{message.Body.OrderId} Type：{message.Body.Type}");     if (message.Body.Type == 1 && !message.Properties.Headers.ContainsKey("biz-delayed"))    {        message.Properties.Headers.Add("biz-delayed", 1);        message.Properties.Expiration = TimeSpan.FromHours(1);         await bus.Advanced.PublishAsync(dlxExchange, "", true, message);         Console.WriteLine(#34;{DateTime.Now}:OrderId:{message.Body.OrderId} Type：{message.Body.Type} 已延遲消費");    }    else    {        //假裝在消費        Thread.Sleep(1000);         Console.WriteLine(#34;{DateTime.Now}:OrderId:{message.Body.OrderId} Type：{message.Body.Type} 已成功消費");    }}           

上述代碼中，EasyNetQ設定隊列死信交換機的API為WithDeadLetterExchange，設定消息過期時間的API為Properties.Expiration。

運作DLXConsumer項目，可以看到Type為1的消息被延遲，其它則被正常消費

打開RabbitMQ背景确認，原本orders.notification裡的消息已經被消費掉了，同時多了一個orders.notification_dlx隊列，并且orders.notification_delay隊列相比orders.notification多了一個DLX标簽，Type為1的消息就是被轉移該隊列。

進入orders.notification_delay隊列，交換機與隊列正常綁定，x-dead-letter-exchange也已被設定

檢查隊列中的消息，可以看到Properties裡的expiration: 3600000 和headers: biz-delayed: 1

再過3600000毫秒，orders.notification_dlx隊列就會被投遞到orders.notification交換機，隊列orders.notification也就會收到這些資訊，這時因為消息頭裡有biz-delayed，消費者會正常将其消費。

使用延遲交換機實作

使用延遲交換機，需要RabbitMQ伺服器安裝rabbitmq_delayed_message_exchange插件，原理是投遞到延遲交換機的消息，會延遲指定時間（x-delay參數設定）後，自動投遞到該交換機綁定的另一交換機上。直接看代碼。

docker環境安裝rabbitmq_delayed_message_exchange插件

這裡介紹下docker環境如何安裝rabbitmq_delayed_message_exchange插件，首先在github https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases 下載下傳與你RabbitMQ伺服器大版本比對的Release，将檔案複制到RabbitMQ的/plugins目錄下，指令如下

docker cp {rabbitmq_delayed_message_exchange檔案路徑} {rabbitmq容器id}:/pluginsdocker exec -it {rabbitmq容器id} rabbitmq-plugins enable rabbitmq_delayed_message_exchange           

以我本機為例，插件啟用成功。

下面給解決方法添加一個DMConsumer項目。

cd srcdotnet new console -n DMConsumer cd DMConsumerdotnet add package EasyNetQ  -s https://api.nuget.org/v3/index.json dotnet add package Newtonsoft.Json  -s https://api.nuget.org/v3/index.json  cd ../..dotnet sln add ./src/DMConsumer/DMConsumer.csproj           

DMConsumer完整實作如下

using Core; using EasyNetQ;using EasyNetQ.Topology; var bus = RabbitHutch.CreateBus("host=localhost;port=5672;virtualHost=/;username=guest;password=guest;requestedHeartbeat=10");var sourceExchange = await bus.Advanced.ExchangeDeclareAsync(name: "orders.notification", ExchangeType.Direct, durable: true, autoDelete: false);var sourceQueue = await bus.Advanced.QueueDeclareAsync(name: "orders.notification"); var dmExchange = await bus.Advanced.ExchangeDeclareAsync(name: "orders.notification_dm", configure => configure.AsDelayedExchange(ExchangeType.Direct));//兩個交換機綁定await bus.Advanced.BindAsync(dmExchange, sourceExchange, ""); bus.Advanced.Consume<OrderNotification>(sourceQueue, OrderNotificationHandler);Console.ReadLine(); async Task OrderNotificationHandler(IMessage<OrderNotification> message, MessageReceivedInfo msgInfo){    Console.WriteLine(#34;{DateTime.Now}: 開始消費 OrderId:{message.Body.OrderId} Type：{message.Body.Type}");     if (message.Body.Type == 1 && !message.Properties.Headers.ContainsKey("biz-delayed"))    {        message.Properties.Headers["biz-delayed"] = 1;        message.WithDelay(TimeSpan.FromHours(1));         await bus.Advanced.PublishAsync(dmExchange, "", true, message);         Console.WriteLine(#34;{DateTime.Now}:OrderId:{message.Body.OrderId} Type：{message.Body.Type} 已延遲消費");    }    else    {        //假裝在消費        //Thread.Sleep(1000);         Console.WriteLine(#34;{DateTime.Now}:OrderId:{message.Body.OrderId} Type：{message.Body.Type} 已成功消費");    }}           

相比于DLXConsumer，DMConsumer裡，我們不需要額外的隊列，隻需要建立orders.notification_dm交換機即可，同時直接将交換機綁定到orders.notification交換機，EasyNetQ裡使用AsDelayedExchange訓示該交換機為延遲交換機，使用WithDelay設定消息延遲時間

通過檢視EasyNetQ源碼 https://github.com/EasyNetQ/EasyNetQ/blob/master/Source/EasyNetQ/DelayedExchangeExtensions.cs , 它封裝延遲交換機的設定

啟動Producer再生成一些資料，然後運作DMConsumer,效果和DLXConsumer一樣

打開RabbitMQ背景，可以看到多了一個類型為x-delayed-message的orders.notification_dm交換機，帶有DM和Args兩個标簽

進入交換機，可以看到裡面已經存儲了13條消息。

總結

自此，利用隊列的死信交換機政策和利用rabbitmq_delayed_message_exchange插件實作RabbitMQ消息延遲已經介紹完畢，下面是.NET6 demo完整的項目結構

其實除了這兩種，EasyNetQ也有一個排程器（Scheduler）可以實作延遲消息，但似乎需要依賴資料庫，不是主流的做法不推薦使用。

如有任何問題或者意見，歡迎評論。