Galosys.Foundation.NewLife.RocketMQ 26.5.20.1

Galosys.Foundation.NewLife.RocketMQ

成熟度: 🟡 预览 — 基于 NewLife.RocketMQ 3.0，核心功能完成

基于 NewLife.RocketMQ 的 RocketMQ 客户端实现，支持 Remoting + gRPC 双协议，兼容 RocketMQ 4.x 和 5.x。

为什么选择 NewLife 实现

安装

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

快速开始

1. 最小配置

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

NewLife 实现连接 NameServer（默认 9876），与 Apache 实现连接 Proxy（8081）不同。

2. 注册服务

services.AddRocketMQ(configuration);

3. 发送消息

using Galosys.Foundation.RocketMQ.Abstractions;

public class OrderService
{
 private readonly RmqTemplate _template;

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

 public async Task SendAsync() =>
 await _template.SendAsync("order_created", new OrderCreatedMessage { OrderNo = "ORD001" });

 public async Task SendDelayAsync() =>
 await _template.SendAsync("order_created", new OrderCreatedMessage { OrderNo = "ORD001" }, delaySeconds: 30);
}

延迟消息自动映射到 RocketMQ 18 级预设延迟等级（1s/5s/10s/30s/1m/2m/.../2h）。

4. 消费消息

using Galosys.Foundation.RocketMQ.Abstractions;

[Handler]
public class OrderHandler : IMessageHandler<OrderCreatedMessage>
{
 [RmqHandler("order_created")]
 public Task<bool> HandleAsync(OrderCreatedMessage msg) => Task.FromResult(true);
}

完整配置

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

详细配置说明参见 。

实现细节

切换实现

从 Apache 实现切换到 NewLife 实现，业务代码无需任何修改：

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

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

同时将 Endpoints 从 Proxy 地址（如 localhost:8081）改为 NameServer 地址（如 localhost:9876）。

依赖

• Galosys.Foundation.RocketMQ.Abstractions
• NewLife.RocketMQ 3.0（含 NewLife.Core）