NewLife.RocketMQ 3.0.2026.216-beta1437

NewLife.RocketMQ - 企业级纯托管 RocketMQ 客户端

纯托管企业级 RocketMQ 客户端，支持 .NET Framework 4.5+ / .NET Standard 2.0+ / .NET Core / .NET 5+。

完全使用 C# 实现，零外部依赖（无需 Java、gRPC、Protobuf 第三方库）

🎯 产品定位

NewLife.RocketMQ 是新生命团队开发的企业级纯托管 RocketMQ 客户端，专为 .NET 生态设计：

• ✅ 双协议支持：同时支持 RocketMQ Remoting 协议（4.x）和 gRPC Proxy 协议（5.x）
• ✅ 零外部依赖：纯 C# 实现，无需 Java 或 gRPC 运行时，内置轻量级 Protobuf 编解码器
• ✅ 多云适配：统一接口适配阿里云、华为云、腾讯云及 Apache ACL 认证体系
• ✅ 生产就绪：完整的消费重试、死信队列、事务回查、顺序消费等企业级特性
• ✅ 高性能：VIP 通道、连接池、消息压缩、并发控制等性能优化手段

支持 Apache RocketMQ 4.0+ 和 5.x 版本

🚀 核心特性

生产者（Producer）

• ✅ 同步/异步/单向发送
• ✅ 批量消息发送
• ✅ 延迟消息（18级定时 + 任意时间延迟）
• ✅ 事务消息（半消息 + 事务回查）
• ✅ 顺序消息
• ✅ Request-Reply 模式
• ✅ 消息压缩（ZLIB）
• ✅ 消息轨迹追踪

消费者（Consumer）

• ✅ Pull 模式消费
• ✅ 集群消费 / 广播消费
• ✅ 消费者负载均衡（Rebalance）
• ✅ Tag 过滤 / SQL92 表达式过滤
• ✅ 多 Topic 订阅
• ✅ 消费重试 + 死信队列（DLQ）
• ✅ 顺序消费（队列锁定）
• ✅ Pop 消费模式
• ✅ 消费限流（并发控制）
• ✅ 按时间戳消费

管理功能

• ✅ Topic 创建/更新/删除
• ✅ 消费组创建/更新/删除
• ✅ 消息查询（按 ID / 按 Key）
• ✅ 消费统计查询
• ✅ 集群信息查询
• ✅ 偏移量管理与重置

云厂商适配

• ✅ 阿里云消息队列 RocketMQ（实例 ID 路由 + HTTP NameServer 发现）
• ✅ 华为云 DMS for RocketMQ（SSL/TLS + 实例 ID 路由）
• ✅ 腾讯云 TDMQ RocketMQ（Namespace 前缀路由）
• ✅ Apache RocketMQ ACL 认证（HMAC-SHA1 签名）

协议支持

• ✅ Remoting 协议（RocketMQ 4.x/5.x Broker）：成熟稳定，功能完整
• ✅ gRPC Proxy 协议（RocketMQ 5.x Proxy）：内置 Protobuf 编解码器，支持任意时间延迟消息

📦 安装

NuGet 包管理器

Install-Package NewLife.RocketMQ

.NET CLI

dotnet add package NewLife.RocketMQ

PackageReference

<PackageReference Include="NewLife.RocketMQ" Version="3.0.*" />

🔧 快速开始

1. 生产者发送消息

using NewLife.RocketMQ;

var producer = new Producer
{
    Topic = "test_topic",
    NameServerAddress = "127.0.0.1:9876",
    Group = "producer_group"
};

producer.Start();

// 同步发送
var result = producer.Publish("Hello RocketMQ!");
Console.WriteLine($"消息ID: {result.MsgId}");

// 异步发送
await producer.PublishAsync("异步消息");

// 批量发送
await producer.PublishBatch(new[] { "消息1", "消息2", "消息3" });

2. 消费者接收消息

using NewLife.RocketMQ;

var consumer = new Consumer
{
    Topic = "test_topic",
    Group = "consumer_group",
    NameServerAddress = "127.0.0.1:9876"
};

consumer.OnConsume = (q, messages) =>
{
    foreach (var msg in messages)
    {
        Console.WriteLine($"收到消息: {msg.BodyString}");
    }
    return true; // 返回 true 表示消费成功
};

consumer.Start();
Console.WriteLine("按任意键退出...");
Console.ReadKey();

3. 延迟消息

// 使用预设延迟等级（18级）
producer.PublishDelay("延迟消息", DelayTimeLevels.s30);

// gRPC 模式支持任意时间延迟
producer.GrpcProxyAddress = "http://127.0.0.1:8081";
await producer.PublishDelayViaGrpcAsync("任意延迟", DateTime.Now.AddMinutes(30));

4. 事务消息

var producer = new Producer
{
    Topic = "tx_topic",
    Group = "tx_group",
    NameServerAddress = "127.0.0.1:9876"
};

