项目地址：eventhorizon-cli/EventHorizon.RocketMQ

NuGet：EventHorizon.RocketMQ.Grpc | EventHorizon.RocketMQ.Remoting

前言

我一直很喜欢 RocketMQ。它并不只是吞吐高；在电商这类高并发场景中，从可靠投递、顺序消费到延迟与事务消息，RocketMQ 提供了一套相当完整的消息能力。

很久以前，我就想写一个 .NET 的 RocketMQ 客户端。那时 RocketMQ 5 还没有发布，客户端主要使用传统的 Remoting 协议。Remoting 协议经过多年演进，除了消息收发外，还涉及 NameServer 路由、Broker 通信、心跳、负载均衡、消费进度、重试和事务消息等大量细节，实现完整客户端的工作量很大。

当时项目只写了一个开头，后来受限于时间和精力便搁置了。

后来 RocketMQ 5 引入了 gRPC 协议，并发布了官方 C# 客户端。官方客户端已能满足通过 RocketMQ 5 Proxy 接入的消息收发需求；新项目接入 RocketMQ 时，应优先评估官方客户端。

不过，官方 C# 客户端主要面向 gRPC 协议，而实际环境中仍有大量采用 NameServer 发现 Broker 的 Remoting 部署。除此之外，.NET 应用通常还需要和 DI、Generic Host、Options、日志以及 OpenTelemetry 更自然地集成。

我希望这个项目不仅能完成消息的收发，也能提供更好的使用体验：应用可以通过 DI 注册和注入 Producer、Consumer 与 Admin API，让客户端跟随 Generic Host 一起启动和停止；需要连接多个集群时，可以使用 Keyed Service 区分不同客户端；同时通过 OpenTelemetry 将消息发送、接收和消费处理纳入现有的 Trace 和 Metrics 体系。

借助最近的 Vibe Coding，我重新把这个项目捡了起来。在实现 RocketMQ 5 gRPC 客户端的同时，也补充了 classic Remoting 协议的支持，并尝试以更符合 .NET 应用习惯的方式组织两种协议下常见的 Producer、Consumer 和 Admin API。

功能概览

✅ 表示客户端已实现对应 API。实际是否可用，还取决于 RocketMQ 服务端版本、Broker 配置和 Proxy 能力。

上表列出了项目当前实现的主要能力。为了节约篇幅，本文不会逐一介绍每个 API，也不会把所有消息类型和消费者模型都展开。下面只选择几个最常用的场景：协议选择、客户端注册、发送消息、消费消息和 OpenTelemetry 接入。

完整的功能说明、协议差异和可运行示例，可以参考项目仓库中的 README 与 samples。

它不是官方客户端的替代品

RocketMQ 5 的 gRPC 协议是面向多语言客户端设计的新协议。它需要 RocketMQ 5 服务端和 Proxy，使用的是重新设计过的 API。对于能够使用这条链路的新项目，应优先评估官方 C# 客户端。

EventHorizon.RocketMQ 不尝试在 API 层面兼容或替代官方客户端。它提供的是另一种 .NET 客户端实现，并同时保留 classic Remoting 的接入能力。

两条协议的边界如下：

同一应用可以同时引用两个 Package，但它们的 Message、发送回执、异常和 Consumer 模型都是协议专用的类型。需要同时使用两种协议时，应该显式为同名类型设置别名，而不是把其中一个协议的模型传递给另一个协议。

using GrpcMessage = EventHorizon.RocketMQ.Grpc.Producer.Message;
using RemotingMessage = EventHorizon.RocketMQ.Remoting.Producer.Message;

快速接入

按应用实际使用的协议安装对应的 Package；只有同时使用两种协议时，才需要同时安装两个：

dotnet add package EventHorizon.RocketMQ.Grpc
dotnet add package EventHorizon.RocketMQ.Remoting

两个 Package 的公开 API 是各自独立的，不会在客户端 API 层面相互依赖。

先启动本地测试环境

下面的示例使用仓库提供的 本地 Apache RocketMQ 5.5.0 环境。在仓库根目录执行：

