1. 从一次 Agent 集群重启说起为什么传统 Topic 扛不住 AI 会话先说个我踩过的坑。去年做一个多智能体协作的工单分析系统Master Agent 负责拆任务十几个 Worker Agent 并行调模型处理。上线第一周就出事了某个 Worker 节点 OOM 重启结果它手上那批会话的消费进度全丢了用户看到的是「分析到一半突然重来」。更麻烦的是为了给每个用户会话留独立通道我们提前建了几千个 TopicBroker 的元数据压力肉眼可见地涨Consumer Group 一重平衡整个 Agent 集群跟着抖。这不是我们架构写得烂而是传统消息模型和 AI 场景的错配。传统 MQ 的设计目标是「一条流水线多人抢活干」——订单、支付、日志采集这类场景Topic 数量可控消费者组负载均衡是优点。但 AI 应用长这样一个用户就是一个独立会话百万级会话同时在跑Agent 之间异步协作任务派发和结果回传是两条线Worker 挂了重启希望从断点续传而不是从头再来消息来了才唤醒不想靠全量轮询把 CPU 烧光。Apache RocketMQ 5.5.0 给出的答案就是 LiteTopic。它随开源版本正式落地核心是把「会话」当成一等公民一个 Agent 或一个用户会话对应一条 LiteTopic按需动态创建消费位点持久化在 Broker 侧事件驱动投递。这篇文章我会带你从零跑通一条完整链路——创建 LiteTopic、配置 AI 消费端、验证消息触发智能体响应全部给可复制的片段。适合正在做 Multi-Agent 系统、AI 工作流编排或者被长会话状态管理折磨的消息中间件开发者。先建立整体印象Master Agent 发任务到父 Topic 下的某条 LiteTopicWorker Agent 订阅并处理调完 LLM 后把结果异步回传到另一条 LiteTopicMaster 再接收。RocketMQ 在中间当「消息高速公路」而每条会话通道是独立的、可断点续传的、有 TTL 自动回收的。2. LiteTopic 前置准备两层结构、RocksDB 索引与事件驱动到底解决了什么在动手之前得先搞清楚 LiteTopic 的模型不然配置项会看得云里雾里。LiteTopic 用的是两层结构。父 Topic 当命名空间用比如AGENT_TASK_NS同一业务域下的所有 LiteTopic 都挂在它下面LiteTopic 才是真正的会话通道比如TASK_agent_001、RESULT_session_abc。这个设计的好处是你不需要为每个会话提前建 Topic父 Topic 建一次LiteTopic 首次订阅时自动生效。底层索引层换成了 RocksDB消费位点以 KV 方式存储配合 Ready Set 事件队列只有写入或可读事件触发时才唤醒对应的 LiteTopic避免百万通道全量扫描。对比一下传统 Topic 和 LiteTopic 的关键差异你就能明白为什么 AI 场景要换模型维度传统 TopicLiteTopic创建方式通常需要提前创建按需动态创建首次订阅自动生效会话规模Topic 多了 Broker 扛不住单 Broker 可承载百万级轻量通道消费进度Consumer Group 共享更接近单连接语义减少重平衡唤醒机制长轮询 全量扫描事件驱动有消息才唤醒资源回收需手动清理支持 TTL 自动回收过期会话一句话传统 Topic 适合「一条流水线多人抢活干」LiteTopic 适合「每个 Agent、每个用户各有一条专属通道」。还有一个容易被忽略的能力Consumption Suspend消费暂停。GPU 资源紧张时AI 平台常要对不同用户做差异化限流。传统限流的问题是一个慢用户可能阻塞整个消费线程。LiteTopic 把限流粒度下沉到会话级别——某个会话超限了只暂停这一条 LiteTopic其他用户照常消费。阿里云百炼网关已经在生产环境用这套方案做 AI 推理的流量治理这个思路值得借鉴。环境上你需要准备RocketMQ 5.5.0 的二进制包、JDK 8 以上、能跑 NameServer/Broker/Proxy 的机器。下面所有配置我都按本地单机验证来写生产环境把地址换成集群地址即可。3. 可复制配置broker.conf、父 Topic 与 LiteGroup 三件套这一节是全文最核心的部分配置写错后面全跑不通。我按「启动顺序」给你完整片段。第一步下载解压并追加 LiteTopic 配置。注意enableLmq、enableMultiDispatch、storeTypedefaultRocksDB这三个是关键缺一个 LiteTopic 都起不来wget https://mirrors.aliyun.com/apache/rocketmq/5.5.0/rocketmq-all-5.5.0-bin-release.zip unzip rocketmq-all-5.5.0-bin-release.zip cd rocketmq-all-5.5.0-bin-release cat conf/broker.conf EOF enableLmqtrue enableMultiDispatchtrue storeTypedefaultRocksDB EOF第二步按顺序启动 NameServer、Broker、Proxynohup sh bin/mqnamesrv nohup sh bin/mqbroker -n localhost:9876 -c conf/broker.conf nohup sh bin/mqproxy -n localhost:9876 第三步创建父 Topic 和 LiteGroup。父 Topic 要带message.typeLITE属性LiteGroup 要绑定父 Topic这两处是 LiteTopic 能路由的前提# 创建 LiteTopic 父 Topic sh bin/mqadmin updateTopic -b localhost:10911 -t AGENT_TASK_NS -a message.typeLITE # 创建 LiteGroup并绑定父 Topic sh bin/mqadmin updateSubGroup -b localhost:10911 -g executor-group -o true \ --attributes lite.bind.topicAGENT_TASK_NS如果你用配置文件方式管理等价的一份broker.conf关键项如下方便你对照brokerClusterName DefaultCluster brokerName broker-a brokerId 0 namesrvAddr localhost:9876 enableLmq true enableMultiDispatch true storeType defaultRocksDB这里有个细节要提醒storeType必须是defaultRocksDB如果你沿用旧的存储类型消费位点的 KV 存储会失效表现就是 Agent 重启后进度丢失。我实测下来这个坑最容易在从旧版本升级时踩到。配置三件套Base URL、Key、Model ID在 AI 消费端同样要写全。下面 Worker 端我会用setEndpoints指向 Proxy 地址消费组用executor-group模型调用部分用占位服务你替换成自己的推理服务即可。4. 端到端验证Master 派任务、Worker 订阅、消息触发智能体响应配置好了现在跑通链路。我按「发送端 → 消费端 → 验证结果」三步走。发送端Master Agent的核心是setLiteTopic()它指定目标 Agent 的专属通道无需提前创建import org.apache.rocketmq.client.apis.*; import org.apache.rocketmq.client.apis.message.Message; import org.apache.rocketmq.client.apis.producer.Producer; public class MasterAgentDemo { static final String PARENT_TOPIC AGENT_TASK_NS; public static void main(String[] args) throws ClientException { ClientServiceProvider provider ClientServiceProvider.loadService(); ClientConfiguration config ClientConfiguration.newBuilder() .setEndpoints(localhost:8081) .build(); Producer producer provider.newProducerBuilder() .setClientConfiguration(config) .setTopics(PARENT_TOPIC) .build(); String executorAgentId agent_001; String taskPayload {\task\:\分析用户反馈并生成摘要\}; Message task provider.newMessageBuilder() .setTopic(PARENT_TOPIC) .setLiteTopic(TASK_ executorAgentId) .setBody(taskPayload.getBytes()) .build(); producer.send(task); System.out.println(任务已派发给 Agent: executorAgentId); } }消费端Worker Agent用LitePushConsumersubscribeLite()动态订阅专属通道首次订阅自动创建 LiteTopic。消息监听里调 LLM再把结果回传import org.apache.rocketmq.client.apis.*; import org.apache.rocketmq.client.apis.consumer.ConsumeResult; import org.apache.rocketmq.client.apis.consumer.LitePushConsumer; public class WorkerAgentDemo { static final String PARENT_TOPIC AGENT_TASK_NS; public static void main(String[] args) throws ClientException { ClientServiceProvider provider ClientServiceProvider.loadService(); ClientConfiguration config ClientConfiguration.newBuilder() .setEndpoints(localhost:8081) .build(); String executorAgentId agent_001; LitePushConsumer worker provider.newLitePushConsumerBuilder() .setClientConfiguration(config) .setConsumerGroup(executor-group) .bindTopic(PARENT_TOPIC) .setMessageListener(msg - { String task new String(msg.getBody()); String result llmService(task); sendReply(RESULT_ msg.getMessageId(), result); return ConsumeResult.SUCCESS; }) .build(); worker.subscribeLite(TASK_ executorAgentId); System.out.println(Worker Agent 已就绪等待任务...); } private static String llmService(String task) { return 【AI 摘要结果】用户反馈整体偏正面建议关注物流时效。; } private static void sendReply(String liteTopic, String result) { // 用 Producer.setLiteTopic() 发送回 Master Agent } }如果你只想先验证 LiteTopic 能不能通用官方精简写法最快Message message provider.newMessageBuilder() .setTopic(yourParentTopic) .setLiteTopic(lite-topic-1) .setKeys(session-abc) .setBody(Hello LiteTopic!.getBytes()) .build(); try { SendReceipt receipt producer.send(message); System.out.println(发送成功, messageId receipt.getMessageId()); } catch (LiteTopicQuotaExceededException e) { System.err.println(LiteTopic 配额已满请评估并扩容); }验证成功的标志有三个发送端打印出 messageIdWorker 端控制台出现「已就绪」并消费到任务把 Worker 进程 kill 掉再重启Broker 自动恢复消费位点从断点继续而不是从头再来。第三个是 LiteTopic 最值钱的地方一定要亲手验一遍。5. 本篇常见错排查401、local proxy failed、reading choices 与 OAuth跑 Demo 时最容易卡在几个报错上我按真实日志给你对照。报错一401 Unauthorized或authentication failed。如果你在消费端接了模型服务401 通常是 API Key 没配或过期。检查三件套是否写全Base URL、Key、Model ID。Base URL 用https://taotoken.net/apiKey 在控制台生成Model ID 要和实际调用的模型一致。少任何一个都会 401。报错二local proxy failed或connect to proxy failed。这是 Proxy 没起来或端口不对。确认mqproxy进程在跑setEndpoints(localhost:8081)的端口和 Proxy 实际监听端口一致。我见过有人把 8081 写成 9876那是 NameServer 端口结果一直连不上。报错三reading choices相关解析异常。这通常出现在模型返回体解析阶段说明请求发出去了但响应结构和你代码里取字段的路径不匹配。先打印原始响应体确认返回的是标准结构再调整解析逻辑。别急着改 LiteTopic 配置问题不在消息层。报错四OAuth 相关报错。如果你用的是需要 OAuth 的模型服务token 过期会报这个。刷新 token 后重试同时确认消费端没有缓存旧 token。报错五LiteTopicQuotaExceededException。这是 LiteTopic 配额超限说明单 Broker 承载的轻量通道到了上限。评估一下是不是有大量会话没被 TTL 回收或者需要扩容 Broker。排查顺序建议先确认 NameServer/Broker/Proxy 三个进程都在再确认父 Topic 带message.typeLITE、LiteGroup 绑定了父 Topic最后才查模型侧的三件套。消息层和模型层的问题要分开定位不然容易互相甩锅。6. 把链路接进你的 Agent 系统从 Demo 到生产的几个实用建议Demo 跑通只是起点。真正接进生产有几个点值得提前想清楚。第一会话命名要有规范。我用的是TASK_agentId和RESULT_sessionId两套前缀前者按 Agent 维度分后者按会话维度分。命名乱了后面排查问题时你根本不知道哪条 LiteTopic 对应哪个用户。第二TTL 要按业务设。LiteTopic 支持 TTL 自动回收过期会话但默认值不一定适合你。长会话类应用 TTL 设短了会误删设长了会堆积。建议按「会话最长存活时间 缓冲」来定。第三消费暂停要用在刀刃上。GPU 紧张时用 Consumption Suspend 按会话粒度限流比全局限流温和得多。VIP 用户优先、普通用户排队这套逻辑在 LiteTopic 上实现起来很自然。第四断点续传要验证。这是 LiteTopic 相对传统 Topic 最大的优势但前提是storeTypedefaultRocksDB配对。上线前一定做一次「kill Worker 再重启」的演练确认进度没丢。如果你正在做 Multi-Agent 系统或 AI 工作流编排建议花一个下午把这条链路跑一遍。模型侧接入用 TaoToken 的 API 就行Base URL 填https://taotoken.net/apiKey 在控制台生成模型对话调试可以直接在模型对话页验证返回结构长期跑 Agent 任务的话 Coding Plan 更划算。接入文档里有完整的参数说明遇到报错先对照文档排查比盲目改配置快得多。