// 设置事务回查回调
producer.OnCheckTransaction = (msg, transactionId) =>
{
    // 根据 transactionId 查询本地事务状态
    var success = CheckLocalTransaction(transactionId);
    return success ? TransactionState.Commit : TransactionState.Rollback;
};

producer.Start();

// 发送半消息
var sendResult = producer.PublishTransaction("订单创建");

try
{
    // 执行本地事务
    ExecuteLocalTransaction(sendResult.TransactionId);
    
    // 提交事务
    producer.EndTransaction(sendResult, TransactionState.Commit);
}
catch
{
    // 回滚事务
    producer.EndTransaction(sendResult, TransactionState.Rollback);
}

5. Request-Reply 模式

生产者端：

var producer = new Producer
{
    Topic = "request_topic",
    NameServerAddress = "127.0.0.1:9876"
};
producer.Start();

// 同步请求（等待响应）
var response = producer.Request("请求消息", timeout: 5000);
Console.WriteLine($"收到响应: {response.BodyString}");

// 异步请求
var reply = await producer.RequestAsync("异步请求", timeout: 5000);

消费者端：

var consumer = new Consumer
{
    Topic = "request_topic",
    Group = "request_group",
    NameServerAddress = "127.0.0.1:9876"
};

consumer.OnConsume = (q, messages) =>
{
    foreach (var msg in messages)
    {
        if (!String.IsNullOrEmpty(msg.CorrelationId))
        {
            // 处理请求并发送响应
            var result = ProcessRequest(msg.BodyString);
            consumer.SendReply(msg, result);
        }
    }
    return true;
};

consumer.Start();

6. 顺序消息

// 发送顺序消息（相同 key 的消息进入同一队列）
var queue = producer.SelectQueue("order_123");
producer.Publish("顺序消息1", queue);
producer.Publish("顺序消息2", queue);

// 消费端启用顺序消费
var consumer = new Consumer
{
    Topic = "test_topic",
    Group = "consumer_group",
    NameServerAddress = "127.0.0.1:9876",
    OrderConsume = true  // 启用顺序消费
};

🌩️ 云厂商接入

阿里云消息队列 RocketMQ

using NewLife.RocketMQ;

var producer = new Producer
{
    Topic = "test_topic",
    NameServerAddress = "http://MQ_INST_xxx.aliyuncs.com:80",
    CloudProvider = new AliyunProvider
    {
        AccessKey = "你的AccessKey",
        SecretKey = "你的SecretKey",
        InstanceId = "MQ_INST_xxx"  // 可选，自动从地址解析
    }
};
producer.Start();

华为云 DMS for RocketMQ

var producer = new Producer
{
    Topic = "test_topic",
    NameServerAddress = "华为云实例地址:9876",
    CloudProvider = new HuaweiProvider
    {
        AccessKey = "你的AK",
        SecretKey = "你的SK",
        InstanceId = "实例ID",
        EnableSsl = true  // 启用SSL加密
    }
};
producer.Start();

腾讯云 TDMQ RocketMQ

var producer = new Producer
{
    Topic = "test_topic",
    NameServerAddress = "腾讯云实例地址:9876",
    CloudProvider = new TencentProvider
    {
        AccessKey = "腾讯云SecretId",
        SecretKey = "腾讯云SecretKey",
        Namespace = "命名空间"
    }
};
producer.Start();

Apache RocketMQ ACL 认证

var producer = new Producer
{
    Topic = "test_topic",
    NameServerAddress = "127.0.0.1:9876",
    CloudProvider = new AclProvider
    {
        AccessKey = "RocketMQ AccessKey",
        SecretKey = "RocketMQ SecretKey"
    }
};
producer.Start();

🎓 进阶使用

消费重试与死信队列

var consumer = new Consumer
{
    Topic = "test_topic",
    Group = "consumer_group",
    NameServerAddress = "127.0.0.1:9876",
    EnableRetry = true,           // 启用消费重试
    MaxReconsumeTimes = 3         // 最大重试次数
};

consumer.OnConsume = (q, messages) =>
{
    foreach (var msg in messages)
    {
        try
        {
            ProcessMessage(msg);
        }
        catch
        {
            // 返回 false 触发重试
            return false;
        }
    }
    return true;
};

// 超过最大重试次数后，消息自动进入死信队列：%DLQ%{ConsumerGroup}

Tag 过滤

// 生产者发送带 Tag 的消息
producer.Publish("消息内容", tag: "TagA");

// 消费者订阅指定 Tag
var consumer = new Consumer
{
    Topic = "test_topic",
    Group = "consumer_group",
    Tags = "TagA || TagB",  // 订阅 TagA 或 TagB
    NameServerAddress = "127.0.0.1:9876"
};

SQL92 表达式过滤

// 生产者发送带自定义属性的消息
var msg = new Message
{
    BodyString = "消息内容",
    ["age"] = "25",
    ["city"] = "Shanghai"
};
producer.Publish(msg);

// 消费者使用 SQL92 表达式过滤
var consumer = new Consumer
{
    Topic = "test_topic",
    Group = "consumer_group",
    ExpressionType = "SQL92",
    Subscription = "age > 18 AND city = 'Shanghai'",
    NameServerAddress = "127.0.0.1:9876"
};

