Apache RocketMQ 面向 AI 演进：LiteTopic 支撑百万级多 Agent 会话协作

过去一年，阿里云内部有两类 AI 系统在 RocketMQ 上完成了生产落地。一类是编程智能体 Qoder Cloud Agents，单个任务的执行周期可达数小时甚至数天；另一类是百炼平台的模型推理网关，需要在百万级租户规模下完成请求调度。两者的业务形态差异很大，但对消息系统提出的诉求是一致的：会话状态、任务交接和故障恢复需要由基础设施可靠承载，而不是由应用层反复实现。

本文说明 AI 应用给异步协作带来的新约束，RocketMQ 为此进行的一次面向 AI 场景的演进（LiteTopic），以及它在上述两个业务中解决的具体问题。

相比传统应用，AI 应用在通道拓扑、任务时长和成本结构三个维度都发生了根本变化。

通道拓扑的变化。传统应用的业务流在设计期即已确定，链路上的节点、消息的流向在系统上线后基本不变，因此消息通道的数量和拓扑可以预先规划。AI Agent 的执行路径由运行期决策产生，任务被自主拆解、动态派生，通道的数量与拓扑无法在设计期枚举。

任务时长的变化。传统异步消费的处理耗时在毫秒到百毫秒量级，AI 任务包含多轮推理和工具调用，单次执行常达分钟级，长会话可跨越数天，且耗时高度不可预测。同一任务内的交互轮次显著增加，中间状态需要被频繁读写，状态管理与会话管理由此成为必须在基础设施层解决的问题。

成本结构的变化。GPU 的单位算力成本远高于 CPU，因此在 AI 场景中，任何形式的算力空转都会直接反映在成本上。这一约束贯穿后文的多处设计选择。

在这些约束下，传统的事件驱动模型出现了三处不适配。

第一是积压斜率的变化。入口流量不变，但单条消息的处理耗时从毫秒级上升到分钟级，队列积压的增长速度完全不同，原先通过增加消费者即可吸收的流量抖动，现在需要成倍的算力才能消化。

第二是队头阻塞。传统模型中一个队列由一组消费者共享，其隐含前提是队列内的消息同质且可互换。在 AI 场景中，一条消息对应一个具体用户的一次具体会话，其耗时长是任务自身特征，并非系统异常。在共享队列模型下，该消息会阻塞其后所有任务，导致无关会话被动等待。缓解方式是细化队列粒度，而真正有效的粒度是会话级；传统 Topic 的创建与元数据成本决定了"一个会话一个 Topic"无法成立。

第三是重复推理的成本。单次推理的算力开销已经发生，若因进程重启、实例调度或网络异常导致中间结果丢失，则需重新计算。反之，若历史执行结果能够可靠持久化并在恢复后被读回，即可避免重复调用模型。消息与事件的持久化能力在此不仅关系可靠性，也直接关系成本。

由此可以归纳出消息系统需要补齐的三项能力：支撑海量、动态生成且彼此独立的高速通道；将隔离粒度从 Topic 级下沉到会话级，使每个会话具备独立的顺序和故障边界；识别消息与会话的归属关系，从而提供会话级的有序性与亲和性。这三项能力共同指向一个新的 Topic 模型。

RocketMQ LiteTopic 的核心思路是将每个 AI 会话映射为一个独立的轻量级 Topic，而非让所有会话共享同一个业务 Topic。

将粒度切至会话级会引入两个必须先行解决的工程问题：百万级通道同时存在时，消费队列索引如何组织；百万级订阅关系下，如何确定当前哪些通道有消息可投。LiteTopic 对应做了两处改造，存储层以 RocksDB KV 引擎替代基于文件的消费队列索引，投递层以事件驱动替代长轮询。

需要说明的是，这不是对 RocketMQ 存储内核的重写。顺序写、零拷贝、主从复制、刷盘策略等经过大规模生产验证的能力保持不变，LiteTopic 只在其上增加一层索引以及一套新的订阅与投递语义，这决定了它的成熟度和接入成本。

