Smart.RabbitMQ 2.0.0

Smart.RabbitMQ

English | 中文

<a name="english"></a>

English

Smart.RabbitMQ is a high-level abstraction library built on top of RabbitMQ.Client, providing a simple, thread-safe, and feature-rich API for RabbitMQ message queue operations. It supports .NET 8, 9, and 10.

Features

• Simple API: Intuitive and easy-to-use interface for producing and consuming messages.
• Factory Pattern: IRabbitMQFactory for creating producer and consumer instances.
• Connection Sharing: All producers and consumers created by the same factory share a single connection, reducing resource overhead.
• Thread-Safe Internals: Connection and channel creation are guarded by a semaphore, and binding de-duplication is concurrency-safe. A single producer/consumer still owns one channel, so it should not be shared by concurrent publishers (RabbitMQ requires one channel per thread).
• Lazy Initialization: Channels are initialized on first use to avoid blocking constructors.
• Configurable Recovery: AutomaticRecoveryEnabled, NetworkRecoveryIntervalSeconds and RequestedHeartbeatSeconds are exposed in RabbitMqOptions.
• Reliable Publishing: Publisher confirms are enabled by default, messages are published as Persistent with content_type / content_encoding / message_id / timestamp, and unroutable messages can be observed via the mandatory flag plus the BasicReturn event.
• Safe Failure Handling: Deserialization failures are rejected without requeue (no poison-message loop), handler failures follow the requeueOnFailure policy, and both are reported through MessageHandleFailed.
• Multiple Exchange Types: Support for Direct, Fanout, Topic, and Headers exchanges (Headers requires binding arguments, now supported by QueueBindAsync).
• Message Acknowledgment: Flexible manual and automatic acknowledgment with error handling.
• Dependency Injection: Seamless integration with Microsoft.Extensions.DependencyInjection.
• JSON Serialization: Built-in JSON serialization with camelCase naming policy (case-insensitive on read).
• Queue Binding Management: Automatic prevention of duplicate bindings.
• Configuration Validation: Built-in validation for RabbitMQ connection settings.
• Health Check: IRabbitMQCheck.CheckAsync probes the broker with a short-lived dedicated connection.

Installation

Install the package via NuGet:

dotnet add package Smart.RabbitMQ

Core Components

IRabbitMQFactory

Factory interface for creating producer and consumer instances. All instances share the same connection.

public interface IRabbitMQFactory : IRabbitMQCheck, IAsyncDisposable
{
    Task<SmartRabbitMQProducer> CreateProducerAsync();
    Task<SmartRabbitMQConsumer<T>> CreateConsumerAsync<T>() where T : class;
    Task<SmartRabbitMQConsumer> CreateConsumerAsync();
}

// Inherited from IRabbitMQCheck
Task<IConnection> CreateConnectionAsync();                 // standalone connection, caller must dispose
Task<bool> CheckAsync(CancellationToken token = default);  // transient probe, shared connection untouched

SmartRabbitMQProducer

Message producer with support for publishing messages to exchanges.

public class SmartRabbitMQProducer : IAsyncDisposable
{
    public event AsyncEventHandler<BasicReturnEventArgs> BasicReturn;

    public Task ExchangeDeclareAsync(string exchange, ExchangeType type = ExchangeType.Direct, bool durable = true, CancellationToken token = default);
    public Task PublishAsync<T>(string exchange, string routingKey, T message, bool mandatory = false, CancellationToken token = default);
    public Task PublishAsync(string exchange, string routingKey, string message, bool mandatory = false, CancellationToken token = default);
}

SmartRabbitMQConsumer<T>

Generic message consumer with event-driven message processing.

public class SmartRabbitMQConsumer<T> : IAsyncDisposable where T : class
{
    public event AsyncMessageReceivedHandler<T> MessageReceived;
    public event EventHandler<MessageErrorEventArgs<T>> MessageHandleFailed;

    public Task ExchangeDeclareAsync(string exchange, ExchangeType type = ExchangeType.Direct, bool durable = true, CancellationToken token = default);
    public Task<string> QueueDeclareAsync(string queue = "", bool durable = true, IDictionary<string, object?>? queueParams = null, bool exclusive = false, bool autoDelete = false, CancellationToken token = default);
    public Task QueueBindAsync(string queue, string exchange, string routingKey, IDictionary<string, object?>? arguments = null, CancellationToken token = default);
    public Task BasicQosAsync(ushort prefetchCount = 0, CancellationToken token = default);
    public Task<string> StartConsumeAsync(bool autoAck = false, bool multiple = false, bool requeueOnFailure = true, CancellationToken token = default);
    public Task CancelConsumeAsync(string consumerTag);
}