docker compose -f test-environments/rocketmq/compose.yaml up -d --wait

环境会启动 NameServer、Broker、cluster-mode Proxy 和 Dashboard；宿主机上的 gRPC Proxy 地址是
127.0.0.1:8081，Remoting 的 NameServer 地址是 127.0.0.1:9876。一次性 resource-init 服务会在
Compose 命令返回前自动创建普通 Topic eventhorizon-test-topic，因此接下来的发送和订阅示例可以直接使用它。

后文会使用 OpenTelemetry 查看 Metrics 和 Trace，可一并启动仓库提供的 Grafana OTEL LGTM 环境：

docker compose -f test-environments/otel-lgtm/compose.yaml up -d --wait

Grafana 地址为 http://127.0.0.1:3000，默认账号和密码均为 admin。该环境同时提供 OTLP Collector，后面的示例会把客户端遥测数据导出到这里。

如果需要验证其他场景，也可以自行创建 Topic：打开本地 RocketMQ Dashboard，在 Topic 管理页面创建：

也可以在环境启动后执行命令：

docker compose -f test-environments/rocketmq/compose.yaml exec broker sh mqadmin updateTopic \
  -n nameserver:9876 -c DefaultCluster -t my-test-topic

本地环境为了便于验证暴露了上述端口；生产环境仍应由运维显式准备 Topic、Consumer Group、ACL、TLS 和 Broker 对外可达地址。

gRPC：通过 Proxy 连接 RocketMQ 5

下面的示例注册一个 gRPC Producer。Endpoint 可以配置一个或多个以分号分隔的 Proxy 地址；当应用运行在 Generic Host 中时，客户端会随 Host 一起启动和停止。

using EventHorizon.RocketMQ.Grpc;
using EventHorizon.RocketMQ.Grpc.Producer;
var builder = WebApplication.CreateBuilder(args);
var rocketMQ = builder.Services.AddRocketMQGrpc(options =>
{
    // 仓库测试环境暴露的 gRPC Proxy；多个地址可用分号分隔。
    options.Endpoint = "127.0.0.1:8081";
    // 控制单次请求、路由缓存和客户端心跳的时序。
    options.RequestTimeout = TimeSpan.FromSeconds(3);
    options.RouteCacheDuration = TimeSpan.FromMinutes(5);
    options.HeartbeatInterval = TimeSpan.FromSeconds(30);
    // 测试环境未启用 TLS/ACL。生产环境按需启用 TLS，并从安全配置读取 AK/SK 和可选 SecurityToken。
    options.UseTLS = false;
    // options.AccessKey = builder.Configuration["RocketMQ:AccessKey"];
    // options.AccessSecret = builder.Configuration["RocketMQ:AccessSecret"];
    // options.SecurityToken = builder.Configuration["RocketMQ:SecurityToken"];
    // 多租户环境可按需设置 Namespace，为 Topic 和 Consumer Group 添加前缀。
    // options.Namespace = "tenant-a";
});
rocketMQ.AddGrpcProducer();

Remoting：通过 NameServer 发现 Broker

classic Remoting 客户端首先连接 NameServer，再根据路由信息直接连接对应的 Broker：

using EventHorizon.RocketMQ.Remoting;
using EventHorizon.RocketMQ.Remoting.Producer;
var builder = WebApplication.CreateBuilder(args);
var rocketMQ = builder.Services.AddRocketMQRemoting(options =>
{
    // 仓库测试环境的 NameServer；多个地址同样可以用分号分隔。
    options.NamesrvAddr = "127.0.0.1:9876";
});
rocketMQ.AddRemotingProducer(options =>
{
    options.GroupName = "eventhorizon-blog-remoting-producer";
});

这意味着应用进程必须能访问 NameServer 返回的每个 Broker 地址。对于容器、Kubernetes 或跨网络部署，Broker 对外注册的地址是否可达，和客户端代码本身同样重要。

多集群或多实例