多 Topic 订阅

var consumer = new Consumer
{
    Topics = "topic1;topic2;topic3",  // 订阅多个 Topic
    Group = "consumer_group",
    NameServerAddress = "127.0.0.1:9876"
};

consumer.OnConsume = (q, messages) =>
{
    foreach (var msg in messages)
    {
        Console.WriteLine($"Topic: {msg.Topic}, 消息: {msg.BodyString}");
    }
    return true;
};

Pop 消费模式

var consumer = new Consumer
{
    Topic = "test_topic",
    Group = "consumer_group",
    NameServerAddress = "127.0.0.1:9876"
};

consumer.Start();

// Pop 消费（不自动提交偏移）
var messages = await consumer.PopMessageAsync(timeout: 10000);
foreach (var msg in messages)
{
    try
    {
        ProcessMessage(msg);
        // 手动确认
        await consumer.AckMessageAsync(msg);
    }
    catch
    {
        // 修改不可见时间（延长处理时间）
        await consumer.ChangeInvisibleTimeAsync(msg, 30000);
    }
}

VIP 通道

var producer = new Producer
{
    Topic = "test_topic",
    NameServerAddress = "127.0.0.1:9876",
    VipChannelEnabled = true  // 启用 VIP 通道（使用 BrokerPort - 2）
};

消息压缩

var producer = new Producer
{
    Topic = "test_topic",
    NameServerAddress = "127.0.0.1:9876",
    CompressOverBytes = 4096  // 消息体超过 4KB 自动 ZLIB 压缩
};

消费限流

var consumer = new Consumer
{
    Topic = "test_topic",
    Group = "consumer_group",
    NameServerAddress = "127.0.0.1:9876",
    MaxConcurrentConsume = 10  // 最多同时处理 10 条消息
};

📚 文档

• 架构设计文档
• 功能分析与兼容性报告
• Request-Reply 使用指南
• 在线教程

🏗️ 架构特点

双协议架构

MqBase (业务基类)
├── Producer (生产者)
└── Consumer (消费者)

通信层
├── Remoting 协议（4.x/5.x Broker）
│   ├── ClusterClient (TCP 长连接)
│   ├── NameClient (路由发现)
│   └── BrokerClient (心跳/注销)
│
└── gRPC 协议（5.x Proxy，netstandard2.1+）
    ├── GrpcClient (HTTP/2 客户端)
    ├── GrpcMessagingService (11 个 RPC 方法)
    └── ProtoWriter/ProtoReader (自研 Protobuf 编解码器)

云厂商适配层

ICloudProvider (统一云厂商接口)
├── AliyunProvider (阿里云)
├── HuaweiProvider (华为云)
├── TencentProvider (腾讯云)
└── AclProvider (Apache ACL)

🔬 测试覆盖

30+ 测试类覆盖核心功能：

• ✅ 核心功能：ProducerTests、ConsumerTests、CommandTests、MessageTests
• ✅ 高级特性：TransactionCheckTests、BatchMessageTests、RetryTests、OrderConsumeTests
• ✅ 协议兼容：IPv6Tests、MessageId5xTests、MQVersionTests、ProtoTests
• ✅ 云厂商：AliyunTests、CloudProviderTests
• ✅ 性能优化：CompressionTests、ConcurrentConsumeTests、VipChannelTests

📊 性能优势

🤝 参与贡献

欢迎提交 Issue 和 Pull Request！

1. Fork 本仓库
2. 创建特性分支 (git checkout -b feature/AmazingFeature)
3. 提交更改 (git commit -m 'Add some AmazingFeature')
4. 推送到分支 (git push origin feature/AmazingFeature)
5. 提交 Pull Request

📄 许可协议

本项目采用 MIT License 开源协议。

新生命项目矩阵

各项目默认支持 net10.0/net9.0/netstandard2.1/netstandard2.0/net4.62/net4.5

新生命开发团队

新生命团队（NewLife）成立于2002年，是新时代物联网行业解决方案提供者，致力于提供软硬件应用方案咨询、系统架构规划与开发服务。

团队主导的80多个开源项目已被广泛应用于各行业，Nuget累计下载量高达400余万次。

团队开发的大数据中间件 NewLife.XCode、蚂蚁调度计算平台 AntJob、星尘分布式平台 Stardust、缓存队列组件 NewLife.Redis 以及物联网平台 FIoT，均成功应用于电力、高校、互联网、电信、交通、物流、工控、医疗、文博等行业，为客户提供了大量先进、可靠、安全、高质量、易扩展的产品和系统集成服务。

我们将不断通过服务的持续改进，成为客户长期信赖的合作伙伴，通过不断的创新和发展，成为国内优秀的 IoT 服务供应商。

新生命团队始于2002年，部分开源项目具有20年以上漫长历史，源码库保留有2010年以来所有修改记录

• 网站：https://newlifex.com
• 开源：https://github.com/newlifex
• QQ群：1600800 / 1600838
• 微信公众号：