Configuration

RabbitMqOptions

Configuration options for RabbitMQ connection.

public class RabbitMqOptions
{
    public string? ClientName { get; set; }
    public string HostName { get; set; } = "localhost";
    public int Port { get; set; } = 5672;
    public string UserName { get; set; } = "guest";
    public string Password { get; set; } = "guest";
    public string VirtualHost { get; set; } = "/";

    public bool AutomaticRecoveryEnabled { get; set; } = true;   // connection auto-recovery
    public int NetworkRecoveryIntervalSeconds { get; set; } = 10; // recovery retry interval
    public int RequestedHeartbeatSeconds { get; set; } = 60;      // 0 disables heartbeat
    public bool EnablePublisherConfirms { get; set; } = true;     // publish waits for broker ack
}

Usage Examples

1. Basic Producer

using Smart.RabbitMQ;

var options = new RabbitMqOptions
{
    HostName = "localhost",
    Port = 5672,
    UserName = "guest",
    Password = "guest",
    VirtualHost = "/"
};

var factory = new SmartRabbitMQFactory(options);
var producer = await factory.CreateProducerAsync();

// Declare exchange
await producer.ExchangeDeclareAsync("my.exchange", ExchangeType.Direct);

// Publish message (persistent, publisher confirms enabled by default)
var message = new { Id = 1, Content = "Hello RabbitMQ" };
await producer.PublishAsync("my.exchange", "my.routing.key", message);

// Optional: observe unroutable messages instead of letting the broker drop them silently
producer.BasicReturn += (sender, args) =>
{
    Console.Error.WriteLine($"Message returned: {args.ReplyText}");
    return Task.CompletedTask;
};

await producer.PublishAsync("my.exchange", "my.routing.key", message, mandatory: true);

// Dispose producer (connection is managed by factory)
await producer.DisposeAsync();

// Dispose factory when done (closes shared connection)
await factory.DisposeAsync();

2. Basic Consumer

using Smart.RabbitMQ;

var options = new RabbitMqOptions
{
    HostName = "localhost",
    Port = 5672,
    UserName = "guest",
    Password = "guest",
    VirtualHost = "/"
};

var factory = new SmartRabbitMQFactory(options);
var consumer = await factory.CreateConsumerAsync<MyMessage>();

// Subscribe to message event
consumer.MessageReceived += async (sender, args) =>
{
    Console.WriteLine($"Received message: {args.Data?.Content}");
    Console.WriteLine($"Delivery Tag: {args.DeliveryTag}");
    Console.WriteLine($"Exchange: {args.Exchange}");
    Console.WriteLine($"Routing Key: {args.RoutingKey}");
};

// Declare exchange
await consumer.ExchangeDeclareAsync("my.exchange", ExchangeType.Direct);

// Declare queue
var queueName = await consumer.QueueDeclareAsync("my.queue", durable: true);

// Bind queue to exchange
await consumer.QueueBindAsync(queueName, "my.exchange", "my.routing.key");

// Set QoS (prefetch count)
await consumer.BasicQosAsync(prefetchCount: 10);

// Start consuming
var consumerTag = await consumer.StartConsumeAsync(autoAck: false);

// Cancel consuming when done
await consumer.CancelConsumeAsync(consumerTag);

// Dispose consumer (connection is managed by factory)
await consumer.DisposeAsync();

// Dispose factory when done (closes shared connection)
await factory.DisposeAsync();

3. Dependency Injection Integration

using Microsoft.Extensions.DependencyInjection;
using Smart.RabbitMQ;

// Configure services
var services = new ServiceCollection();

var rabbitMQOptions = new RabbitMqOptions
{
    HostName = "localhost",
    Port = 5672,
    UserName = "guest",
    Password = "guest",
    VirtualHost = "/"
};

services.AddSmartRabbitMQ(rabbitMQOptions);

// Build service provider
var serviceProvider = services.BuildServiceProvider();

// Get factory
var factory = serviceProvider.GetRequiredService<IRabbitMQFactory>();