LiteTopic 挂载在 Parent Topic 之下，构成两层结构。

Parent Topic 承担命名空间与控制边界的职责，需要预先创建，权限、路由、配额等管控能力挂载在这一层。其数量很少，通常一个业务域或一个 Worker 池对应一个即可。

LiteTopic 是运行期的动态单位，具备有序、隔离、可重放三个属性。它无需预创建，第一条消息到达时自动创建，空闲后按 TTL 自动回收。业务侧需要做的仅是在发送消息时标明该消息所属的会话。

在这一模型下，逻辑通道的创建与销毁成本降至可以按会话使用的程度，通道生命周期与会话生命周期对齐，运维侧无需为通道的增减进行任何操作。

Topic 模型的变化要求消费模型同步调整。RocketMQ 原有的集群消费以 Group 为单位组织，同组内所有实例订阅完全相同的一组 Topic，并在组内做负载均衡。该模型适合同质任务，但无法表达实例之间的差异，也就无法指定某条消息由某个实例处理。

LiteTopic 将订阅粒度细化为同时支持 Group 级与单个 Client 级，消息可以被路由到特定实例，并保证一个 LiteTopic 在任一时刻仅被一个 Client 独占消费。由此，同一会话的消息不会被拆分到多个实例并发处理，避免了顺序错乱与状态竞争。同时，同一消费组内的不同实例可以订阅不同的 LiteTopic 子集，使负载能够按会话维度摊开。

RocketMQ 的存储是分层的，消息体统一顺序追加至 CommitLog，ConsumeQueue 仅是指向物理位置的轻量索引。LiteTopic 沿用该结构，改动集中在索引层：后台 dispatch 线程并行构建两套索引，一套是按 Parent Topic 组织的标准 ConsumeQueue，一套是按 LiteTopic 组织的 LiteCQ。消息本身仍只写入一份，写放大受控，消费侧可按 LiteTopic 精确拉取所属通道的消息。

原有的 ConsumeQueue 采用文件实现，一个队列对应一个独立文件。在数十至数百个 Topic 的规模下不存在问题，但百万级 LiteTopic 意味着百万级小文件，文件句柄、内存映射与元数据管理的开销将非线性增长，写入也会退化为大量随机 I/O。

改造后，所有 LiteCQ 索引统一收敛到一个 KV 引擎中，Key 由 Topic、QueueID、Offset 组成，Value 为消息在 CommitLog 中的物理偏移量。百万级 LiteTopic 共享同一套 KV 存储，单条索引查询为 O(1)，元数据规模不再随通道数量膨胀。RocksDB 的 LSM 结构适合高频写入与海量小记录，与该场景的访问特征一致。从“每队列一个文件”转为“共享 KV、O(1) 元数据”，是百万通道共存的前提条件。

存储层解决了通道的容纳问题，投递层需要解决活跃通道的定位问题。长轮询的工作方式是消费者周期性发起请求，Broker 扫描相关队列。当队列数量达到百万级，周期性全量扫描的 CPU 开销显著上升，而其中大部分扫描无效，因为同一时刻真正有消息的通道只占很小比例。

LiteTopic 引入了两个结构：Lite Subscription Set 保存订阅关系，Ready Set 聚合当前确实有消息可投的通道。消息写入时触发事件匹配，命中订阅关系的消息被聚合进 Ready Set，消费者仅从 Ready Set 获取任务。这一改造与 epoll 相对 select 的改进思路一致，调度成本由 O(订阅总数) 降为 O(活跃数)：存量会话可以是百万级，而单位时间内活跃的通道仅为千级，系统只需为活跃部分付出调度开销。

上述机制在业务侧表现为四项能力，分别对应第一节提出的问题。

• 按需创建：第一条消息到达时通道自动创建，无需预先申请或在配置中登记，解决海量通道的资源供给问题。

