【消息队列】RabbitMQ “消息队列模式” 以及NET8集成

在 .NET 8 中集成 RabbitMQ 消息队列，可以使用官方推荐的 RabbitMQ.Client 库或封装好的 MassTransit/EasyNetQ 等高级库。以下是 RabbitMQ 的基本集成代码和常见消息模式的实现。

RabbitMQ 本身并没有直接支持延时消息的功能，但是可以通过一些机制来实现延时消息的效果。以下是两种常用的方法：

TTL（Time To Live）+ 死信交换机（Dead Letter Exchange, DLX）

可以为队列或消息设置 TTL，当消息的 TTL 到期后，如果没有被消费，就会变成死信。
设置了死信交换机（DLX）的队列中的死信会被转发到指定的 DLX 上，然后可以由绑定到这个 DLX 的队列进行处理，这样就实现了延时消息的功能。

使用插件 rabbitmq-delayed-message-exchange

RabbitMQ 提供了一个官方插件 rabbitmq-delayed-message-exchange，它允许你创建一个特殊的交换机类型，该交换机能够接受带有延迟时间的消息，并在指定的时间后将消息投递给相应的队列。
这个插件需要安装并启用，并且要求 Erlang/OPT 版本在 18.0 及以上。

一、. RabbitMQ 基础集成（.NET 8）

安装 NuGet 包

dotnet add package RabbitMQ.Client

配置 RabbitMQ 连接

