Hyz.RabbitMQ.Client 0.0.8

Hyz.RabbitMQ.Client

一款统一、优雅的 RabbitMQ 客户端库，专为 .NET 打造。
一个包，零模板代码。

特性

• 🚀 一行注册 — AddRabbitMq() 即可完成所有配置
• 📮 发布与消费 — IPublisherService / IConsumerService 开箱即用
• 🔁 批量处理 — PublishBatchAsync / ConsumeBatchAsync，支持自定义批次大小和超时
• 🌐 多连接管理 — 命名连接，轻松接入多节点 RabbitMQ 集群
• 🧩 源码生成器 — 基于特性声明队列/交换机/绑定（编译时生成）
• 🔍 订阅者扫描 — 自动发现程序集中标记了 [RabbitMqConsumer] 的处理器
• 🎯 IAsyncEnumerable — 现代化 await foreach 消费方式，告别回调地狱
• 🔄 自动重连 — 内置指数/线性/固定退避策略
• 🏗️ 拓扑管理 — IExchangeManager / IQueueManager 运行时管理交换机、队列、绑定
• ✅ 发布确认 — PublishWithConfirmationAsync 确保消息可靠投递

安装

dotnet add package Hyz.RabbitMQ.Client

支持 .NET Framework 4.6.1+, .NET Core 2.0+, .NET 8.0+。

快速开始

1. 注册服务

using Hyz.RabbitMQ.Extensions;

builder.Services.AddRabbitMq(options =>
{
    options.HostName = "localhost";
    options.Port     = 5672;
    options.UserName = "guest";
    options.Password = "guest";
    options.AutoReconnect = true;
});

2. 发布消息

var publisher = sp.GetRequiredService<IPublisherService>();
var message   = new MessageBody(Encoding.UTF8.GetBytes("Hello RabbitMQ!"));

await publisher.PublishAsync("my-queue", message);

3. 消费消息

var consumer = sp.GetRequiredService<IConsumerService>();

await foreach (var msg in consumer.ConsumeAsync("my-queue"))
{
    var text = Encoding.UTF8.GetString(msg.Body);
    Console.WriteLine($"收到消息: {text}");
    await msg.AckAsync();
}

核心概念

发布者 (Publisher)

消费者 (Consumer)

MessageBody

var body1 = new MessageBody(bytes);
var body2 = "text".ToMessageBody();
var body3 = new MessageBody(myObject, serializer);

ConsumerOptions

PublishOptions

var options = new PublishOptions
{
    DeliveryMode  = DeliveryModes.Persistent,
    ContentType   = "application/json",
    CorrelationId = Guid.NewGuid().ToString(),
    Priority      = 5,
    Expiration    = "60000"   // 60 秒过期
};

序列化

库内置 SystemTextJsonSerializer（默认）和 MessagePackSerializer，并可自定义序列化器。

using Hyz.RabbitMQ.Serialization;

// 对象 → MessageBody（默认 JSON）
var body = myOrder.ToMessageBody();

// 对象 → MessageBody（使用 MessagePack）
var msgPackSerializer = new MessagePackSerializer();
var body = myOrder.ToMessageBody(msgPackSerializer);

// MessageBody → 对象
var order = body.FromMessageBody<Order>();

// 字节数组 → 对象
var order = bytes.FromMessageBody<Order>();

// 字符串 ↔ MessageBody
var body = "hello".ToMessageBodyFromString();
var text = body.ToStringContent();

自定义序列化器

实现 IMessageSerializer 接口即可：

public class ProtobufSerializer : IMessageSerializer
{
    public string ContentType => "application/x-protobuf";
    public ReadOnlyMemory<byte> Serialize<T>(T obj) where T : class => /* ... */;
    public T? Deserialize<T>(ReadOnlyMemory<byte> bytes) where T : class => /* ... */;
}

消息处理

ReceivedMessageContext

消费者回调中接收的消息上下文，包含完整的消息元数据：

消息确认

await foreach (var msg in consumer.ConsumeAsync("orders"))
{
    try
    {
        ProcessMessage(msg);
        await msg.AckAsync();        // 确认处理成功
    }
    catch
    {
        await msg.NackAsync(true);   // 拒绝并重新入队
    }
}

消息处理器接口

// 单条消息处理
public class OrderHandler : IMessageHandler
{
    public Task<HandleResult> HandleAsync(ReceivedMessageContext ctx, CancellationToken ct)
    {
        var order = ctx.Body.FromMessageBody<Order>();
        // 处理订单...
        return Task.FromResult(HandleResult.SuccessResult);
    }
}

// 批量消息处理
public class BatchOrderHandler : IBatchMessageHandler
{
    public Task<BatchHandleResult> HandleBatchAsync(
        IReadOnlyList<ReceivedMessageContext> messages, CancellationToken ct)
    {
        var orders = messages.Select(m => m.Body.FromMessageBody<Order>()).ToList();
        // 批量处理...
        return Task.FromResult(BatchHandleResult.AllSuccess(orders.Count));
    }
}

HandleResult 结果类型

拓扑管理

IExchangeManager 和 IQueueManager 用于在运行时管理 RabbitMQ 交换机与队列，需手动实例化：

using Hyz.RabbitMQ.Core;

var provider = sp.GetRequiredService<IConnectionProvider>();
var exchangeManager = new ExchangeManager(provider);
var queueManager = new QueueManager(provider);

交换机管理 (IExchangeManager)