• 自动过期：空闲通道按 TTL 自动回收，无需业务显式销毁。按需创建必须配套自动回收，否则历史会话的残留元数据会持续累积。

• 差异化订阅：消息可精确投递至单个 Client，同一消费组内的实例可订阅不同的 LiteTopic 子集，实现会话级隔离。

• 消费挂起：消费者返回 Suspend(N) 时，仅当前 LiteTopic 暂停 N 毫秒，其余通道不受影响。这一能力用于替代消费线程内的 sleep 限速：sleep 期间消费线程被占用，其本可处理的其他会话只能等待；Suspend 由 Broker 记录该通道下次可投递的时间，消费线程立即返回线程池继续服务其他会话。在 GPU 成本占主导的场景中，这一差异直接影响资源利用率。

生产端在构造消息时增加一个会话维度的标识：

// 生产：按会话维度写入，通道由 Broker 自动创建Message msg = provider.newMessageBuilder()        .setTopic(PARENT_TOPIC)        .setLiteTopic("session-" + sessionId)        .setBody(payload)        .build();producer.send(msg);

消费端订阅 Parent Topic 下的全部通道，需要限速时返回 Suspend：

// 消费：订阅父 Topic 下所有通道，按通道级别调速consumer.subscribe(PARENT_TOPIC, "*");// 仅挂起当前通道，不阻塞消费线程return ConsumeResult.suspend(Duration.ofMillis(50));

以上为示意代码，实际 API 以所用版本为准。原有的编程模型、客户端依赖与运维体系均无需更换，改造集中在消息的会话维度标识和限速方式两处。

本节以 Qoder Cloud Agents 为主，它较为完整地使用了上述能力；百炼网关代表另一类问题，放在后半部分。

核心难点：脑手分离架构下的有序交接与故障恢复

Qoder Cloud Agents 是运行在云上的编程智能体服务，接收需求后自主完成代码读取、修改、测试与迭代修正，单个任务的执行周期较长。其架构为脑手分离，Agent Runtime 负责决策，Sandbox Worker 池负责执行，整体事件驱动。会话状态不绑定在具体实例上，因此可以做到无损伸缩，并支撑万级规模的并发推理实例。

需要澄清的是，分布式架构本身不会缩短单次推理的耗时，其收益体现在资源利用率、并发能力与成本效率上。该系统的主要难点在于执行状态的可恢复性、任务交接的有序性、Agent 循环中复杂语义的表达，以及故障恢复与版本演进，这些均属于消息与事件层面的问题。

基于 LiteTopic 的实现：请求/响应双通道，每个会话一条 lane

Agent Runtime 与 Sandbox Worker 之间使用两个 Parent Topic 完成双向异步通信，一个承载 request，一个承载 response。Worker 仅订阅 request 方向，Runtime 仅订阅 response 方向，两端均不阻塞在对方的处理过程上。每个会话在两个 Parent Topic 下各拥有一条 lane，即以会话标识命名的 LiteTopic，随首条消息创建，会话结束后按 TTL 回收。

流经消息系统的是任务指令、执行结果与状态变迁等事实与事件，不包含庞大的中间状态。执行进度由 Checkpoint 承载并持久化至存储，大对象与产物同样存放在存储侧，消息中仅保留引用。这里的边界需要明确：消息系统负责有序、可靠地传递事件，而任务当前进展的权威来源是 Checkpoint，不通过消息的 ACK 时机推断。因此实例重启、会话迁移与事件重放都不会导致进度失真。

在这一结构下，会话内消息有序而会话间并行执行，顺序性与并发度可以同时满足。流量突增时由 Broker 承接积压，Worker 按自身处理能力拉取，形成的是通道级的局部背压，单个会话变慢不会传导为全局过载，故障影响也被约束在单条通道内。隔离层次上，Parent Topic 隔离不同的 Worker 池，LiteTopic 隔离业务单元，脑与手可以各自独立伸缩。

