背景

最近遇到一個比較特殊需求，需要修改一個的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 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 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 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($"{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 https://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 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($"{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原始碼 https://github.com/EasyNetQ/EasyNetQ/blob/master/Source/EasyNetQ/DelayedExchangeExtensions.cs , 它封裝延遲交換機的設定