// 声明一个持久化的 Topic 交换机
var exchange = await exchangeManager.DeclareAsync(
    "shop.events",
    ExchangeType.Topic,
    durable: true);

// 绑定队列到交换机
await exchangeManager.BindAsync("shop.events", "orders", "order.created");

// 删除交换机（仅当无队列绑定时）
await exchangeManager.DeleteAsync("shop.events", ifUnused: true);

队列管理 (IQueueManager)

// 声明持久化队列并配置死信与消息 TTL
var queue = await queueManager.DeclareAsync("orders", durable: true, arguments: new Dictionary<string, object?>
{
    ["x-dead-letter-exchange"] = "shop.dlx",
    ["x-message-ttl"] = 60000   // 消息 60 秒未消费则转死信
});
Console.WriteLine($"队列消息数: {queue.MessageCount}, 消费者: {queue.ConsumerCount}");

// 清空队列消息
var purged = await queueManager.PurgeAsync("orders");

// 删除空队列
var deleted = await queueManager.DeleteAsync("orders", ifEmpty: true);

ExchangeType 枚举

连接配置详解

RabbitMqConnectionOptions 支持所有常用 RabbitMQ 连接参数：

重连策略

TLS 配置

services.AddRabbitMq(options =>
{
    options.HostName = "rabbitmq.example.com";
    options.Port = 5671;
    options.EnableTls = true;
    options.TlsOptions = new TlsOptions
    {
        CertPath = "/path/to/client.p12",
        CertPassphrase = "your-password",
        CheckCertificateRevocation = true
    };
});

多连接管理

// 注册多个命名连接
services.AddRabbitMq("Conn1", opts => opts.HostName = "rabbit1.local");
services.AddRabbitMq("Conn2", opts => opts.HostName = "rabbit2.local");

// 按名称获取服务
var pub1 = sp.GetRequiredKeyedService<IPublisherService>("Conn1");
var pub2 = sp.GetRequiredKeyedService<IPublisherService>("Conn2");

订阅者扫描

[RabbitMqConsumer(Queue = "orders", PrefetchCount = 5)]
public class OrderHandler : IMessageHandler
{
    public Task<HandleResult> HandleAsync(ReceivedMessageContext ctx)
    {
        var text = Encoding.UTF8.GetString(ctx.Body);
        Console.WriteLine(text);
        return Task.FromResult(HandleResult.Success);
    }
}

// 扫描并启动
var host = new RabbitMqSubscriberHost(logger, connectionManager);
host.ScanAndRegister(typeof(OrderHandler).Assembly);
await host.StartAsync();

源码生成器

在包含生成代码的 partial 类上通过特性声明交换机、队列和绑定关系，编译器自动生成注册代码。

声明交换机、队列和绑定

using Hyz.RabbitMQ.Abstractions.Attributes;

[RabbitMqExchange(Name = "shop", Type = "direct")]
[RabbitMqQueue(Name = "orders", Durable = true)]
[RabbitMqBinding(Exchange = "shop", RoutingKey = "order.created")]
public static partial class ShopSubscriptions
{
    [RabbitMqSubscribe(Queue = "orders")]
    public static partial Task OnOrderCreatedAsync(ReceivedMessageContext ctx);

    [RabbitMqBatchSubscribe(Queue = "batch-orders", BatchSize = 50)]
    public static partial Task OnBatchOrdersAsync(IList<ReceivedMessageContext> batch);
}

特性详解

[RabbitMqExchange] — 声明交换机

[RabbitMqQueue] — 声明队列

[RabbitMqBinding] — 声明绑定

[RabbitMqSubscribe] — 单条订阅方法

[RabbitMqBatchSubscribe] — 批量订阅方法

高级特性

死信队列 (DLX)

当消息处理失败（达到最大重试次数）或被拒绝（requeue = false）时，可自动路由到死信队列：

// 订阅者扫描方式
[RabbitMqConsumer(
    Queue = "orders",
    MaxRetryCount = 3,
    DeadLetterExchange = "shop.dlx",
    DeadLetterRoutingKey = "order.failed")]
public class OrderHandler : IMessageHandler { /* ... */ }

// 源码生成器方式
[RabbitMqQueue(
    Name = "orders",
    DeadLetterExchange = "shop.dlx",
    DeadLetterRoutingKey = "order.failed")]
[RabbitMqExchange(Name = "shop.dlx", Type = "topic")]
[RabbitMqBinding(Exchange = "shop.dlx", RoutingKey = "order.failed")]
public static partial class ShopDlxSubscriptions { /* ... */ }

RPC 请求/响应模式

利用 CorrelationId 和 ReplyTo 实现远程过程调用：

// 客户端 — 发送 RPC 请求
var options = new PublishOptions
{
    CorrelationId = Guid.NewGuid().ToString(),
    ReplyTo = "rpc.reply.queue"
};
await publisher.PublishToExchangeAsync("rpc.exchange", "rpc.method", request, options);

// 服务端 — 处理并回复
await foreach (var msg in consumer.ConsumeAsync("rpc.queue"))
{
    var response = HandleRequest(msg);
    await publisher.PublishAsync(msg.ReplyTo!, response, new PublishOptions
    {
        CorrelationId = msg.CorrelationId
    });
    await msg.AckAsync();
}

依赖项

安装本包时会自动引入以下依赖：

许可证

MIT License - 详见 LICENSE 文件

如果这个项目对你有帮助，请给它一个 ⭐️