Hyz.RabbitMQ.Client 0.0.5

Hyz.RabbitMQ.Client

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

特性

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

安装

dotnet add package Hyz.RabbitMQ.Client

需要 .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 秒过期
};

多连接管理

// 注册多个命名连接
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();

源码生成器

[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);
}

依赖项

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

许可证

MIT License - 详见 LICENSE 文件

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