// Create producer
var producer = await factory.CreateProducerAsync();

// Create consumer
var consumer = await factory.CreateConsumerAsync<MyMessage>();

// Or get the pre-created singletons.
// Note: only the string consumer is registered; generic consumers are created through the factory.
var producer2 = serviceProvider.GetRequiredService<SmartRabbitMQProducer>();
var stringConsumer = serviceProvider.GetRequiredService<SmartRabbitMQConsumer>();

// Health check (short-lived probe, does not touch the shared connection)
var check = serviceProvider.GetRequiredService<IRabbitMQCheck>();
bool healthy = await check.CheckAsync();

4. String Message Consumer

using Smart.RabbitMQ;

var factory = new SmartRabbitMQFactory(options);
var consumer = await factory.CreateConsumerAsync(); // String consumer

consumer.MessageReceived += async (sender, args) =>
{
    Console.WriteLine($"Received string message: {args.Data}");
};

await consumer.ExchangeDeclareAsync("my.exchange");
var queueName = await consumer.QueueDeclareAsync("my.queue");
await consumer.QueueBindAsync(queueName, "my.exchange", "my.routing.key");
await consumer.StartConsumeAsync();

5. Advanced Consumer with Error Handling

using Smart.RabbitMQ;

var consumer = await factory.CreateConsumerAsync<MyMessage>();

// Acknowledge / reject is handled by the library:
//   · handler completes  -> BasicAck (BasicAck with multiple when multiple: true)
//   · handler throws     -> BasicNack(requeue: requeueOnFailure)
//   · body is not JSON   -> BasicNack(requeue: false), so a poison message cannot loop forever
consumer.MessageReceived += async (sender, args) =>
{
    await ProcessMessageAsync(args.Data);
};

// Observe failures without interfering with the acknowledgement flow
consumer.MessageHandleFailed += (sender, e) =>
{
    Console.Error.WriteLine(
        $"Error processing message (deliveryTag={e.ReceivedArgs.DeliveryTag}, retriable={e.IsRetriable}): {e.Error.Message}");
};

// In production, prefer requeueOnFailure: false and configure a dead-letter exchange
await consumer.StartConsumeAsync(autoAck: false, requeueOnFailure: false);

6. Queue Parameters

using Smart.RabbitMQ;

var queueParams = new Dictionary<string, object?>
{
    ["x-message-ttl"] = 60000,           // Message TTL: 60 seconds
    ["x-max-length"] = 10000,             // Max queue length
    ["x-expires"] = 3600000,               // Queue expires after 1 hour
    ["x-dead-letter-exchange"] = "dlx"    // Dead letter exchange
};

var queueName = await consumer.QueueDeclareAsync(
    queue: "my.queue",
    durable: true,
    queueParams: queueParams
);

7. Exchange Types

using Smart.RabbitMQ;

// Direct Exchange (Point-to-Point)
await producer.ExchangeDeclareAsync("direct.exchange", ExchangeType.Direct);

// Fanout Exchange (Broadcast)
await producer.ExchangeDeclareAsync("fanout.exchange", ExchangeType.Fanout);

// Topic Exchange (Pattern Matching)
await producer.ExchangeDeclareAsync("topic.exchange", ExchangeType.Topic);

// Headers Exchange (Header-based Routing)
await producer.ExchangeDeclareAsync("headers.exchange", ExchangeType.Headers);

// A headers binding REQUIRES arguments, otherwise nothing is ever routed
await consumer.QueueBindAsync(
    queue: "my.queue",
    exchange: "headers.exchange",
    routingKey: string.Empty,
    arguments: new Dictionary<string, object?>
    {
        ["x-match"] = "all",
        ["type"] = "order"
    });

Best Practices

1. Use Manual Acknowledgment: Set autoAck: false for better control over message processing and error handling.

2. Set Prefetch Count: Use BasicQosAsync to limit the number of unacknowledged messages and prevent consumer overload.

await consumer.BasicQosAsync(prefetchCount: 10);

3. Dispose Resources: Dispose producer and consumer instances when done. The factory manages the shared connection.

await using var producer = await factory.CreateProducerAsync();
await using var consumer = await factory.CreateConsumerAsync<MyMessage>();

// When completely done, dispose the factory to close the connection
await factory.DisposeAsync();

