# rocketmq-dotnet-client

**Repository Path**: ymjake/rocketmq-dotnet-client

## Basic Information

- **Project Name**: rocketmq-dotnet-client
- **Description**: No description available
- **Primary Language**: Unknown
- **License**: Apache-2.0
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-09-17
- **Last Updated**: 2026-11-03

## Categories & Tags

**Categories**: Uncategorized

**Tags**: Csharp, rocketmq

## README

# YMJake.RocketMQ

[![NuGet](https://img.shields.io/nuget/vpre/YMJake.RocketMQ.svg?label=NuGet)](https://www.nuget.org/packages/YMJake.RocketMQ)
[![License](https://img.shields.io/badge/license-Apache--2.0-blue.svg)](https://gitee.com/ymjake/rocketmq-dotnet-client/blob/master/LICENSE)

[English](https://gitee.com/ymjake/rocketmq-dotnet-client/blob/master/README.md) |
[简体中文](https://gitee.com/ymjake/rocketmq-dotnet-client/blob/master/README.zh-CN.md)

这是一个**非官方的 Apache RocketMQ .NET 客户端**，使用 **RocketMQ 5.x gRPC 代理协议**。

当前提供 `net8.0` 和 `net10.0` 目标，为 Producer、SimpleConsumer、PushConsumer 以及 Lite 主题消费者
提供强类型、基于 Task 的异步 API。协议中的客户端部分（设置注册、遥测流、队列分配、长轮询、回执
句柄、消息体校验、断线重连）。

> **预览版。** 当前包版本为 `0.1.0-preview.1`，`1.0` 之前公开 API 仍可能调整。

## 环境要求

- .NET 8 SDK 或更高版本；类库目标为 `net8.0` 和 `net10.0`。
- 启用了**代理（Proxy）**的 RocketMQ 5.x 部署（gRPC，默认端口 `8081`）。客户端连接的是代理接入点，
  不是 NameServer 或 Broker 地址。
- Lite 主题需要服务端开启 lite 能力；`LiteSimpleConsumer` 还需要 RocketMQ 5.5.1 及以上，更早的
  代理会拒绝 `LITE_SIMPLE_CONSUMER`。

## 安装

```shell
dotnet add package YMJake.RocketMQ --prerelease
```

包名与命名空间不同，代码中使用：

```csharp
using RocketMQ.Client;
```

## 功能与状态

`✅` 表示客户端已实现对应 API；实际使用时仍需满足相应的服务端能力与配置要求。

| 功能                     | 状态 | 说明                                                         |
| ------------------------ | :--: | ------------------------------------------------------------ |
| 普通消息生产             |  ✅   | `IProducer.SendAsync`。                                      |
| 批量消息生产             |  ✅   | 每个请求一个主题、只返回一个回执；不支持 FIFO 消息；部分失败时抛出异常。 |
| FIFO 消息生产            |  ✅   | 设置 `Message.MessageGroup`，主题需为 FIFO 主题。            |
| 定时/延迟消息生产        |  ✅   | 设置 `Message.DeliveryTimestamp`。                           |
| 优先级消息生产           |  ✅   | 设置 `Message.Priority`；不支持优先级的服务端会忽略。        |
| 事务消息                 |  ✅   | `BeginTransaction` / `ITransaction.Commit` / `Rollback`；`TransactionChecker` 负责裁决孤儿半消息。 |
| 定时消息撤回             |  ✅   | 在投递前用发送回执中的 `RecallHandle` 调用 `RecallAsync`。   |
| Lite 消息生产            |  ✅   | 设置 `Message.LiteTopic`；配额耗尽时抛出 `LiteTopicQuotaExceededException`。 |
| SimpleConsumer           |  ✅   | `ReceiveAsync`、`AckAsync`、`ChangeInvisibleDurationAsync`，以及运行时订阅/退订。 |
| 并发监听的 PushConsumer  |  ✅   | 后台长轮询；`WorkersPerQueue` 个监听任务并发消费同一条已分配队列。 |
| FIFO 监听的 PushConsumer |  ✅   | FIFO 消费组对单条队列只用一个 worker，保证同一 `MessageGroup` 的顺序。 |
| Push 重试与死信          |  ✅   | 监听失败按服务端重试策略的退避重新投递；超过 `MaxAttempts` 后转入死信队列。 |
| 运行时订阅               |  ✅   | Push 与 Simple 消费者均支持 `SubscribeAsync` / `Unsubscribe`。 |
| LitePushConsumer         |  ✅   | 运行时 `SubscribeLiteAsync` / `UnsubscribeLiteAsync`，首次订阅应用 `OffsetOption`，并同步代理推送的配额与长度限制。 |
| LiteSimpleConsumer       |  ✅   | Lite 订阅集合，加上显式的接收、ack 与修改不可见时长。        |
| Tag 与 SQL 过滤          |  ✅   | `FilterExpression` 支持 `ExpressionType.Tag` 和 `ExpressionType.Sql92`；SQL 过滤需服务端开启过滤能力。 |
| 访问凭证与请求签名       |  ✅   | 可插拔的 `ICredentialsProvider`、`StaticCredentialsProvider`，每个请求都做 ACL 签名。 |
| 消息体完整性             |  ✅   | 校验 CRC32/MD5/SHA1 摘要并解码 GZIP/ZLIB；失败时通过 `MessageView.IsCorrupted` 暴露。 |
| 遥测、设置同步与重连     |  ✅   | 遥测流、心跳、周期性路由与设置刷新、带退避的自动重连。       |
| 强类型异常               |  ✅   | 代理状态码映射为 `RocketMQ.Client.Exceptions` 中的异常类型。 |

以下内容**刻意不做**：classic Remoting 协议、Admin API、Microsoft DI / Generic Host 集成、
OpenTelemetry metrics 与 tracing。

## 快速开始

### 生产者

```csharp
using RocketMQ.Client;

var config = new ClientConfig { Endpoints = "localhost:8081" };

// 创建时预热主题路由，服务端缺少该主题会让启动直接失败；不填 Topics 则首次发送时才查路由。
var options = new ProducerOptions { MaxAttempts = 3 };
options.Topics.Add("TopicTest");

await using var producer = await Producer.Create(config, options);
var receipt = await producer.SendAsync(new Message("TopicTest", "hello"));
```

### 推送消费者

```csharp
using System.Text;
using RocketMQ.Client;

var config = new ClientConfig { Endpoints = "localhost:8081" };

var options = new PushConsumerOptions { ConsumerGroup = "my-consumer-group" };
options.SubscriptionExpressions["TopicTest"] = FilterExpression.All;

await using var consumer = await PushConsumer.Create(config, options, async (view, _) =>
{
    await Console.Out.WriteLineAsync(Encoding.UTF8.GetString(view.Body));
    return ConsumeResult.Success;
});
```

### 拉取消费者

```csharp
using RocketMQ.Client;

var config = new ClientConfig { Endpoints = "localhost:8081" };

var options = new SimpleConsumerOptions
{
    ConsumerGroup = "my-consumer-group",
    AwaitDuration = TimeSpan.FromSeconds(30),
};
options.SubscriptionExpressions["TopicTest"] = FilterExpression.All;

await using var consumer = await SimpleConsumer.Create(config, options);

while (true)
{
    var messages = await consumer.ReceiveAsync("TopicTest", 16, TimeSpan.FromSeconds(15));
    foreach (var message in messages)
    {
        await consumer.AckAsync(message);
    }
}
```

### Lite 消费者

```csharp
using System.Text;
using RocketMQ.Client;

var config = new ClientConfig { Endpoints = "localhost:8081" };

// 一个绑定主题承载所有 lite 主题；未显式订阅它时，客户端会自动以 "*" 订阅。
var options = new PushConsumerOptions { ConsumerGroup = "my-lite-group" };

await using var consumer = await LitePushConsumer.Create(config, "TopicTest", options, async (view, _) =>
{
    await Console.Out.WriteLineAsync($"{view.LiteTopic}: {Encoding.UTF8.GetString(view.Body)}");
    return ConsumeResult.Success;
});

await consumer.SubscribeLiteAsync("order-1", OffsetOption.Min);
```

发送时用 `new Message("TopicTest", "hello") { LiteTopic = "order-1" }`。

Lite 消费者要求服务端开启 lite 能力：绑定主题需接受 lite 消息（`message.type=LITE`），消费组需是
lite 组（`lite.bind.topic=<绑定主题>`）。

### 访问凭证

```csharp
var config = new ClientConfig
{
    Endpoints = "localhost:8081",
    CredentialsProvider = new StaticCredentialsProvider(accessKey, accessSecret),
};
```

## 投递语义

- **Push：** 返回 `ConsumeResult.Success` 即 ack；返回 `ConsumeResult.Failure` 或监听器抛出异常时消息
  不 ack，服务端按重试策略退避后重新投递。FIFO 消息在本地重新投递，保留回执句柄并递增
  `MessageView.DeliveryAttempt`；超过 `MaxAttempts` 后转入死信队列。
- **Simple：** 收到的消息在指定时长内不可见；到期前 ack，或用 `ChangeInvisibleDurationAsync` 延长
  不可见时长。
- **顺序：** FIFO 顺序由服务端消费组配置（`isConsumeMessageOrderly`）决定，客户端在设置握手时读取。

## 兼容性

gRPC 推送消费是客户端主动长轮询，不是协议级的 Broker 推送。SQL 过滤需要目标服务端开启过滤配置。
本项目为非官方项目，与 Apache Software Foundation 无隶属或背书关系。

## 许可证

[Apache License 2.0](https://gitee.com/ymjake/rocketmq-dotnet-client/blob/master/LICENSE)。