背景

最近遇到一個比較特殊需求，需要修改一個的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 , 它封裝延遲交換機的設定