一个 Host 连接多个集群，或者注册同一角色的多个实例时，可以使用 .NET Keyed Service。注册名同时也是解析服务时使用的 key：

using EventHorizon.RocketMQ.Grpc;
using EventHorizon.RocketMQ.Grpc.Producer;
using Microsoft.Extensions.DependencyInjection;
builder.Services
    .AddRocketMQGrpc("local-rocketmq", options => options.Endpoint = "127.0.0.1:8081")
    .AddGrpcProducer();
public sealed class OrderPublisher(
    [FromKeyedServices("local-rocketmq")] IGrpcProducer producer)
{
}

默认注册仍然使用普通构造函数注入。只有在不使用 Generic Host、直接从独立 ServiceProvider 解析客户端时，才需要自行调用客户端的 StartAsync 和 StopAsync。

发送消息

注册完成后，直接注入协议专用的 Producer 即可。下面是最常见的 gRPC 发送路径：

using System.Text;
using EventHorizon.RocketMQ.Grpc.Producer;
public sealed class OrderPublisher(IGrpcProducer producer)
{
    public async Task<string> PublishAsync(
        string orderId,
        CancellationToken cancellationToken)
    {
        var message = new Message("eventhorizon-test-topic", Encoding.UTF8.GetBytes(orderId))
        {
            Tag = "created",
            MessageGroup = orderId
        };
        message.Keys.Add(orderId);
        message.Properties["source"] = "checkout";
        var receipt = await producer.SendAsync(message, cancellationToken);
        return receipt.MessageId;
    }
}

设置 MessageGroup 会将消息标记为 FIFO，并按该 Group 选择稳定队列；同组由 FIFO Consumer 串行处理，不同组可在可用并发度内并行。普通消息不需要设置它。

gRPC 的 FIFO、定时/延迟、优先级和 Lite 是互斥的消息类型，不能在同一条消息上同时设置；事务消息也不能与它们组合。分别使用方式如下：

• 设置 DeliveryTimestamp 发送定时或延迟消息；在投递前可使用发送回执中的 RecallHandle 撤回。
• 设置非负 Priority 发送优先级消息。
• 设置 LiteTopic 发送 Lite 消息。
• 使用 SendTransactionAsync 发送事务消息，并在 Producer 启动前为事务 Topic 配置检查器。

事务消息不会天然保证业务一致性。事务检查器必须能根据持久化的本地业务状态给出 Commit、Rollback 或 Unknown，不能只依赖进程内内存状态。

Remoting Producer 也提供普通、FIFO、延迟、优先级、Lite 和事务消息；除此之外还支持批量发送、单向发送、指定队列、队列选择器和请求-响应。发送结果是 RemotingSendResult，部分 Broker 返回的非成功发送状态不会直接表现为异常，因此业务应检查返回状态。

消费消息

gRPC 的消费模型

gRPC 的 Consumer 模型较少，但把“应用自己控制接收”与“客户端自动分发”分得很清楚：

gRPC 的 Push 和 LitePush 都是客户端主动发起分配查询和 ReceiveMessage 长轮询，不是 Broker 主动向应用建立推送连接。SimpleConsumer 则把接收和确认时机明确交给业务代码。

PushConsumer

业务不需要自己写长轮询循环时，可以使用 IGrpcPushConsumer。它由客户端自动查询分配结果、通过长轮询接收消息，并分发给 Handler：

using EventHorizon.RocketMQ.Grpc.Consumer;
using EventHorizon.RocketMQ.Grpc.Consumer.Push;
using Microsoft.Extensions.DependencyInjection;
rocketMQ.AddGrpcPushConsumer<OrderCreatedHandler>(ServiceLifetime.Scoped, options =>
{
    options.GroupName = "eventhorizon-blog-grpc-push-consumer";
    options.MaxConcurrency = 8;
    options.MaxDeliveryAttempts = 16;
    options.InvisibleDuration = TimeSpan.FromSeconds(30);
    options.ConsumeTimeout = TimeSpan.FromMinutes(15);
    options.Subscribe("eventhorizon-test-topic", new FilterExpression("created"));
});
public sealed class OrderCreatedHandler : IGrpcPushMessageHandler
{
    public ValueTask<ConsumeResult> HandleAsync(
        GrpcMessageView message,
        CancellationToken cancellationToken)
    {
        // 处理 message.Body。
        return ValueTask.FromResult(ConsumeResult.Success);
    }
}