4. Handle Exceptions: Do not rethrow inside MessageReceived — the library already nacks for you and the rethrown exception would escape into the client's dispatch loop. Subscribe MessageHandleFailed for logging instead.

5. Use Durable Queues and Persistent Messages: durable: true only keeps the queue definition; messages are persistent because the library publishes them with delivery_mode=2.

6. Configure a Dead-Letter Exchange: Add x-dead-letter-exchange to queueParams and use requeueOnFailure: false in production, so failing messages land in the DLX instead of being redelivered forever.

7. Validate Configuration: AddSmartRabbitMQ and the SmartRabbitMQFactory constructor both call RabbitMqOptions.Validate(); invalid settings throw ArgumentException immediately.

8. One Channel per Concurrent Publisher: A producer owns a single channel. For highly concurrent publishing, create one producer per concurrent unit via CreateProducerAsync() rather than sharing a single instance across threads.

Event Arguments

MessageReceivedArgs<T>

Arguments passed to the message received event.

public class MessageReceivedArgs<T> : EventArgs
{
    public string? ConsumerTag { get; init; }
    public ulong DeliveryTag { get; init; }
    public string? Exchange { get; init; }
    public string? RoutingKey { get; init; }
    public bool Redelivered { get; init; }
    public T? Data { get; }
}

// Payload of MessageHandleFailed
public class MessageErrorEventArgs<T> : EventArgs
{
    public Exception Error { get; }
    public MessageReceivedArgs<T> ReceivedArgs { get; }
    public bool IsRetriable { get; }   // false => rejected without requeue (DLX candidate)
}

Project Structure

Smart.RabbitMQ/
├── SmartRabbitMQFactory.cs           - Factory for creating producers and consumers
├── SmartRabbitMQProducer.cs          - Message producer implementation
├── SmartRabbitMQConsumer.cs          - Generic and string message consumer implementation
├── SerializeConvert.cs               - JSON serialization utilities
├── RabbitMqOptions.cs                - Configuration options
├── MessageReceivedArgs.cs            - Event argument classes (MessageReceivedArgs / MessageErrorEventArgs)
├── ExchangeType.cs                   - Exchange type enumeration
├── ExchangeTypeExtensions.cs         - AMQP exchange type name mapping
├── IRabbitMQFactory.cs               - Factory interface
├── IRabbitMQCheck.cs                 - Health check contract
└── RabbitMQServiceExtensions.cs      - Dependency injection extensions

Supported .NET Versions

• .NET 8
• .NET 9
• .NET 10

<a name="chinese"></a>

中文

Smart.RabbitMQ 是一个基于 RabbitMQ.Client 构建的高级抽象库，为 RabbitMQ 消息队列操作提供了简单、线程安全且功能丰富的 API。它支持 .NET 8, 9 和 10。

功能特性

• 简单易用的 API: 直观且易于使用的生产者和消费消息接口。
• 工厂模式: IRabbitMQFactory 用于创建生产者和消费者实例。
• 连接共享: 同一工厂创建的所有生产者和消费者共享同一个连接，减少资源开销。
• 线程安全: 所有操作都设计为线程安全的并发访问。
• 延迟初始化: 通道在首次使用时初始化，避免阻塞构造函数。
• 自动重连: 内置连接恢复功能，支持可配置的重试间隔。
• 多种交换机类型: 支持 Direct、Fanout、Topic 和 Headers 交换机。
• 消息确认: 灵活的手动和自动确认机制，带有错误处理。
• 依赖注入: 与 Microsoft.Extensions.DependencyInjection 无缝集成。
• JSON 序列化: 内置 JSON 序列化，使用 camelCase 命名策略。
• 队列绑定管理: 自动防止重复绑定。
• 配置验证: 内置 RabbitMQ 连接设置的验证功能。

安装

通过 NuGet 安装：

dotnet add package Smart.RabbitMQ

核心组件

IRabbitMQFactory

用于创建生产者和消费者实例的工厂接口。所有实例共享同一个连接。

public interface IRabbitMQFactory : IRabbitMQCheck, IAsyncDisposable
{
    Task<SmartRabbitMQProducer> CreateProducerAsync();
    Task<SmartRabbitMQConsumer<T>> CreateConsumerAsync<T>() where T : class;
    Task<SmartRabbitMQConsumer> CreateConsumerAsync();
}

