Galosys.Foundation.RocketMQ.Client 26.5.20.1

Galosys.Foundation.RocketMQ.Client

成熟度: 🟡 预览 — 基于 Apache rocketmq.client 5.x，核心功能完成

基于 Apache rocketmq.client 5.x 的 RocketMQ 客户端实现，使用 gRPC 协议与 5.x Proxy 通信。

前置条件

• RocketMQ 5.x 且已启动 Proxy 组件（gRPC 协议，默认端口 8081）
• 不兼容 4.x NameServer 直连模式（如需 4.x 兼容请使用 Galosys.Foundation.NewLife.RocketMQ）

安装

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

快速开始

1. 最小配置

{
 "RocketMQ": {
 "Endpoints": "localhost:8081"
 }
}

注意： Apache 5.x 客户端使用 gRPC 端口（默认 8081），不是 NameServer 的 9876。

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

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:8081",
 "AccessKey": null,
 "SecretKey": null,
 "Ssl": false,
 "RequestTimeout": 10,
 "ConsumerGroup": "my_service",
 "Handlers": {
 "order_created": {
 "PrefetchCount": 16,
 "Concurrency": 4,
 "MaxRetryCount": 3,
 "RetryBaseDelayMs": 1000
 }
 }
 }
}

详细配置说明参见 。

实现细节

依赖

• Galosys.Foundation.RocketMQ.Abstractions
• rocketmq.client 5.1.0