Handler 返回 Success 时确认消息；返回 Retry 时请求重新投递；返回 DeadLetter 时将消息转入死信队列。非 FIFO 消费中，处理超过 ConsumeTimeout 后，客户端会停止延长消息的不可见时间，使其后续重新投递。若业务代码忽略取消令牌而继续执行，客户端无法强制终止它；因此 Handler 必须响应取消请求，并保证处理幂等。

需要自行控制接收批次、确认时机和不可见时间时，应选择上表中的 IGrpcSimpleConsumer；使用 LiteTopic 时，则选择 IGrpcLitePushConsumer 并完成相应的服务端资源准备。

classic Remoting 的消费模型

Remoting 提供的消费模型更多，应根据业务需要选择：

PushConsumer

Remoting PushConsumer 同样由客户端完成队列分配、拉取和回调分发，但 Handler 的入参是一批消息：

using EventHorizon.RocketMQ.Remoting.Consumer;
using EventHorizon.RocketMQ.Remoting.Consumer.Push;
using Microsoft.Extensions.DependencyInjection;
rocketMQ.AddRemotingPushConsumer<RemotingOrderCreatedHandler>(
    ServiceLifetime.Scoped,
    options =>
    {
        options.GroupName = "eventhorizon-blog-remoting-push-consumer";
        options.MaxConcurrency = 8; // 最多同时执行的 Handler 数。
        options.BatchSize = 32; // 单次从 Broker 拉取的最大消息数。
        options.ConsumeMessageBatchSize = 16; // 单次交给 Handler 的最大消息数。
        options.MaxDeliveryAttempts = 16;
        options.Subscribe("eventhorizon-test-topic", new FilterExpression("created"));
    });
public sealed class RemotingOrderCreatedHandler : IRemotingPushMessageHandler
{
    public ValueTask<ConsumeResult> HandleAsync(
        IReadOnlyList<RemotingMessageView> messages,
        RemotingPushConsumeContext context,
        CancellationToken cancellationToken)
    {
        foreach (var message in messages)
        {
            // 根据 message.Body 执行业务处理。
        }
        return ValueTask.FromResult(ConsumeResult.Success);
    }
}

BatchSize 控制单次长轮询向 Broker 请求的消息上限，ConsumeMessageBatchSize 控制单次回调交给 Handler 的消息上限。仅在并发、非 FIFO 批次中，Handler 返回 Success 前可将 context.AckIndex 设为最后一条已成功消息的从零开始索引，以确认连续前缀并重试余下消息；Retry 和 DeadLetter 始终作用于整个批次。带 MessageGroup 的 FIFO 消息和 ConsumeOrderly 都按单条串行处理；广播模式没有 Broker 的重试或死信流程。

无论使用哪种 Consumer，RocketMQ 的重试语义都要求业务具备幂等性。网络超时、Consumer 崩溃或确认结果丢失时，一条已处理消息仍可能再次到达。

OpenTelemetry：把消息链路接进现有观测体系

两个协议 Package 都内置了 OpenTelemetry Activity 和 Metrics，不需要额外安装独立的 instrumentation Package。应用负责选择 resource、采样器、processor 和 exporter：

builder.Services.AddOpenTelemetry()
    .WithTracing(tracing => tracing
        .AddAspNetCoreInstrumentation()
        // 应用自己创建的业务 span 也需要显式注册 source。
        .AddSource(GrpcConsumerMessageHandler.ActivitySourceName)
        .AddRocketMQGrpcInstrumentation()
        .AddRocketMQRemotingInstrumentation()
        .AddOtlpExporter())
    .WithMetrics(metrics => metrics
        .AddAspNetCoreInstrumentation()
        .AddRocketMQGrpcInstrumentation()
        .AddRocketMQRemotingInstrumentation()
        .AddOtlpExporter());

