【.NET 6】RabbitMQ延遲消息指南
背景
最近遇到一個比較特殊需求，需要修改一個的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 RabbitMQDemo
cd RabbitMQDemo
dotnet new sln -n RabbitMQDemo
mkdir src
cd src
dotnet new console -n Producer
cd Producer
dotnet add package EasyNetQ -s //api.nuget.org/v3/index.json
dotnet add package Newtonsoft.Json -s //api.nuget.org/v3/index.json
cd ../..
dotnet sln add ./src/Producer/Producer.csproj
我們給Producer項目添加了兩個包 ——EasyNetQ是用來簡便RabbitMQ操作，添加Newtonsoft.Json則是因為EasyNetQ從v7版本開始移除了對前者的依賴，需要使用者自行添加。
接下來定義消息的數據結構，添加一個類庫Core到解決方案，
cd src
dotnet 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($"{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 src
dotnet new console -n DLXConsumer
cd DLXConsumer
dotnet add package EasyNetQ -s //api.nuget.org/v3/index.json
dotnet add package Newtonsoft.Json -s //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($"{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($"{DateTime.Now}:OrderId:{message.Body.OrderId} Type：{message.Body.Type} 已延遲消費");
}
else
{
//假裝在消費
Thread.Sleep(1000);
Console.WriteLine($"{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 //github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases 下載與你RabbitMQ服務器大版本匹配的Release，將文件複製到RabbitMQ的/plugins目錄下，命令如下
docker cp {rabbitmq_delayed_message_exchange文件路徑} {rabbitmq容器id}:/plugins
docker exec -it {rabbitmq容器id} rabbitmq-plugins enable rabbitmq_delayed_message_exchange
以我本機為例，插件啟用成功。
下面給解決方法添加一個DMConsumer項目。
cd src
dotnet new console -n DMConsumer
cd DMConsumer
dotnet add package EasyNetQ -s //api.nuget.org/v3/index.json
dotnet add package Newtonsoft.Json -s //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($"{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($"{DateTime.Now}:OrderId:{message.Body.OrderId} Type：{message.Body.Type} 已延遲消費");
}
else
{
//假裝在消費
//Thread.Sleep(1000);
Console.WriteLine($"{DateTime.Now}:OrderId:{message.Body.OrderId} Type：{message.Body.Type} 已成功消費");
}
}
相比於DLXConsumer，DMConsumer里，我們不需要額外的隊列，只需要創建orders.notification_dm交換機即可，同時直接將交換機綁定到orders.notification交換機，EasyNetQ里使用AsDelayedExchange指示該交換機為延遲交換機，使用WithDelay設置消息延遲時間
通過查看EasyNetQ源碼 //github.com/EasyNetQ/EasyNetQ/blob/master/Source/EasyNetQ/DelayedExchangeExtensions.cs , 它封裝延遲交換機的設置
啟動Producer再生成一些數據，然後運行DLXConsumer看效果，和DLXConsumer一樣
打開RabbitMQ後台，可以看到多了一個類型為x-delayed-message的orders.notification_dm交換機，帶有DM和Args兩個標籤
進入交換機，可以看到裏面已經存儲了13條消息。
總結
自此，利用隊列的死信交換機策略和利用rabbitmq_delayed_message_exchange插件實現RabbitMQ消息延遲已經介紹完畢，下面是.NET6 demo完整的項目結構
其實除了這兩種，EasyNetQ也有一個調度器（Scheduler）可以實現延遲消息，但似乎需要依賴數據庫，不是主流的做法不推薦使用。
如有任何問題或者意見，歡迎評論。