// 继承自 IRabbitMQCheck
Task<IConnection> CreateConnectionAsync();                 // 独立连接，调用方必须自行释放
Task<bool> CheckAsync(CancellationToken token = default);  // 临时探测，不影响共享连接

SmartRabbitMQProducer

消息生产者，支持向交换机发布消息。

public class SmartRabbitMQProducer : IAsyncDisposable
{
    public event AsyncEventHandler<BasicReturnEventArgs> BasicReturn;

    public Task ExchangeDeclareAsync(string exchange, ExchangeType type = ExchangeType.Direct, bool durable = true, CancellationToken token = default);
    public Task PublishAsync<T>(string exchange, string routingKey, T message, bool mandatory = false, CancellationToken token = default);
    public Task PublishAsync(string exchange, string routingKey, string message, bool mandatory = false, CancellationToken token = default);
}

SmartRabbitMQConsumer<T>

泛型消息消费者，支持事件驱动的消息处理。

public class SmartRabbitMQConsumer<T> : IAsyncDisposable where T : class
{
    public event AsyncMessageReceivedHandler<T> MessageReceived;
    public event EventHandler<MessageErrorEventArgs<T>> MessageHandleFailed;

    public Task ExchangeDeclareAsync(string exchange, ExchangeType type = ExchangeType.Direct, bool durable = true, CancellationToken token = default);
    public Task<string> QueueDeclareAsync(string queue = "", bool durable = true, IDictionary<string, object?>? queueParams = null, bool exclusive = false, bool autoDelete = false, CancellationToken token = default);
    public Task QueueBindAsync(string queue, string exchange, string routingKey, IDictionary<string, object?>? arguments = null, CancellationToken token = default);
    public Task BasicQosAsync(ushort prefetchCount = 0, CancellationToken token = default);
    public Task<string> StartConsumeAsync(bool autoAck = false, bool multiple = false, bool requeueOnFailure = true, CancellationToken token = default);
    public Task CancelConsumeAsync(string consumerTag);
}

配置

RabbitMqOptions

RabbitMQ 连接配置选项。

public class RabbitMqOptions
{
    public string? ClientName { get; set; }
    public string HostName { get; set; } = "localhost";
    public int Port { get; set; } = 5672;
    public string UserName { get; set; } = "guest";
    public string Password { get; set; } = "guest";
    public string VirtualHost { get; set; } = "/";

    public bool AutomaticRecoveryEnabled { get; set; } = true;   // connection auto-recovery
    public int NetworkRecoveryIntervalSeconds { get; set; } = 10; // recovery retry interval
    public int RequestedHeartbeatSeconds { get; set; } = 60;      // 0 disables heartbeat
    public bool EnablePublisherConfirms { get; set; } = true;     // publish waits for broker ack
}

使用示例

1. 基础生产者

using Smart.RabbitMQ;

var options = new RabbitMqOptions
{
    HostName = "localhost",
    Port = 5672,
    UserName = "guest",
    Password = "guest",
    VirtualHost = "/"
};

var factory = new SmartRabbitMQFactory(options);
var producer = await factory.CreateProducerAsync();

// 声明交换机
await producer.ExchangeDeclareAsync("my.exchange", ExchangeType.Direct);

// 发布消息（默认持久化并启用发布者确认）
var message = new { Id = 1, Content = "Hello RabbitMQ" };
await producer.PublishAsync("my.exchange", "my.routing.key", message);

// 可选：感知无法路由的消息，而不是让 Broker 静默丢弃
producer.BasicReturn += (sender, args) =>
{
    Console.Error.WriteLine($"消息被退回：{args.ReplyText}");
    return Task.CompletedTask;
};

await producer.PublishAsync("my.exchange", "my.routing.key", message, mandatory: true);

// 释放生产者（连接由工厂管理）
await producer.DisposeAsync();

// 完成后释放工厂（关闭共享连接）
await factory.DisposeAsync();

2. 基础消费者

using Smart.RabbitMQ;

var options = new RabbitMqOptions
{
    HostName = "localhost",
    Port = 5672,
    UserName = "guest",
    Password = "guest",
    VirtualHost = "/"
};

var factory = new SmartRabbitMQFactory(options);
var consumer = await factory.CreateConsumerAsync<MyMessage>();