客户端会为 Producer 发送、非空接收、Push/LitePush 处理和 Consumer 确认等操作产生遥测数据。Producer 的 send Activity 会将当前 Trace Context 写入消息属性；消息经 Broker 到达 Consumer 后，PushConsumer 会读取这份上下文，并通过 Activity Link 建立关联。因此，发送端和消费端的遥测数据可以关联查看。

这里需要区分“上下文已经传到 Consumer”和“瀑布图一定是一棵严格的父子树”。PushConsumer 先记录本地的 receive，再以它为父节点创建 process；Producer 的 send 则作为 Activity Link 关联到消费侧。这样既保留了跨 Broker 的因果关系，也能正确处理长轮询和批量接收，避免把多个不同的生产端硬接成一个父节点。

Handler 执行时，当前 Activity 已经是 PushConsumer 创建的 process。业务代码只需创建自己的 Activity，不需要再解析或传递消息属性；它会自然挂在 process 下面。下面以库存预占为例，模拟处理耗时并创建业务 Activity：

using System.Diagnostics;
using EventHorizon.RocketMQ.Grpc.Consumer;
using EventHorizon.RocketMQ.Grpc.Consumer.Push;
internal sealed class GrpcConsumerMessageHandler : IGrpcPushMessageHandler
{
    internal const string ActivitySourceName = "EventHorizon.RocketMQ.Sample.Consumer";
    private static readonly ActivitySource ActivitySource = new(ActivitySourceName);
    public async ValueTask<ConsumeResult> HandleAsync(
        GrpcMessageView message,
        CancellationToken cancellationToken)
    {
        // 模拟库存预占或订单状态落库等业务处理。
        using var activity = ActivitySource.StartActivity("inventory.reserve", ActivityKind.Internal);
        await Task.Delay(TimeSpan.FromMilliseconds(650), cancellationToken);
        return ConsumeResult.Success;
    }
}

消费端本地的父子树仍然反映实际执行顺序：receive -> process -> inventory.reserve。Producer 的 send 不会被强行设为 process 的父节点，而是作为 process 的 Activity Link 关联起来；在 Grafana 中展开 process 的 References，即可看到对应的 View Linked Span。

Consumer Trace
receive
`-- process
    |-- inventory.reserve
    `- - Activity Link / View Linked Span - - -> Producer Trace: send

这条 Link 说明 Trace Context 已随消息跨越 Producer、Broker 到达 Consumer；同时，本地 Trace 树仍保留消息消费的异步与批量处理语义。

仓库提供了一个端到端的 OpenTelemetry 示例。其中，ProducerWebApi 和 ConsumerWebApi 分别运行在独立的 ASP.NET Core Minimal API Host 中。每个 Host 都注册 gRPC 与 Remoting 客户端；Producer 通过协议专用路由发送消息，Consumer 则使用彼此独立的 Consumer Group，避免两种协议共享消费位点。示例会向 OTLP/gRPC Collector 导出数据，并提供 Grafana Dashboard，用于查看发送量、发送时延、处理时延和 Producer/Consumer Trace。

下面是本地示例分别发送 gRPC 和 Remoting 消息后的 Producer dashboard：

Consumer dashboard 则展示接收、处理、确认和提交等消费侧操作：

在 Grafana Drilldown 中打开 Consumer 的 Trace，并展开 process 的 References 后，可以同时看到业务 inventory.reserve 子 span，以及跳转到 Producer send 的关联：

本地运行示例时，先按前面的步骤启动 RocketMQ 测试环境，再启动 Grafana OTEL LGTM：

docker compose -f test-environments/otel-lgtm/compose.yaml up -d --wait
ASPNETCORE_URLS=http://127.0.0.1:5242 \
  dotnet run --no-launch-profile --project samples/opentelemetry/ConsumerWebApi
