.NET 下 RabbitMQ 隊列、死信隊列、延時隊列及小應用
作者：conan
RabbitMQ 提供了強大的消息隊列功能，在 .NET 應用程序中，我們可以利用其隊列、死信隊列、延時隊列等特性，實現異步通信、任務調度、日志記錄。?
引言
RabbitMQ 是一款廣泛使用的開源消息代理軟件，它基于 AMQP 協議，提供了可靠、靈活的消息傳遞服務。在 .NET 應用程序中，我們可以利用 RabbitMQ 來實現異步通信、解耦服務、平衡負載等功能。本文將詳細介紹如何在 .NET 中使用 RabbitMQ 的隊列、死信隊列、延時隊列，以及一些實際應用場景。
RabbitMQ 隊列基礎
安裝 RabbitMQ.Client
在 .NET 項目中使用 RabbitMQ，首先需要安裝 RabbitMQ.Client 庫。可以通過 NuGet 包管理器來安裝：
- 使用包管理器控制臺：
Install-Package RabbitMQ.Client- 使用 .NET CLI：
dotnet add package RabbitMQ.Client創建生產者和消費者
生產者
生產者負責發送消息到 RabbitMQ 服務器。以下是一個簡單的生產者示例：
using RabbitMQ.Client;
using System.Text;
class Producer
{
public static void SendMessage(string message)
{
var factory = new ConnectionFactory() { HostName = "localhost" };
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
channel.QueueDeclare(queue: "hello", durable: false, exclusive: false, autoDelete: false, arguments: null);
var body = Encoding.UTF8.GetBytes(message);
channel.BasicPublish(exchange: "", routingKey: "hello", basicProperties: null, body: body);
Console.WriteLine(" [x] Sent {0}", message);
}
}
}消費者
消費者負責從 RabbitMQ 服務器接收消息。以下是一個簡單的消費者示例：
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Text;
class Consumer
{
public static void ReceiveMessage()
{
var factory = new ConnectionFactory() { HostName = "localhost" };
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
channel.QueueDeclare(queue: "hello", durable: false, exclusive: false, autoDelete: false, arguments: null);
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
var body = ea.Body.ToArray();
var message = Encoding.UTF8.GetString(body);
Console.WriteLine(" [x] Received {0}", message);
};
channel.BasicConsume(queue: "hello", autoAck: true, consumer: consumer);
Console.WriteLine(" Press [enter] to exit.");
Console.ReadLine();
}
}
}死信隊列
死信隊列（Dead Letter Queue，簡稱 DLQ）用于存儲和處理那些因為某些原因無法被正常消費的消息。以下是幾種常見的死信隊列形成場景：
- 消息 TTL（Time To Live）過期
- 隊列達到最大長度
- 消息被拒絕（basic.reject 或 basic.nack）并且 requeue=false
實現死信隊列
以下是一個使用死信隊列的示例：
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Text;
class DeadLetterQueueExample
{
public static void Setup()
{
var factory = new ConnectionFactory() { HostName = "localhost" };
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
// 聲明死信交換機和死信隊列
channel.ExchangeDeclare("dead_letter_exchange", ExchangeType.Direct);
channel.QueueDeclare("dead_letter_queue", durable: false, exclusive: false, autoDelete: false, arguments: null);
channel.QueueBind("dead_letter_queue", "dead_letter_exchange", "dead_letter_routing_key");
// 聲明普通隊列，并設置死信交換機和死信路由鍵
var args = new Dictionary<string, object>
{
{ "x-dead-letter-exchange", "dead_letter_exchange" },
{ "x-dead-letter-routing-key", "dead_letter_routing_key" }
};
channel.QueueDeclare("normal_queue", durable: false, exclusive: false, autoDelete: false, arguments: args);
// 發送消息到普通隊列
var body = Encoding.UTF8.GetBytes("This message will be dead lettered.");
channel.BasicPublish("", "normal_queue", null, body);
}
}
public static void ConsumeDeadLetter()
{
var factory = new ConnectionFactory() { HostName = "localhost" };
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
var body = ea.Body.ToArray();
var message = Encoding.UTF8.GetString(body);
Console.WriteLine($"Received dead letter message: {message}");
};
channel.BasicConsume("dead_letter_queue", autoAck: true, consumer: consumer);
Console.WriteLine(" Press [enter] to exit.");
Console.ReadLine();
}
}
}延時隊列
RabbitMQ 本身沒有直接支持延時隊列的功能，但可以通過 TTL（Time To Live）+ 死信隊列的組合來實現。以下是實現延時隊列的步驟：
- 創建一個普通隊列，并設置其死信交換機和死信路由鍵。
- 將需要延遲處理的消息發送到這個隊列，并設置消息的過期時間（TTL）。
- 當消息過期后，RabbitMQ 會將其發送到死信隊列，而死信隊列可以由消費者按照正常的方式進行處理。
實現延時隊列
以下是一個使用延時隊列的示例：
using RabbitMQ.Client;
using System.Text;
class DelayQueueExample
{
public static void SendMessage(string message, int delayMilliseconds)
{
var factory = new ConnectionFactory() { HostName = "localhost" };
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
// 聲明死信交換機和死信隊列
channel.ExchangeDeclare("delay_exchange", ExchangeType.Direct);
channel.QueueDeclare("delay_queue", durable: false, exclusive: false, autoDelete: false, arguments: null);
channel.QueueBind("delay_queue", "delay_exchange", "delay_routing_key");
// 聲明延時隊列，并設置死信交換機和死信路由鍵
var args = new Dictionary<string, object>
{
{ "x-dead-letter-exchange", "delay_exchange" },
{ "x-dead-letter-routing-key", "delay_routing_key" }
};
channel.QueueDeclare("normal_queue", durable: false, exclusive: false, autoDelete: false, arguments: args);
// 發送消息到普通隊列，并設置 TTL
var properties = channel.CreateBasicProperties();
properties.Expiration = delayMilliseconds.ToString();
var body = Encoding.UTF8.GetBytes(message);
channel.BasicPublish("", "normal_queue", properties, body);
}
}
}小應用示例
訂單超時自動取消
假設我們有一個在線商城，用戶下單后需要在指定時間內完成支付，否則訂單將自動取消。我們可以使用延時隊列來實現這一功能：
- 用戶下單時，將訂單信息發送到延時隊列，并設置 TTL 為指定的超時時間。
- 如果用戶在超時時間內完成支付，可以從延時隊列中移除該訂單的消息。
- 如果用戶未在超時時間內完成支付，訂單消息將被發送到死信隊列。
- 一個專門的消費者監聽死信隊列，當收到訂單消息時，自動取消該訂單，并進行相應的后續處理。
日志記錄
在分布式系統中，日志記錄是一個重要的功能。我們可以使用 RabbitMQ 的隊列來實現日志的異步記錄：
- 各個服務在生成日志時，將日志信息發送到一個日志隊列。
- 一個專門的日志服務監聽日志隊列，當收到日志消息時，將其存儲到日志數據庫或文件系統中。
任務調度
RabbitMQ 可以用于實現任務調度系統：
- 將需要執行的任務發送到任務隊列，每個任務可以包含任務的詳細信息和執行時間。
- 任務消費者從任務隊列中獲取任務，并根據任務的執行時間將其放入延時隊列。
- 當任務的執行時間到達時，任務消息從延時隊列中釋放，并被任務消費者獲取。
- 任務消費者執行任務，并將任務的執行結果發送到結果隊列。
結論
RabbitMQ 提供了強大的消息隊列功能，在 .NET 應用程序中，我們可以利用其隊列、死信隊列、延時隊列等特性，實現異步通信、任務調度、日志記錄。
責任編輯：武曉燕
來源：
程序員編程日記
