Rabbit.Common.MQ.RabbitMQ 1.4.1

Rabbit.Common.MQ.RabbitMQ

Native RabbitMQ client with producer/consumer pattern, delayed messages, dead letter queues, and automatic reconnection. Inject IMessageProducer to publish, implement IMessageConsumer<T> to consume.

原生 RabbitMQ 客户端，轻量级高性能消息队列组件。

安装

dotnet add package Rabbit.Common.MQ.RabbitMQ

依赖

• Rabbit.Common.Lite
• RabbitMQ.Client 7.2.1

特性

• ✅ 轻量级（~500KB）
• ✅ 高性能（原生 RabbitMQ 客户端）
• ✅ 自动重连
• ✅ 延迟消息（TTL + 死信）
• ✅ 死信队列
• ✅ 消费者自动注册
• ✅ 自动重试（可配置次数和间隔）

使用方式

Console 项目

using Microsoft.Extensions.DependencyInjection;
using Rabbit.Common.MQ.RabbitMQ;

// 1. 创建 DI 容器
var services = new ServiceCollection();

// 2. 注册 RabbitMQ
services.AddRabbitMQ(options =>
{
    options.HostName = "localhost";
    options.Port = 5672;
    options.UserName = "guest";
    options.Password = "guest";
    options.VirtualHost = "/";
});

// 仅生产者模式（不启动消费者）
// services.AddRabbitMQProducer(options => { ... });

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

// 4. 发布消息
var producer = provider.GetRequiredService<IMessageProducer>();
await producer.PublishAsync("order.created", new Order { Id = 1, Amount = 99.9m });

// 延迟消息
await producer.PublishDelayedAsync("order.cancel", order, TimeSpan.FromMinutes(30));

Web API 项目

using Rabbit.Common.MQ.RabbitMQ;

services.AddRabbitMQ(options =>
{
    options.HostName = "localhost";
    options.Port = 5672;
    options.UserName = "guest";
    options.Password = "guest";
    options.VirtualHost = "/";
    options.PublisherConfirms = true;
    options.PersistentMessages = true;
});

// 或从配置文件读取
services.AddRabbitMQ("RabbitMQ");

消费消息

using Rabbit.Common.MQ.RabbitMQ;

// 消费者自动扫描注册，只需添加特性
[RabbitMQConsumer("order.created", "order-service")]
public class OrderConsumer : IMessageConsumer<Order>
{
    public async Task ConsumeAsync(Order message, CancellationToken cancellationToken)
    {
        // 处理消息...
        await Task.CompletedTask;
    }
}

常用示例

发布消息

// 发布普通消息
await _producer.PublishAsync("order.created", new Order { Id = 1, Amount = 99.9m });

// 发布到指定交换机
await _producer.PublishAsync("orders", "order.created", order);

// 发布延迟消息（30分钟后触发）
await _producer.PublishDelayedAsync("order.cancel", order, TimeSpan.FromMinutes(30));

死信队列

// 启用死信队列（消费失败后自动进入死信队列）
[RabbitMQConsumer("order.created", "order-service", EnableDeadLetter = true)]
public class OrderConsumer : IMessageConsumer<Order> { }

// 死信队列消费者（处理失败消息）
[RabbitMQConsumer("#", "order-service.dlq")]
public class DeadLetterConsumer : IMessageConsumer<Order> { }

重试机制

// 配置重试参数
services.AddRabbitMQ(options =>
{
    options.MaxRetryCount = 3;      // 最大重试 3 次
    options.RetryInterval = 5000;   // 每次间隔 5 秒
});

// 消费失败后自动重试，达到上限后进入死信队列

Topic 交换机路由

// 发布：支持通配符路由
[RabbitMQConsumer("order.*", "all-orders", Exchange = "orders", ExchangeType = "topic")]
public class AllOrdersConsumer : IMessageConsumer<Order> { }

// 发布消息
await _producer.PublishAsync("orders", "order.created", order);  // 匹配 order.*
await _producer.PublishAsync("orders", "order.paid", order);     // 匹配 order.*

Fanout 广播

// 多个消费者同时接收同一条消息
[RabbitMQConsumer("", "service-a", Exchange = "broadcast", ExchangeType = "fanout")]
public class ServiceAConsumer : IMessageConsumer<Order> { }

[RabbitMQConsumer("", "service-b", Exchange = "broadcast", ExchangeType = "fanout")]
public class ServiceBConsumer : IMessageConsumer<Order> { }

// 发布广播消息
await _producer.PublishAsync("broadcast", "", order);

仅生产者模式

// 不启动消费者，仅用于发送消息
services.AddRabbitMQProducer(options =>
{
    options.HostName = "localhost";
    options.MaxRetryCount = 0;  // 禁用重试
});

配置选项

public class RabbitMQOptions
{
    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;
    public int NetworkRecoveryInterval { get; set; } = 5;

    // 消息选项
    public bool PersistentMessages { get; set; } = true;
    public bool PublisherConfirms { get; set; } = true;

    // 消费选项
    public int ConsumerConcurrency { get; set; } = 1;
    public ushort PrefetchCount { get; set; } = 10;

    // 死信队列配置
    public string DeadLetterExchangeSuffix { get; set; } = ".dlx";
    public string DeadLetterQueueSuffix { get; set; } = ".dlq";

    // 重试配置
    public string RetryExchangeSuffix { get; set; } = ".retry";
    public int MaxRetryCount { get; set; } = 3;
    public int RetryInterval { get; set; } = 5000;
}

重试机制

当消息消费失败时，会自动进行重试：

1. 首次失败：消息发送到重试队列（带 TTL）
2. TTL 到期：消息通过死信机制返回原队列重新消费
3. 重试次数记录：在消息头 x-retry-count 中记录
4. 达到上限：超过 MaxRetryCount 后进入死信队列

消费者特性

[RabbitMQConsumer(
    routingKey: "order.created",     // 路由键
    queue: "order-service",          // 队列名
    Exchange = "orders",              // 交换机（可选）
    ExchangeType = "topic",           // 交换机类型
    Durable = true,                   // 持久化
    Concurrency = 1,                  // 并发数
    EnableDeadLetter = true           // 启用死信
)]

与 CAP 的区别

许可证

MIT