Galosys.Foundation.RabbitMQ.Client 26.7.20.2

Galosys.Foundation.RabbitMQ.Client

目标框架:
et8.0\ ·
et10.0\（#if NET10_0\ 条件编译）

定位: RabbitMQ 高性能客户端，连接池 + Publisher Confirms + OTel 可观测

基于 RabbitMQ.Client 的 RabbitMQ 实现，实现 Galosys.Foundation.Amqp 的 IAmqpPublisher / IAmqpConsumer 接口。提供共享连接管理、批量发布、Publisher Confirms、消费并发控制、DLX/DLQ、OpenTelemetry 追踪/指标、健康检查。

架构

IAmqpPublisher                    IAmqpConsumer
  └─ ConvertAndSendAsync            ├─ HandleAsync
                                    └─ CancelAsync

RabbitPublisher : IAmqpPublisher  RabbitConsumer : IAmqpConsumer
  └─ 共享 IConnection + 临时通道    └─ 连接池 + 信号量并发控制

AmqpTemplate : IAmqpPublisher, IAmqpConsumer  [密封门面]
  └─ 组合 RabbitPublisher + RabbitConsumer

功能特性

• 统一发送 — RabbitPublisher.ConvertAndSendAsync 为框架唯一发送方法，内部 seq 追踪 + ACK/NACK + 超时
• WaitForConfirm 控制 — 默认等待 broker ACK（超时可配 PublishConfirmTimeoutMs），非核心可 WaitForConfirm=false
• 连接管理 — 发布端共享单 IConnection + 临时通道，消费端 ConnectionPool 多连接 Round-Robin 通道分配
• 批量发布 — 同通道批量发送 + Properties 复用 + 逐条 Confirm 追踪
• 拓扑缓存 — TopologyCache 避免重复声明队列/交换机
• DLX/DLQ — 不可达消息转发到 {routingKey}_dlx + 递归防护
• 消费重试 — x-delayed-message 指数退避 + DLQ fallback
• OTel 追踪 — W3C traceparent 注入，Producer/Consumer Span
• OTel 指标 — 覆盖发送、消费、延迟、批量大小、连接/通道活跃数
• 健康检查 — RabbitMQHealthCheck 聚合发布者/消费者连接池状态
• IMessageBus — RabbitMessageBus 实现 Core 的 IMessageBus 接口（Send + Publish）

前置条件

• RabbitMQ 3.x+
• 延迟消息需 rabbitmq_delayed_message_exchange 插件
• 队列深度采集需开启 rabbitmq_management 插件，且配置 "RabbitMQ:Management:Enabled": true

安装

<PackageReference Include="Galosys.Foundation.RabbitMQ.Client" Version="x.x.x" />

快速开始

1. 最小配置

{
  "RabbitMQ": {
    "HostName": "localhost",
    "Port": 5672,
    "UserName": "guest",
    "Password": "guest",
    "VirtualHost": "/"
  }
}

2. 注册服务

services.AddAmqp(configuration);
services.AddRabbitMQ(configuration);

3. 发送消息

using RabbitMQ.Client;

public class OrderService
{
    private readonly AmqpTemplate _amqp;

    public OrderService(AmqpTemplate amqp) => _amqp = amqp;

    public async Task SendAsync() =>
        await _amqp.SendAsync("order_created", new[] { new { OrderNo = "ORD001" } });

    public async Task BroadcastAsync() =>
        await _amqp.BroadcastAsync("order_events", new[] { new { OrderNo = "ORD001" } });

    public async Task DelayAsync() =>
        await _amqp.DelayAsync("order_created", new[] { new { OrderNo = "ORD001" } }, 5000);
}

4. 消费消息

using RabbitMQ.Client;

[Handler]
public class OrderHandler : IMessageHandler
{
    [AmqpHandler("order_created")]
    public async Task<bool> HandleAsync(IMessage msg)
    {
        // 处理消息
        return true; // true = ACK, false = Nack 进入 DLQ
    }
}

5. 缓冲聚批发送（日志/指标场景）

var batch = serviceProvider.GetRequiredService<BatchingAmqpTemplate>();
await batch.SendAsync(new PendingMessage
{
    RoutingKey = "app.log",
    Body = Encoding.UTF8.GetBytes("log entry"),
    Properties = new AmqpProperties().Default()
});

详细配置

完整配置示例

{
  "RabbitMQ": {
    "HostName": "rabbit1, rabbit2, rabbit3",
    "UserName": "user",
    "Password": "pass",
    "VirtualHost": "/",
    "AutoCreateEnabled": true,
    "NetworkRecoveryInterval": 10,
    "RequestedHeartbeat": 45,
    "Limit": {
      "ProducerMaxConnections": 4,
      "ProducerChannelMaxPerConnection": 10,
      "ConsumerMaxConnections": 4,
      "ConsumerChannelMaxPerConnection": 50,
      "MaxMessageSize": 65536,
      "MaxTps": 0,
      "BatchSize": 100,
      "BatchTimeoutMs": 50,
      "PublishConfirmTimeoutMs": 3000,
      "RetryCount": 3,
      "PipelineCapacity": 5000
    },
    "Management": {
      "Enabled": true,
      "BaseUrl": "http://localhost:15672"
    }
  }
}

可观测性

健康检查

注册端点 /health 可查看 RabbitMQ 状态。检查标签 "rabbitmq" / ["rabbitmq", "messaging"]。

状态：Healthy（全部正常）/ Degraded（部分断连）/ Unhealthy（全部断开）。

OpenTelemetry 追踪

ActivitySource 名称：Galosys.Foundation.RabbitMQ

指标

Meter 名称：Galosys.Foundation.RabbitMQ

组件列表

依赖

• Galosys.Foundation.Amqp：抽象层
• Galosys.Foundation.Core：IMessageBus、IObjectSerializer
• RabbitMQ.Client：NuGet