// 订阅消息事件
consumer.MessageReceived += async (sender, args) =>
{
    Console.WriteLine($"收到消息: {args.Data?.Content}");
    Console.WriteLine($"投递标签: {args.DeliveryTag}");
    Console.WriteLine($"交换机: {args.Exchange}");
    Console.WriteLine($"路由键: {args.RoutingKey}");
};

// 声明交换机
await consumer.ExchangeDeclareAsync("my.exchange", ExchangeType.Direct);

// 声明队列
var queueName = await consumer.QueueDeclareAsync("my.queue", durable: true);

// 绑定队列到交换机
await consumer.QueueBindAsync(queueName, "my.exchange", "my.routing.key");

// 设置 QoS（预取计数）
await consumer.BasicQosAsync(prefetchCount: 10);

// 开始消费
var consumerTag = await consumer.StartConsumeAsync(autoAck: false);

// 完成后取消消费
await consumer.CancelConsumeAsync(consumerTag);

// 释放消费者（连接由工厂管理）
await consumer.DisposeAsync();

// 完成后释放工厂（关闭共享连接）
await factory.DisposeAsync();

3. 依赖注入集成

using Microsoft.Extensions.DependencyInjection;
using Smart.RabbitMQ;

// 配置服务
var services = new ServiceCollection();

var rabbitMQOptions = new RabbitMqOptions
{
    HostName = "localhost",
    Port = 5672,
    UserName = "guest",
    Password = "guest",
    VirtualHost = "/"
};

services.AddSmartRabbitMQ(rabbitMQOptions);

// 构建服务提供者
var serviceProvider = services.BuildServiceProvider();

// 获取工厂
var factory = serviceProvider.GetRequiredService<IRabbitMQFactory>();

// 创建生产者
var producer = await factory.CreateProducerAsync();

// 创建消费者
var consumer = await factory.CreateConsumerAsync<MyMessage>();

// 或者直接获取预先创建的单例。
// 注意：容器中只注册了字符串消费者，泛型消费者需通过工厂创建
var producer2 = serviceProvider.GetRequiredService<SmartRabbitMQProducer>();
var stringConsumer = serviceProvider.GetRequiredService<SmartRabbitMQConsumer>();

// 健康检查（临时探测，不影响共享连接）
var check = serviceProvider.GetRequiredService<IRabbitMQCheck>();
bool healthy = await check.CheckAsync();

4. 字符串消息消费者

using Smart.RabbitMQ;

var factory = new SmartRabbitMQFactory(options);
var consumer = await factory.CreateConsumerAsync(); // 字符串消费者

consumer.MessageReceived += async (sender, args) =>
{
    Console.WriteLine($"收到字符串消息: {args.Data}");
};

await consumer.ExchangeDeclareAsync("my.exchange");
var queueName = await consumer.QueueDeclareAsync("my.queue");
await consumer.QueueBindAsync(queueName, "my.exchange", "my.routing.key");
await consumer.StartConsumeAsync();

5. 高级消费者与错误处理

using Smart.RabbitMQ;

var consumer = await factory.CreateConsumerAsync<MyMessage>();

// 确认/拒绝由库内部完成：
//   · 处理器正常返回 -> BasicAck（multiple 为 true 时批量确认）
//   · 处理器抛出异常 -> BasicNack(requeue: requeueOnFailure)
//   · 消息体不是合法JSON -> BasicNack(requeue: false)，因此不会出现毒消息无限重投
consumer.MessageReceived += async (sender, args) =>
{
    await ProcessMessageAsync(args.Data);
};

// 观测失败，且不影响确认流程
consumer.MessageHandleFailed += (sender, e) =>
{
    Console.Error.WriteLine(
        $"消息处理失败（deliveryTag={e.ReceivedArgs.DeliveryTag}，是否重新入队={e.IsRetriable}）：{e.Error.Message}");
};

// 生产环境建议 requeueOnFailure: false，并配合死信交换机使用
await consumer.StartConsumeAsync(autoAck: false, requeueOnFailure: false);

6. 队列参数

using Smart.RabbitMQ;

var queueParams = new Dictionary<string, object?>
{
    ["x-message-ttl"] = 60000,           // 消息 TTL: 60 秒
    ["x-max-length"] = 10000,             // 队列最大长度
    ["x-expires"] = 3600000,               // 队列在 1 小时后过期
    ["x-dead-letter-exchange"] = "dlx"    // 死信交换机
};

