Galosys.Foundation.RocketMQ.Abstractions 26.8.18.1

Galosys.Foundation.RocketMQ.Abstractions

成熟度: 🟢 稳定 — 核心功能 + 高级特性（事务/顺序/批量）已完成

RocketMQ 抽象层模块，定义统一的消息发送/消费抽象基类、配置模型、[RmqHandler] 注解、可观测性基类。用户代码仅依赖此模块的 API，切换底层实现（Apache / NewLife）只需改 PackageReference。

架构

Galosys.Foundation.RocketMQ.Abstractions  ← 本模块（抽象层）
    ▲                    ▲
    │                    │
Galosys.Foundation.RocketMQ.Client      Galosys.Foundation.NewLife.RocketMQ
(Apache rocketmq.client 5.x)            (NewLife.RocketMQ 3.0)

功能特性

• 统一 API — RmqTemplate + [RmqHandler] + IMessageHandler<T>，切换实现零改动
• 声明式消费 — [RmqHandler("topic", "tag")] 方法级注解，自动扫描注册
• 事务消息 — SendTransactionAsync 半消息 + 本地事务 + Commit/Rollback
• 顺序消息 — SendOrderedAsync 相同 shardingKey 路由到同一队列
• 批量发送 — SendBatchAsync 一次发送多条消息
• 可观测性基类 — RmqTracing（OTel ActivitySource）+ RmqMeter（6 个 Counter/Histogram）
• 健康检查基类 — RmqHealthCheck（IHealthCheck）
• 生命周期管理 — RmqBootstrapper（IHostedLifecycleService，自动扫描 + 优雅关闭）

安装

通常不直接安装此包，而是选择一个实现包（二选一）：

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


<PackageReference Include="Galosys.Foundation.NewLife.RocketMQ" Version="x.x.x" />

快速开始

1. 最小配置

{
  "RocketMQ": {
    "Endpoints": "localhost:9876"
  }
}

ConsumerGroup 默认取应用名称并将 . 替换为 _，通常无需显式配置。

2. 注册服务

// 方式一：模块自动发现（推荐）
await Host.CreateDefaultBuilder()
    .UseModularization()
    .Locate()
    .RunAsync();

// 方式二：手动注册
services.AddRocketMQ(configuration);

3. 发送消息

发送侧无泛型约束，支持任意 POCO（无需实现 IMessage）：

using Galosys.Foundation.RocketMQ.Abstractions;

// 纯 POCO，无需 IMessage
public class OrderCreatedEvent
{
    public string OrderNo { get; set; }
    public decimal Amount { get; set; }
}

public class OrderService
{
    private readonly RmqTemplate _template;

    public OrderService(RmqTemplate template) => _template = template;

    // 普通消息
    public async Task SendAsync() =>
        await _template.SendAsync("order_created", new OrderCreatedEvent { OrderNo = "ORD001", Amount = 99.9m });

    // 指定 Tag + Keys（消息键，用于按 key 查询）
    public async Task SendWithTagAsync() =>
        await _template.SendAsync("order_created", tag: "new", keys: "ORD001", new OrderCreatedEvent { OrderNo = "ORD001" });

    // 延迟消息
    public async Task SendDelayAsync() =>
        await _template.SendAsync("order_created", new OrderCreatedEvent { OrderNo = "ORD001" }, delaySeconds: 30);

    // 事务消息（半消息 → 本地事务 → Commit/Rollback）
    public async Task SendTransactionAsync() =>
        await _template.SendTransactionAsync(
            new OrderCreatedEvent { OrderNo = "ORD001" },
            "order_created", "new",
            localTransaction: async () =>
            {
                // 执行本地事务（如扣库存、写订单）
                return true; // true=Commit, false=Rollback
            });

    // 顺序消息（相同 shardingKey 路由到同一队列，保证消费顺序）
    public async Task SendOrderedAsync() =>
        await _template.SendOrderedAsync(
            new OrderCreatedEvent { OrderNo = "ORD001" },
            "order_created", shardingKey: "ORD001");

    // 批量发送
    public async Task SendBatchAsync() =>
        await _template.SendBatchAsync(
            new[]
            {
                new OrderCreatedEvent { OrderNo = "ORD001" },
                new OrderCreatedEvent { OrderNo = "ORD002" },
            },
            "order_created");
}

消费侧仍需 IMessageHandler<T>（T : IMessage），发送侧无此限制。

4. 消费消息

using Galosys.Foundation.RocketMQ.Abstractions;

[Handler]
public class OrderHandler : IMessageHandler<OrderCreatedMessage>
{
    [RmqHandler("order_created")]
    public Task<bool> HandleAsync(OrderCreatedMessage msg)
    {
        // 处理消息
        return Task.FromResult(true); // true = ACK, false = Nack（重新投递）
    }
}

// 自定义消费组 + Tag 过滤
[Handler]
public class VipOrderHandler : IMessageHandler<OrderCreatedMessage>
{
    [RmqHandler("order_created", "vip", ConsumerGroup = "vip_order_group")]
    public Task<bool> HandleAsync(OrderCreatedMessage msg) => Task.FromResult(true);
}

注册机制： 类标注 [Handler] → DI 自动注册；方法标注 [RmqHandler] → 启动时自动扫描订阅 Topic。类需实现 IMessageHandler 或 IMessageHandler<T>。

详细配置

完整配置示例

{
  "RocketMQ": {
    "Endpoints": "localhost:9876",
    "AccessKey": null,
    "SecretKey": null,
    "Ssl": false,
    "RequestTimeout": 10,
    "ConsumerGroup": "my_service",
    "Handlers": {
      "order_created": {
        "PrefetchCount": 16,
        "Concurrency": 4,
        "MaxRetryCount": 3,
        "RetryBaseDelayMs": 1000
      }
    }
  }
}

根配置（RmqOptions）

Handler 配置（RmqHandlerProperty）

可观测性

健康检查

实现模块在 AddRocketMQ() 中自动注册健康检查，访问 /health 即可查看 RocketMQ 状态。

健康检查名称为 "rocketmq"。

OpenTelemetry 追踪

ActivitySource 名称：Galosys.Foundation.RocketMQ

指标

Meter 名称：Galosys.Foundation.RocketMQ

核心类

Outbox Transport

本模块提供 RocketMQ Outbox 传输支持。只要引用实现包（如 Galosys.Foundation.RocketMQ.Client），RocketMQTransport 即自动注册为 IOutboxTransport。

// 无需手动注册——RocketAbstractionsModule 自动调用 AddRocketMQTransport()
// 传输名称默认 "RocketMQ"，可通过 MessageBus:Outbox:DefaultTransportName 配置

如需手动替换传输实现：

services.RemoveAll<IOutboxTransport>();
services.TryAddSingleton<IOutboxTransport, MyCustomTransport>();

依赖

• Galosys.Foundation.Core
• Microsoft.Extensions.Diagnostics.HealthChecks
• Microsoft.Extensions.Hosting
• System.Diagnostics.Metrics