四类复杂场景：并行、打断、Webhook、挂起恢复

1. 多 Agent 并行

一个会话内会派生多条执行线，任务以 LiteTopic = session-id 有序下发，Worker 抢占后并行执行，结果由调度侧通过 Mailbox 与 Barrier 聚合，从而同时获得分发的确定性、执行的并发度与汇合语义。

2. 打断处理

Agent 执行期间收到用户的追加指令时，新消息先进入该 Agent 的 Mailbox，在执行完成前的下一个安全边界注入；需要中止正在执行的任务时，通过独立的控制通道下发停止指令，随后经 LiteTopic 恢复。数据面与控制面在此必须分开。

3. Webhook 回调

以 LiteTopic = endpoint-id 组织，保证单个 endpoint 的回调有序、不同 endpoint 之间并行，某个 endpoint 变慢或故障时仅阻塞其自身通道。

4. 挂起与恢复

Agent 在等待用户输入、工具返回或子 Agent 完成时，将待处理状态持久化并释放 Worker，等待期间不占用算力；事件返回后，后续任务段经 LiteTopic 下发，由任意空闲 Worker 接手继续执行。传统架构中会话与 Worker 固定绑定，会带来连接中断导致上下文丢失、等待期间实例空转、扩容时会话迁移困难三个瓶颈，改为按需接入后这三点均得到解决。

上述四类场景在同一套 LiteTopic 架构下统一表达，无需为每类场景引入独立机制。

核心难点：限流问题的性质变化

传统网关限流判断调用方的请求速率是否超出配额，超出即拒绝，其前提是后端容量可弹性扩张，限流仅用于防止极端流量打穿系统。模型推理服务不满足该前提，GPU 是当前最稀缺的生产要素，后端处理能力存在硬上限。限流因此从放行与拒绝的二元判断，变为还需确定以何种节奏将请求送往后端。

在百万级租户规模下，单纯的阈值限流会同时放大两类误差。一类是误拒，请求实际可被后端接纳，但瞬时速率超过阈值而被直接拒绝，调用方收到 503，此时后端并未真正饱和；另一类是过载放行，多个维度的流量在同一时刻叠加，各维度均未超过阈值，合并后压垮后端。其原因在于传统限流只有放行与拒绝两种状态，缺少先接纳请求、再按合适节奏送往后端的第三种状态。

基于 LiteTopic 的实现：漏桶阵列按 user + model 精准调速

重构后，网关自身仅保留一层固定窗口的硬上限用于兜底极端情况，请求按 user + model 维度调用 setLiteTopic 写入消息系统，实际调速由 LiteTopic 构成的漏桶阵列承担。漏桶按需创建、按 TTL 回收，无需预先为租户与模型的组合规划资源。消费侧使用统一消费组以 subscribe(parent, "*") 匀速消费，需要调速时返回 Suspend(N ms)，精确控制单个 user + model 组合的输出速率而不影响其他组合。

效果上，限流维度从单一维度扩展到数千个维度，限流比降低至原先的十分之一左右，即在相同后端算力下可接纳的有效请求明显增加，被误拒的请求明显减少；调速精度达到毫秒级，可跟随后端实时水位动态调整输出速率。该方案目前运行于公有云百炼网关，服务百万级租户规模。

两个案例的问题形态不同，一个是有序交接，一个是算力分配，但依赖的是同一组能力：会话粒度的通道、按需创建与自动回收、以及不阻塞消费线程的调速。

社区方面，LiteTopic 已合并进 Apache RocketMQ 的 master 分支，成为主干能力，RocketMQ-A2A 的实现也已开放给社区，目前正在探索 MCP 传输层的实现，将该传输能力延伸至 Agent 与工具之间的通信场景。相关工作获得中国信通院 2025 年度 OSCAR 开源 + AI 尖峰案例奖与 InfoQ AI 开源明星项目奖。

论文主张：把多 Agent 协作视为执行基础设施