ASPNETCORE_URLS=http://127.0.0.1:5241 \
  dotnet run --no-launch-profile --project samples/opentelemetry/ProducerWebApi

启动后，访问 http://127.0.0.1:5241/swagger，调用 /messages/grpc 和 /messages/remoting 即可分别发送消息；Consumer 状态位于 http://127.0.0.1:5242/consumers。Grafana 默认地址为 http://127.0.0.1:3000，可查看预置的 Producer 和 Consumer dashboard。

测试：不只验证“能发送一条消息”

实现 RocketMQ 客户端时，很多问题不会出现在简单的发送示例中。路由刷新、协议编码、并发回调、消费位点、异常响应、Broker 重连和服务端配置差异，都需要有针对性的验证。

项目将测试拆成两层：

1. 单元测试覆盖 gRPC、Remoting 和兼容性相关的客户端行为，不依赖网络或 Docker，适合快速、确定性地验证纯客户端逻辑。
2. 集成测试使用 Testcontainers 启动真实的 Apache RocketMQ 5.5.0 环境。gRPC 测试覆盖 Proxy、NameServer、Broker 的真实链路；Remoting 测试覆盖 NameServer 与 Broker 的直接通信路径。

集成测试既有单 Broker 基线，也有三个 Broker 的路由场景。gRPC 路径覆盖发送、消费、事务、Lite、死信和多 Broker 路由；Remoting 路径覆盖 Admin、Pull、发送、请求-响应、事务、撤回和多 Broker 路由。每个 fixture 使用隔离 Docker 网络和动态端口，不依赖开发机上的手工 Compose 环境。

CI 会采集单元测试和 Docker 集成测试的覆盖率并上传到 Codecov。高覆盖率不能替代生产环境验证，但它至少能让每次改动同时经过纯客户端行为和真实协议链路的检查。

使用前需要知道的边界

EventHorizon.RocketMQ 明确区分客户端能力、协议差异和服务端前提：

1. gRPC Package 只连接 RocketMQ Proxy，不直接连接 NameServer 或 Broker；它要求 RocketMQ 5 的 gRPC 服务端能力。
2. Remoting Package 需要应用能够访问 NameServer 发现到的 Broker 地址；容器网络与 Broker 对外注册地址必须正确。
3. 两个 Package 可以并存，但公开模型、异常、发送结果和 Consumer 语义不能混用。
4. Tag 过滤由客户端 API 支持；SQL 过滤、POP、Lite、优先级、延迟消息撤回等功能还依赖 Broker 或 Proxy 的配置与版本。
5. PushConsumer 不等于服务端主动推送：gRPC 的 PushConsumer / LitePushConsumer 与 classic Remoting 的 PushConsumer 都由客户端发起轮询或拉取。
6. 消费确认和重试提供的是消息系统语义，不会自动解决重复处理。事务检查、重试 Handler 和消费业务都需要幂等设计。
7. OpenTelemetry 包负责产生 Activity 与 Metrics；Collector、Exporter、采样、存储和告警策略仍由应用和平台负责。

总结

EventHorizon.RocketMQ 是面向 .NET 8 及更高版本的非官方 Apache RocketMQ 客户端，同时支持 RocketMQ 5 gRPC 与 classic Remoting。它以更贴合 .NET 应用习惯的方式提供 Producer、Consumer、Admin、DI、Generic Host、日志和 OpenTelemetry 支持。

选择路径可以简单理解为：

• 新项目已部署 RocketMQ 5 Proxy 时，应先评估官方 C# gRPC 客户端；若选择本项目的 gRPC 实现，则使用 EventHorizon.RocketMQ.Grpc。
• 需要对接 NameServer 和 Broker，或依赖 Pull、LitePull、POP、Admin、请求-响应等经典能力时，使用 EventHorizon.RocketMQ.Remoting。
• 同一应用有两类协议需求时，可以同时注册两个 Package，但应保持类型和客户端角色的协议边界。

项目地址：eventhorizon-cli/EventHorizon.RocketMQ

完整功能说明：仓库 README

可运行示例：samples

测试说明：本地与集成测试