csharp
using RabbitMQ.Client;public class RabbitMQService
{private readonly IConnection _connection;private readonly IModel _channel;public RabbitMQService(string hostname = "localhost", string username = "guest", string password = "guest"){var factory = new ConnectionFactory{HostName = hostname,UserName = username,Password = password,DispatchConsumersAsync = true // 启用异步消费};_connection = factory.CreateConnection();_channel = _connection.CreateModel();}public void Dispose(){_channel?.Close();_connection?.Close();}
}

二. RabbitMQ 常见消息模式

(1)、简单队列（Simple Queue）

场景：生产者发送消息到队列，消费者从队列接收消息（一对一）。

生产者（Producer）

csharp
public void SendMessage(string queueName, string message)
{_channel.QueueDeclare(queue: queueName,durable: true, // 持久化队列exclusive: false,autoDelete: false);var body = Encoding.UTF8.GetBytes(message);_channel.BasicPublish(exchange: "",routingKey: queueName,basicProperties: null,body: body);
}

消费者（Consumer）

csharp
public void ReceiveMessages(string queueName)
{_channel.QueueDeclare(queueName, durable: true, exclusive: false, autoDelete: false);var consumer = new AsyncEventingBasicConsumer(_channel);consumer.Received += async (model, ea) =>{var body = Encoding.UTF8.GetString(ea.Body.Span);Console.WriteLine($"Received: {body}");await Task.Yield(); // 模拟异步处理_channel.BasicAck(ea.DeliveryTag, false); // 手动ACK};_channel.BasicConsume(queue: queueName,autoAck: false, // 关闭自动ACKconsumer: consumer);
}

(2)、工作队列（Work Queue）

场景：多个消费者竞争消费同一个队列的消息（任务分发）。

生产者
同简单队列的 SendMessage 方法。
消费者

csharp
// 启动多个消费者实例，RabbitMQ 会轮询分发消息
for (int i = 0; i < 3; i++) // 3个消费者
{Task.Run(() =>{using var service = new RabbitMQService();service.ReceiveMessages("task_queue");Console.WriteLine($"Consumer {i} started...");Thread.Sleep(Timeout.Infinite);});
}

(3)、发布/订阅（Pub/Sub）【Fannout】

场景：一个生产者发送消息到交换机（Exchange），多个队列绑定到交换机，每个队列有自己的消费者（广播模式）。

生产者（发送到交换机）

csharp
public void PublishToExchange(string exchangeName, string message)
{_channel.ExchangeDeclare(exchangeName, ExchangeType.Fanout); // Fanout 广播var body = Encoding.UTF8.GetBytes(message);_channel.BasicPublish(exchangeName, "", null, body);
}

消费者（绑定队列到交换机）

csharp
public void SubscribeToExchange(string exchangeName, string queueName)
{_channel.ExchangeDeclare(exchangeName, ExchangeType.Fanout);_channel.QueueDeclare(queueName, durable: true, exclusive: false, autoDelete: false);_channel.QueueBind(queueName, exchangeName, "");var consumer = new AsyncEventingBasicConsumer(_channel);consumer.Received += (model, ea) =>{var body = Encoding.UTF8.GetString(ea.Body.Span);Console.WriteLine($"Received: {body}");return Task.CompletedTask;};_channel.BasicConsume(queueName, autoAck: true, consumer);
}

(4) 、路由模式（Routing）

场景：根据 RoutingKey 定向投递消息到特定队列。

生产者

csharp
public void SendWithRouting(string exchangeName, string routingKey, string message)
{_channel.ExchangeDeclare(exchangeName, ExchangeType.Direct);var body = Encoding.UTF8.GetBytes(message);_channel.BasicPublish(exchangeName, routingKey, null, body);
}

消费者

csharp
public void ReceiveWithRouting(string exchangeName, string queueName, string routingKey)
{_channel.ExchangeDeclare(exchangeName, ExchangeType.Direct);_channel.QueueDeclare(queueName, durable: true, exclusive: false, autoDelete: false);_channel.QueueBind(queueName, exchangeName, routingKey);var consumer = new AsyncEventingBasicConsumer(_channel);consumer.Received += (model, ea) =>{Console.WriteLine($"Received {ea.RoutingKey}: {Encoding.UTF8.GetString(ea.Body.Span)}");return Task.CompletedTask;};_channel.BasicConsume(queueName, autoAck: true, consumer);
}

(5) 、主题模式（Topic）

场景：使用通配符（*、#）匹配 RoutingKey，实现灵活路由。
*（星号）匹配单个单词
#（井号）匹配多个单词

生产者

csharp
public void SendWithTopic(string exchangeName, string topic, string message)
{_channel.ExchangeDeclare(exchangeName, ExchangeType.Topic);var body = Encoding.UTF8.GetBytes(message);_channel.BasicPublish(exchangeName, topic, null, body);
}

消费者

csharp
public void ReceiveWithTopic(string exchangeName, string queueName, string topicPattern)
{_channel.ExchangeDeclare(exchangeName, ExchangeType.Topic);_channel.QueueDeclare(queueName, durable: true, exclusive: false, autoDelete: false);_channel.QueueBind(queueName, exchangeName, topicPattern);var consumer = new AsyncEventingBasicConsumer(_channel);consumer.Received += (model, ea) =>{Console.WriteLine($"Received {ea.RoutingKey}: {Encoding.UTF8.GetString(ea.Body.Span)}");return Task.CompletedTask;};_channel.BasicConsume(queueName, autoAck: true, consumer);
}

(6)、 RPC（远程过程调用）

场景：客户端发送请求并等待服务端响应（同步通信）。

客户端（RPC Client）

public string Call(string message, string queueName = "rpc_queue")
{var correlationId = Guid.NewGuid().ToString();var replyQueueName = _channel.QueueDeclare().QueueName;var properties = _channel.CreateBasicProperties();properties.ReplyTo = replyQueueName;properties.CorrelationId = correlationId;var body = Encoding.UTF8.GetBytes(message);_channel.BasicPublish("", queueName, properties, body);var tcs = new TaskCompletionSource<string>();var consumer = new EventingBasicConsumer(_channel);consumer.Received += (model, ea) =>{if (ea.BasicProperties.CorrelationId == correlationId){var response = Encoding.UTF8.GetString(ea.Body.Span);tcs.SetResult(response);}};_channel.BasicConsume(replyQueueName, autoAck: true, consumer);return tcs.Task.Result; // 同步等待响应（生产环境建议用异步）
}

服务端（RPC Server）

csharp
public void StartRpcServer(string queueName = "rpc_queue")
{_channel.QueueDeclare(queueName, durable: false, exclusive: false, autoDelete: false);var consumer = new AsyncEventingBasicConsumer(_channel);consumer.Received += async (model, ea) =>{var message = Encoding.UTF8.GetString(ea.Body.Span);Console.WriteLine($"Received RPC call: {message}");var response = $"Response to: {message}";var responseBytes = Encoding.UTF8.GetBytes(response);var properties = ea.BasicProperties;var replyProps = _channel.CreateBasicProperties();replyProps.CorrelationId = properties.CorrelationId;_channel.BasicPublish("",properties.ReplyTo,replyProps,responseBytes);_channel.BasicAck(ea.DeliveryTag, false);await Task.CompletedTask;};_channel.BasicConsume(queueName, autoAck: false, consumer);
}

三. 完整示例（.NET 8 Worker Service）

生产者项目

// Program.cs
var builder = WebApplication.CreateBuilder(args);
var app = builder.Build();var rabbitMQ = new RabbitMQService();
rabbitMQ.SendMessage("hello_queue", "Hello, RabbitMQ!");app.Run();

消费者项目（Worker Service）

csharp
// Program.cs
IHost host = Host.CreateDefaultBuilder(args).ConfigureServices(services =>{services.AddHostedService<Worker>();}).Build();await host.RunAsync();// Worker.cs
public class Worker : BackgroundService
{private readonly RabbitMQService _rabbitMQ;public Worker(){_rabbitMQ = new RabbitMQService();}protected override async Task ExecuteAsync(CancellationToken stoppingToken){_rabbitMQ.ReceiveMessages("hello_queue");while (!stoppingToken.IsCancellationRequested){await Task.Delay(1000, stoppingToken);}}
}

四. 总结

推荐实践
连接管理：使用 IHostedService 或单例模式管理 IConnection 和 IModel。
异常处理：监听 Connection.ConnectionShutdown 事件并重连。
性能优化：启用 DispatchConsumersAsync = true 支持异步消费。
高级封装：考虑使用 MassTransit 或 EasyNetQ 简化开发。
通过以上模式，可以灵活应对异步任务处理、事件驱动架构、微服务通信等场景。