相关设计整理成论文《RocketMQ-A2A: Reliable Session-Level Replayable Event Streams for Large-Scale Multi-Agent Collaboration》，已被 FSE Companion 2026 接收，会议于 2026 年 7 月在加拿大蒙特利尔举行。论文的基本主张是将多 Agent 协作视为执行基础设施，其职责不止于消息送达，还包括故障后的恢复能力与事后的可审计性。

论文设计：双平面架构与可靠性语义

在实现上，论文将两类通信划分为两个平面：任务下发走普通 Topic，面向吞吐，突发流量转化为 MQ 积压，Worker 按自身容量拉取；结果回传与审计走 Lite-Topic，面向会话，每个会话拥有一条有序、持久、可重放的结果流。Supervisor 由此不再阻塞等待长任务返回，而是依据事件推进会话状态。

可靠性语义方面，任务面上 Worker 仅在终态事件成功写入 Lite-Topic 后才对任务进行 ACK；回传与审计面上 Supervisor 按 (sid, tid, seq) 去重，并支持从指定 offset 或 seq 重放。系统提供至少一次投递与可重放语义，即已被 Broker 确认的任务和事件不会被静默丢失，但重复投递可能发生，带副作用的操作仍需业务侧通过幂等键保护。

论文数据：1500 万会话、50k TPS 下延迟仍平稳

论文给出的压测数据如下，5 个 Broker 承载 250 万会话时，12.5k TPS 下平均延迟约 21 毫秒、P95 约 47 毫秒，25k TPS 下平均约 29 毫秒、P95 约 61 毫秒。会话数升至 600 万后，5 Broker 在 25k TPS 下 CPU 使用率约 84%，接近该配置的压力边界。扩容至 10 个 Broker 后，600 万会话、50k TPS 下平均延迟约 26 毫秒、P95 约 81 毫秒，CPU 约 47%；会话数进一步提升至 1500 万并维持 50k TPS，平均延迟约 26 毫秒、P95 约 71 毫秒。在 10 Broker、50k TPS 的固定条件下，存量会话数扩大 2.5 倍而延迟基本持平，代价主要体现为 CPU 使用率上升，即控制面开销主要由活跃会话数决定，与存量会话数弱相关。20 万量级的 Lite-Topic 并发同样保持稳定。

LiteTopic 的作用可以概括为两点：将消息通道的成本降低到可以按会话分配的程度，并将调度成本由与通道总数相关改为与活跃通道相关，使百万级通道共存在工程上可行。有序性、隔离性、可重放与非阻塞调速等能力建立在此基础之上。

对于正在将 Agent 系统推向生产的团队，会话状态、任务交接与故障恢复更适合由基础设施承担，而不是在应用层反复实现。LiteTopic 已在 Apache RocketMQ 主干上，可直接使用，也欢迎在社区中讨论后续的演进方向。

相关链接：

[1] 《百炼网关实践：用 RocketMQ LiteTopic 让限流比降了 10 倍》

[2] AI 异步通信与数据集成一站式方案

https://www.aliyun.com/activity/middleware/ai-mq

[3] ACM Digital Library 完整论文（RocketMQ-A2A）下载

https://dl.acm.org/doi/10.1145/3803437.3805231

[4] RocketMQ-A2A 开源仓库（含可运行示例）

https://github.com/apache/rocketmq-a2a

LiteTopic 解决的是 Agent 之间怎么协作，Agent 消费的数据能不能实时可信，是另一个上游问题。会话粒度的异步协作和实时的数据供给，是消息基础设施服务 AI 的一体两面。

8 月 28 日下午，InfoQ 携手阿里云与 IBM 在上海徐汇滨江举办一场线下沙龙，专门聊怎么让实时数据以低延迟、高可靠、可弹性的方式，持续供给模型与 Agent。

如果你正在把 AI 应用、Agent 或 RAG 推向生产，欢迎扫码报名。