var queueName = await consumer.QueueDeclareAsync(
    queue: "my.queue",
    durable: true,
    queueParams: queueParams
);

7. 交换机类型

using Smart.RabbitMQ;

// 直连交换机（点对点）
await producer.ExchangeDeclareAsync("direct.exchange", ExchangeType.Direct);

// 扇形交换机（广播）
await producer.ExchangeDeclareAsync("fanout.exchange", ExchangeType.Fanout);

// 主题交换机（模式匹配）
await producer.ExchangeDeclareAsync("topic.exchange", ExchangeType.Topic);

// 头交换机（基于头的路由）
await producer.ExchangeDeclareAsync("headers.exchange", ExchangeType.Headers);

// 头交换机的绑定必须提供 arguments，否则消息永远不会被路由
await consumer.QueueBindAsync(
    queue: "my.queue",
    exchange: "headers.exchange",
    routingKey: string.Empty,
    arguments: new Dictionary<string, object?>
    {
        ["x-match"] = "all",
        ["type"] = "order"
    });

最佳实践

1. 使用手动确认: 设置 autoAck: false 以更好地控制消息处理和错误处理。

2. 设置预取计数: 使用 BasicQosAsync 限制未确认消息的数量，防止消费者过载。

await consumer.BasicQosAsync(prefetchCount: 10);

3. 释放资源: 使用完毕后释放生产者和消费者实例。工厂管理共享连接。

await using var producer = await factory.CreateProducerAsync();
await using var consumer = await factory.CreateConsumerAsync<MyMessage>();

// 完全完成后，释放工厂以关闭连接
await factory.DisposeAsync();

4. 处理异常: 不要在 MessageReceived 中重新抛出异常——库已经完成了拒绝（Nack），再抛出只会让异常逃逸到客户端的投递分发循环。需要记录日志时请订阅 MessageHandleFailed。

5. 同时使用持久化队列与持久化消息: durable: true 只保证队列定义存活；消息之所以持久化，是因为库在发布时设置了 delivery_mode=2。

6. 配置死信交换机: 在 queueParams 中加入 x-dead-letter-exchange，并在生产环境使用 requeueOnFailure: false，让处理失败的消息进入死信队列，而不是被无限重投。

7. 校验配置: AddSmartRabbitMQ 与 SmartRabbitMQFactory 构造函数都会调用 RabbitMqOptions.Validate()，非法配置会立即抛出 ArgumentException。

8. 并发发布时不要共享生产者: 一个生产者只持有一个通道。高并发发布场景请通过 CreateProducerAsync() 为每个并发单元创建独立生产者，而不是多线程共享同一实例。

事件参数

MessageReceivedArgs<T>

传递给消息接收事件的参数。

public class MessageReceivedArgs<T> : EventArgs
{
    public string? ConsumerTag { get; init; }
    public ulong DeliveryTag { get; init; }
    public string? Exchange { get; init; }
    public string? RoutingKey { get; init; }
    public bool Redelivered { get; init; }
    public T? Data { get; }
}

// Payload of MessageHandleFailed
public class MessageErrorEventArgs<T> : EventArgs
{
    public Exception Error { get; }
    public MessageReceivedArgs<T> ReceivedArgs { get; }
    public bool IsRetriable { get; }   // false => rejected without requeue (DLX candidate)
}

项目结构

Smart.RabbitMQ/
├── SmartRabbitMQFactory.cs           - 用于创建生产者和消费者的工厂
├── SmartRabbitMQProducer.cs          - 消息生产者实现
├── SmartRabbitMQConsumer.cs          - 泛型与字符串消息消费者实现
├── SerializeConvert.cs               - JSON 序列化工具
├── RabbitMqOptions.cs                - 配置选项
├── MessageReceivedArgs.cs            - 事件参数类（MessageReceivedArgs / MessageErrorEventArgs）
├── ExchangeType.cs                   - 交换机类型枚举
├── ExchangeTypeExtensions.cs         - AMQP 交换机类型名称映射
├── IRabbitMQFactory.cs               - 工厂接口
├── IRabbitMQCheck.cs                 - 健康检查契约
└── RabbitMQServiceExtensions.cs      - 依赖注入扩展

支持的 .NET 版本

• .NET 8
• .NET 9
• .NET 10

Developed by zenglei