从 RocketMQ 5.0 开始，社区引入了 POP 这种消费模式。客户端发起 POP 请求后，Broker 从 queue 中取出消息返回，并在 Broker 侧维护不可见期、ACK、续租和超时恢复这套状态——尤其在非 FIFO 场景下，它允许多个消费者并发处理同一 queue 上的消息。 在实际业务里，POP 往往不是因为“想换一种消费接口”才被拿出来，而是在消费并发已经碰到瓶颈时，才真正体现出价值。 一个很典型的场景是：某个 Topic 的 queue 数已经拉到很高，比如 500；消费客户端也已经拉到 500；但一个客户端处理一个 queue，吞吐还是不够。写入侧没有明显瓶颈，真正卡住的是消费侧并发。这个时候，业务真正需要的，往往不是继续把 Topic 切得更碎，而是让多个客户端共同处理同一个 queue 上的数据。 从这个场景往下看，POP 可以先理解成一种“消息被 Broker 借出一段时间”的消费模型： + 消息取走后，不会立刻算完成 + 消息会先进入不可见期 + 处理成功后再 ACK + 处理时间不够时可以续租 + 超时后还要重新进入后续消费流程 顺着这个语义再来看源码，后面的 invisibleTime、ACK、changeInvisibleTime、revive，以及 Broker 里那套“已投递但未完成”的状态管理，就比较容易串起来。这里把 Broker 为消息保存的这类“消费中”状态统称为“投递记录”。 图注：POP 消费的核心语义——消息被 Broker 借出一段时间，取走后进入不可见期，成功后 ACK，不够则续租，超时则 Revive 重投到 Retry Topic。 一、为什么需要 POP ▍1.1 传统 Pull 为什么不太够用 传统 Pull 模式下，消费进度主要由消费位点驱动；在同一消费组内，一个队列在同一时刻通常只会稳定地分配给一个消费者实例。 这套模型很适合“按位点往前推进”的流式消费，但在下面这类场景里，局限会暴露得更明显： + 单条消息处理时间明显长于一次普通消费循环 + 处理成功与否不能只靠“位点推进了没有”来表达 + 客户端需要显式告诉 Broker：我处理完了、还没处理完、需要再给我一点时间 + 更希望消费语义接近“消息先领出处理、完成后再确认”，而不是单纯按 queue 上的位点顺序往前推进 问题不再只是“怎么把消息拉出来”，而是“怎么让消息被领出来处理一段时间，同时还能继续放大消费并发”。 单靠继续加 queue，本质上还是在用更多 queue 换并发；但当 queue 数已经很高时，很多时候业务更需要的，其实是允许多个客户端共同处理同一个 queue 上的数据。 这也正是非 FIFO POP 真正有意义的地方：它把原来更接近“一个 queue 稳定分给一个消费者实例”的消费方式，往“同一 queue 上可以存在多条被不同消费者领走、但尚未最终确认的消息”这个方向推进了一步。 ▍1.2 POP 的关键变化在哪里 POP 真正补上的，不是一个新的拉消息动作，而是 Broker 侧一层明确的“处理中状态”。 消息被取走后，不会立刻算消费完成，而是先进入一段受 Broker 管理的处理中窗口：这段时间里消息不可见，成功后 ACK，处理不完可以续租，超时则重新进入后续消费流程。 这样一来，Broker 就可以同时跟踪同一 queue 上多条“已经投递、但尚未最终完成”的消息，POP 能支撑更高消费并发的关键也在这里。 后面会反复出现的 invisibleTime、ACK、changeInvisibleTime、revive，本质上都是围绕这层状态展开的。 更详细的对比如下图： 二、POP 的几个核心概念 下面讨论 RocketMQ 5.4.0 里的 KV 路径，前面说的“投递记录”在这里具体指 PopConsumerRecord。旧路径和 FIFO 放到后文再补。 ▍2.1 invisibleTime invisibleTime 是 POP 最核心的语义。消息被 POP 成功后，在 popTime + invisibleTime 之前，这条消息对该消费组后续的 POP 请求不可见。 如果只抓住一件事，那就是：Broker 把消息借出去之后，不会立刻把它算作完成，而是先给它挂上一段“处理中窗口”。这段窗口就是 invisibleTime。 ▍2.2 ACK ACK 的核心语义不是简单推进消费位点，而是把“这条消息已经完成”这件事落到 Broker 侧状态里。 更具体一点说，就是把这条消息对应的 PopConsumerRecord 删除掉。它不是“把 queue 往前推一步”，而是把这条消息从“已投递但未完成”的集合里移出去。 ▍2.3 changeInvisibleTime 如果业务处理超过原来的 invisibleTime，客户端可以调用 changeInvisibleTime 续租。从语义上看，它不是在改消息内容，而是在把这次投递的可见时间整体往后推。 放到实现上，它体现为： + 写一条新的 PopConsumerRecord + 再删除旧记录 所以它本质上是在切换“这次投递对应的超时点”。 ▍2.4 revive revive 主要对应超时恢复机制。 如果一条消息在不可见窗口内一直没有被 ACK，Broker 后续会把它重新投递到该消费组对应的 retry topic，后续 POP 请求再按 popFromRetryProbability 的轮转策略从 retry topic 取消息，这个过程就是 revive。也就是说，revive 不是把消息直接放回原始 queue，而是走重试投递路径。 需要特别注意的是，revive 并不是一个“附加的重试能力”，而是这条消费模型下的必然结果。 从 Broker 视角看，一旦消息返回给客户端后，后续读取位置已经继续前移，而这条消息本身又仍处于“已投递但未确认”的状态，Broker 就不能再只靠 offset 把它重新取回来。 因此，如果客户端在不可见期内一直没有 ACK，Broker 就必须依赖这层消息级未确认状态的扫描与 revive 机制，把消息重新投递到 retry topic，再进入后续 POP 消费流程。 这也是为什么 revive 本质上是在弥补“offset 已前移，但消息尚未完成”的状态缺口。 ▍2.5 POP 还有 offset 吗？ 有。 Broker 至少同时维护两层状态： + pullOffset / consumeOffset —— 决定后续从哪里继续读或推进 + PopConsumerRecord —— 表达“哪些消息已经投递出去、但还没有最终完成” 所以 offset 前移不等于消息已经真正消费完成，只有对应的 PopConsumerRecord 被 ACK 删除、续租切换或超时恢复后，这条消息的状态才算真正闭环。 很多人第一次看 POP，会误以为“既然已经按消息 ACK 了，那是不是就没有 offset 了”。实际上并不是。POP 里 offset 一直都在，只是进入 POP 之后，offset 不再足以单独表达全部消费状态。 如果把 Pull 和 POP 放在一起看，差别不在“有没有 offset”，而在“offset 是否还足够表达全部消费状态”： + 对 Pull 来说，offset 基本就能同时表达“从哪里继续读”和“消费推进到了哪里” + 但对 POP 来说，Broker 侧还多了一层“已投递但未完成”的消息级状态 后面整篇文章，其实就是在把这层状态是怎么建立、怎么变更、怎么收尾讲清楚。 三、先按一条主线读源码 RocketMQ 5.4.0 里新旧两条实现路径仍然并存，但如果一开始就把两套状态模型交叉着讲，读者很容易在“当前这一步到底落在哪套实现里”上迷路。 所以下面只顺着一条线往下走：PopMessageProcessor → PopConsumerService → PopConsumerRecord → ACK / changeInvisibleTime / revive。旧路径和 FIFO，都放到后面再统一说。 图注：POP 源码的三层架构——请求入口（PopMessageProcessor）、核心服务（PopConsumerService）、状态存储（PopConsumerCache + PopConsumerKVStore）。核心投递记录 PopConsumerRecord 贯穿全流程。 ▍3.1 统一入口：PopMessageProcessor POP 请求统一从 PopMessageProcessor.processRequest() 进入。它负责： + 校验 Broker / Topic / ConsumerGroup 权限 + 校验 maxMsgNums，当前限制最大 32 + 校验 queueId + 校验 timerWheelEnable，没开时 POP 直接拒绝 + 构造过滤表达式和 MessageFilter + 根据配置决定走哪条 POP 实现路径 + 在没取到消息时进入 PopLongPollingService 代码位置：broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java，入口方法 processRequest(...) 它是 POP 的入口，但不是全部逻辑所在。 ▍3.2 核心组件：PopConsumerService 当brokerConfig.isPopConsumerKVServiceEnable() 为 true 时，POP 主流程会委托给 PopConsumerService.popAsync(...)。 这一套是这里真正的主角，核心组件包括： PopConsumerContext 封装单次 POP 的上下文，保存本次返回的 GetMessageResult 和本次生成的 PopConsumerRecord，汇总返回给客户端的 startOffsetInfo、msgOffsetInfo。 PopConsumerRecord 表示“一条消息被 POP 出去但还没最终确认”的记录。核心字段：popTime / groupId / topicId / queueId / retryFlag / invisibleTime / offset / attemptTimes / attemptId。其中 attemptTimes 是 revive 重试次数，每次 revive 失败后递增，用于退避策略。visibilityTimeout = popTime + invisibleTime。 PopConsumerKVStore 默认是 RocksDB 存储实现。 PopConsumerCache enablePopBufferMerge 打开时的内存缓冲，缓冲区定期刷盘或触发 revive。 PopConsumerLockService 按 groupId + topicId 做互斥。 代码位置：broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java，入口方法 popAsync(...) 四、POP 请求主流程 下面把一次 POP 请求从入口到返回完整走一遍。 图注：一次完整 POP 请求的 10 步主流程 + 长轮询备选路径（A/B/C）。注意长轮询本质仍是“请求驱动的拉取”，唤醒时重新调用 processRequest()。 ▍4.1 请求进入 Broker 客户端发 POP_MESSAGE 请求后，PopMessageProcessor 先做一轮硬校验： + Broker 是否可读 + Topic 是否存在且可读 + ConsumerGroup 是否存在且允许消费 + requestHeader.getMaxMsgNums() ≤ 32 + Store 是否开启 timer wheel + queueId 是否合法 如果请求带过滤表达式，还会构建普通 Topic 和对应 retry topic 的 SubscriptionData，以及对应的 ExpressionMessageFilter。 ▍4.2 创建 PopConsumerContext 进入 PopConsumerService.popAsync(...) 后，会先创建 PopConsumerContext，里面记录：客户端地址、popTime、invisibleTime、消费组、attemptId、本次累积返回的消息结果和投递记录。这个对象贯穿单次 POP。 ▍4.3 加锁粒度：groupId + topicId PopConsumerService 会先尝试对 groupId + topicId 加锁。注意这里不是“每个队列一把锁”，而是消费组和 Topic 粒度的互斥。 这么做的目的，主要是保证一次 POP 紧密相关的状态变更不会互相踩踏：取消息、建投递记录、提交 offset，都在这把锁保护下串起来。 ▍4.4 先取重试还是先取普通消息 POP 不是永远先取普通 Topic。PopConsumerService 里会根据配置决定是否优先取重试消息： + popFromRetryProbability + popFromRetryProbabilityForPriority 并结合以下条件算出 preferRetry：Topic 类型是否为 PRIORITY，且 requestCount % 100 拿到 probability 后，preferRetry = (probability 0) && (requestCount % (100 / probability) == 0)。 取消息的实际顺序也与 preferRetry 和 queueId 模式有关： + preferRetry=true：先取 retry topic（v1/v2），再取普通 Topic，不再重复取 retry + preferRetry=false 且 queueId==1：先取普通 Topic，再取 retry topic（v1/v2） + preferRetry=false 且 queueId!=1：只取指定普通队列，这次请求里不会再补读 retry topic 也就是说，它不是简单的“普通队列空了再去重试队列”，而是带概率控制、并且受 queueId 模式约束的优先级轮转策略。 ▍4.5 支持 retry topic v1/v2 这里会根据配置决定是否从这些重试主题取消息：KeyBuilder.buildPopRetryTopicV1(...) 和 KeyBuilder.buildPopRetryTopicV2(...)。并且 retryFlag 会记在 PopConsumerRecord 里，后面 revive/返回消息时都要用到。 ▍4.6 queueId 的两种模式 指定队列：如果 queueId != 1，只从指定队列取。 全队列 POP：如果 queueId == 1，会遍历 Topic 的所有读队列。遍历起点和方向不是固定的：结合请求计数做轮转，尽量避免每次都从固定队列开始；同时受 priorityOrderAsc 控制，决定从低 queueId 向高递增，还是从高 queueId 向低递减。 ▍4.7 真正取消息 每个队列最终都会调用 getMessageAsync(...) 去读存储。这里有几个关键点： + POP 使用的是 pullOffset + 如果发现 offset 过小/过大/无效，会修正 offset 后重试读取 + 读到消息后，会同步更新 commitPullOffset(...) 和 commitOffset(...) 要注意：commitPullOffset 更像“下一次从哪里继续试探着拉”；打开 enablePopBufferMerge 时，commitOffset 可能会被压到 PopConsumerCache 里最小的未确认 offset，而不是这次读取的 nextBeginOffset。 commitOffset 仍然是 POP 在 Broker 侧维护的一部分消费进度，但真正决定一条消息是否已经消费完成的，仍然是后面的 ACK、续租和 revive。 ▍4.8 inflight 背压：isPopShouldStop 在每个队列真正取消息之前，PopConsumerService 还会先调用 isPopShouldStop(groupId, topicId, queueId) 做背压判断。 当 enablePopMessageThreshold=true 且该队列当前 inflight 消息数（由 PopConsumerCache 统计的未 ACK 记录数）已经达到 popInflightMessageThreshold 时，isPopShouldStop 返回 true，这次队列迭代直接跳过，不再读存储。 这和长轮询里的 POLLING_FULL 是两个独立的流控层： + POLLING_FULL：长轮询挂起队列容量已满，新请求无法挂起 + isPopShouldStop：某个队列的 inflight 消息过多，主动限制继续发送 ▍4.9 读到消息后如何建模 PopConsumerContext.addGetMessageResult(...) 会先在本次请求上下文里，为每条命中的消息挂一条 PopConsumerRecord。 可以把它理解成：Broker 在说“这条 offset=xxx 的消息，我在 popTime 时刻交给了这个消费组，它要在 invisibleTime 后才算超时。” 这是 POP 跟 Pull 最大的不同。但要注意，这一步只是先把“投递记录”挂在 PopConsumerContext 里；真正写到缓存或 KV Store，是下一步的事。 ▍4.10 投递记录写到哪里 如果这次 POP 确实取到了消息，这些 PopConsumerRecord 就会被真正保存下来： + 如果 enablePopBufferMerge=true 且缓存没满，先写 PopConsumerCache + 否则直接写 PopConsumerKVStore（默认 RocksDB） 所以这里真正的核心投递记录就是 PopConsumerRecord。 投递记录写入路径图 图注：PopConsumerRecord 从创建到持久化的完整写入路径。enablePopBufferMerge=true 时经内存缓冲刷盘，否则直写 RocksDB。存储键为 visibilityTimeout + groupId + topicId + queueId + offset。 ▍4.11 返回给客户端前的重试消息改写 如果取到的是 retry topic 上的消息，返回客户端前，Broker 可能会对“客户端看到的 topic”做一层重编码；但这里要特别注意两件事： + 这一层更多是“响应呈现层”的行为，不等于 ACK/续租时真正使用的 topic + 客户端后续 ACK、续租最终依赖的仍然是 PROPERTY_POP_CK / receipt handle 里携带的 retry 标记，客户端会据此恢复 real topic 所以从客户端视角看，“日志里看到的消息 topic”和“Broker 这次实际从哪个主题取到消息”，不一定始终一一对应。这个问题后面再单独展开。 ▍4.12 没取到消息时的长轮询 如果本次 POP 没取到消息，不一定立刻返回空。只要请求里的 pollTime 0，请求就会进入 PopLongPollingService.polling(...)。 长轮询服务会把请求按 topic + consumerGroup + queueId 挂到 pollingMap 里。已经挂起的请求，主要会在下面几种场景被重新处理： + 新消息到达 + 挂起超时 + 服务停止时的清理唤醒 这里的“重新处理”不是 Broker 手里已经攥着一批消息，等条件满足后直接把它推给挂起请求，而是重新调用一次 processor.processRequest(...)，也就是再走一遍 POP 处理流程。 需要特别注意的是，POP 的长轮询本质上仍然是“请求驱动的拉取”，而不是“服务端主动推送”。 要额外注意的是，POLLING_FULL 不是唤醒场景。如果长轮询队列本身已经满了，请求不会进入挂起队列，而是会直接返回 POLLING_FULL。 五、ACK、续租和 revive 消息返回客户端之后，还剩三件事要闭环：处理完成怎么 ACK，时间不够怎么续租，超时之后又怎么恢复。 ▍5.1 ACK ACK 请求入口是 AckMessageProcessor。最重要的是它最终会落到 PopConsumerService.ackAsync(...)。它的主流程可以理解成这样： 1. AckMessageProcessor.appendAckNew(...) 先从 extraInfo 或 BatchAck 里还原这次 POP 的关键信息 2. 然后调用 PopConsumerService.ackAsync(...) 3. ackAsync(...) 会根据 popTime/groupId/topicId/queueId/invisibleTime/offset 构造对应的 PopConsumerRecord 作为查找 key；底层存储键最终会折算成 visibilityTimeout + groupId + topicId + queueId + offset 4. 然后先尝试从 PopConsumerCache 删除；删不到再去 PopConsumerKVStore 删除 所以这里的 ACK，就是把这条消息对应的“进行中记录”删掉。 ▍5.2 为什么 ACK 不是简单提交 offset 因为 POP 是“消息级确认”，不是“队列级顺序推进”。同一批次返回的多条消息里，可能有的已经处理完成，有的还在处理中，所以 Broker 必须保留消息级的未确认状态，而不能只靠一个队列 offset 表达完成状态。 放在这里看，这层状态就是 PopConsumerRecord。 ▍5.3 changeInvisibleTime ChangeInvisibleTimeProcessor 处理续租请求。它最终会进入 PopConsumerService.changeInvisibilityDuration(...)。这条分支会做两件事： + 写入一条新的 PopConsumerRecord（changedPopTime + changedInvisibleTime） + 删除旧的 PopConsumerRecord 本质上等于把原来的“超时点”整体往后挪。 要注意，这里新生成的记录不是重新写回 PopConsumerCache，而是直接写 PopConsumerKVStore，然后再删除旧记录。 ▍5.4 超时未 ACK 后如何 revive PopConsumerService 自身就是一个背景线程服务，会定期扫描 PopConsumerKVStore 里的过期记录： + 实现上通过 scanExpiredRecords(...) 扫描已经进入过期窗口的 PopConsumerRecord + 再去读原始消息 + 把原始消息重写到 retry topic + 删除旧记录 如果打开了 enablePopBufferMerge，还要再补一层理解：部分记录会先进入 PopConsumerCache，缓存线程会先把未过期记录刷到 store，把已过期记录直接交给 revive 逻辑处理。所以“后台线程只扫 KVStore”是主路径，但不是全部。 如果 revive 失败，不是无限重试，而是会按 attemptTimes 做退避重试。超过最大尝试次数后，日志里会出现 message may be lost。 ▍5.5 revive 重投到哪里 如果原消息本来就来自重试主题，则继续回到对应重试主题。如果原消息来自普通 Topic，则会重投到 POP retry topic（v1 或 v2，取决于 Broker 配置），并且会带上 PROPERTY_FIRST_POP_TIME、PROPERTY_ORIGIN_GROUP 和最新的重试次数。 图注：ACK 删除 PopConsumerRecord 标记完成，changeInvisibleTime 写新记录+删旧记录将超时点后推。两者操作路径、写入目标、最终效果均不同。 六、最后补三个容易混的点 ▍6.1 FIFO 为什么不是这套模型 前面讲的这套模型里，核心一直是“每条消息对应一条 PopConsumerRecord，再围绕它做 ACK、续租和 revive”。FIFO 不是这套思路。 FIFO 更像“带不可见期的顺序消费”。Broker 关心的不是每条消息各自有没有一条独立投递记录，而是同一队列上的顺序窗口有没有被前面的未 ACK 消息卡住。对应到源码上，核心状态从 PopConsumerRecord 换成了 OrderInfo，由 ConsumerOrderInfoManager 维护。 所以 FIFO 不走前面那套消息级 retry/revive 主流程。它的超时恢复也不是后台扫描，而是后续 POP 通过 OrderInfo.needBlock() 判断阻塞窗口是否已经过期；一旦阻塞解除，因为 committedOffset 还没前进，消息会从原始 queue 原地重新投递，而不是进 retry topic。 ▍6.2 旧路径在做什么 旧路径和前面讲的是同一个问题的另一套实现。如果 isPopConsumerKVServiceEnable() 为 false，Broker 不再用 PopConsumerRecord 表示“已投递未确认”，而是换成 PopCheckPoint / AckMsg / BatchAckMsg 这一组对象。 大致可以把它理解成： + POP 取消息后生成 PopCheckPoint + ACK 通过 AckMsg / BatchAckMsg + PopBufferMergeService 表达 + changeInvisibleTime 会追加新的 PopCheckPoint，再去 ACK 旧 checkpoint + 超时恢复依赖 revive topic 和 PopReviveService 旧路径和前文解决的是同一个问题，但中间状态模型已经完全换了一套。理解到 checkpoint、ack 日志和 revive topic 这一层就够了。 ▍6.3 涉及重试消息时，客户端看到的 topic 为什么会不同 这里最容易混淆的一点是：客户端看到的消息 topic，不等于 ACK / 续租真正作用的 real topic。 POP 返回消息时，Broker 可能会对响应里的 topic 做重编码，尤其是消息实际来自 retry topic 时更容易让人误会。但客户端后续 ACK / changeInvisibleTime 依赖的并不是当前消息对象上的 topic，而是 PROPERTY_POP_CK / receipt handle 里的 retry 标记，再据此恢复 real topic。 所以仅凭客户端日志里看到的 topic，不能反推 Broker 这次一定是从普通 topic 读到的，还是从 retry topic 读到的。把这件事分开看就不容易绕：日志里的 topic 更像是响应呈现层，ACK / 续租用的 topic 才是语义层。 图注：客户端日志中的 topic 属于“呈现层”（可能被 Broker 重编码），而 ACK/续租使用的 real topic 属于“语义层”（来自 PROPERTY_POP_CK 中的 retry 标记）。两者不能混为一谈。 七、再往外看：为什么 Kafka 也在演进这类能力 如果把 POP 从 RocketMQ 自己的实现细节里跳出来看，它解决的其实不是一个 RocketMQ 独有的问题，而是一类更普遍的消息消费诉求： + 共享消费 + 显式确认 + 处理中窗口控制 + 超时后的再次投递 RocketMQ POP 是其中一种实现方式。它仍然建立在 RocketMQ 自己的 Topic、queue、retry topic、FIFO/非 FIFO 分支之上，但消费语义上已经明显引入了“消息被借出一段时间、完成后确认、超时后恢复”的队列式特征。 Kafka 近几个版本也在沿着这个方向演进。按照 Apache Kafka 官方文档和 KIP932 的表述： + Kafka 4.1 把 share groups 作为 preview 形态引入 + Kafka 4.2 把 Queues for Kafka / share groups 标记为 productionready + 它提供的是一种不同于传统 consumer group 的共享消费模型 + 不再要求每个 partition 只能分配给一个消费者实例处理 + 消费者可以做逐条 ACK，并跟踪 delivery attempts；同时系统仍然针对 batch processing 做了优化 + 官方文档也明确说，这类能力更适合“records are processed one at a time”，而不是典型的 ordered stream 如果只看高层语义，会发现它和 POP 想解决的问题其实很像：都在补“共享消费 + 显式确认 + 失败后继续处理”这类能力。当然，两者并不是一套实现： + RocketMQ POP 的关键语义是 invisibleTime、ACK、续租、revive，以及围绕这些语义建立的 Broker 侧投递记录 + Kafka share groups 的关键语义是共享消费组、逐条 ACK、delivery attempts，以及围绕 share group 建立的新协调与状态管理 这也说明，“共享消费、逐条确认、处理中窗口控制、失败后继续处理”并不是某一个消息系统偶然长出来的特性，而是消息系统里真实存在的一类业务需求。不同消息系统都在沿着各自的体系往这个方向演进，只是具体实现路径和抽象方式并不相同。 八、关键源码索引 以下以 release5.4.0 分支、提交 b5da00ad0 为代码基线。 | 功能 | 关键类 / 方法 | | :: | | | POP 请求入口 | PopMessageProcessor.processRequest() | | KV 路径主流程 | PopConsumerService.popAsync() | | 重试消息重编码 | PopConsumerService.recodeRetryMessage() | | ACK 主流程 | AckMessageProcessor.processRequest()→ appendAckNew()→ PopConsumerService.ackAsync() | | 续租主流程 | ChangeInvisibleTimeProcessor.processRequestAsync()→ PopConsumerService.changeInvisibilityDuration() | | revive 主流程 | PopConsumerService.revive(PopConsumerRecord) / revive(AtomicLong, int) | | 旧路径buffer merge | PopBufferMergeService | | 旧路径 revive | PopReviveService | | FIFO 顺序状态 | QueueLevelConsumerManager | | 长轮询挂起 | PopLongPollingService.polling() | | 长轮询唤醒 | PopLongPollingService.wakeUp() | 总结 POP 消费机制的核心可以概括为三句话： 第一，POP 补上了 Broker 侧的“处理中状态”。 消息被取走后不立刻算完成，而是进入 invisibleTime 不可见窗口，Broker 通过 PopConsumerRecord 跟踪每条“已投递但未完成”的消息。 第二，消费进度不再仅靠 offset 表达。 offset 决定从哪里继续读，PopConsumerRecord 决定哪些消息还在处理中。ACK 删除记录、续租切换超时点、revive 处理超时恢复，三者共同构成状态闭环。 第三，这是消息系统的一类共性需求。 从 RocketMQ 的 POP 到 Kafka 的 share groups，不同系统都在用各自的方式实现“共享消费 + 显式确认 + 处理中窗口控制”，因为这是实际业务中的真实诉求。

随着大语言模型（LLM）、语音识别（ASR）、语音合成（TTS）等能力逐步成熟，AI Agent 开始从文本交互走向语音交互，典型场景包括 AI 教师、AI 情感聊天、AI 助手等。相比文本输入，语音更自然、更实时，用户可以直接通过说话完成提问、练习、任务触发与多轮对话，这也让“和 Agent 用语音对话”真正进入实际业务场景。 但当 Agent 语音交互进入高并发场景后，很多团队会发现：最先遇到瓶颈的，往往不是模型本身，而是支撑实时交互的消息链路。海量会话管理、高频小包传输、异步结果回推、会话生命周期管理等问题会随之集中出现。要让和 Agent 语音交互真正做到更稳、更快，底层链路设计往往才是关键。 本文结合一个典型的高并发智能语音交互场景，介绍如何基于阿里云 RocketMQ LiteTopic 构建一套更稳定、更可靠、更高效的实时语音消息链路架构。 高并发 Agent 语音交互对技术架构提出哪些关键要求？ 在 Agent 语音交互场景中，系统并不只是“接收一句话、返回一句话”这么简单。一次完整交互背后，往往涉及客户端、网关、业务处理系统，以及 LLM、ASR、TTS 等多个服务之间的协同。 因此，智能语音交互业务会对技术架构提出更高要求： + 海量会话管理：随着业务规模增长，并发连接数和活跃会话数会迅速上升。每个用户的语音交互都是一个独立会话（Session），系统需要同时维持数万甚至数十万个长连接。 + 高频小包传输：用户按下录音键到松开，形成一个独立会话（Session）。在此期间，客户端会将音频流切片成小包持续传输，需要保证语音包连续、不丢失，避免影响业务正确性。 + 严苛的时效性：客户端对延迟极度敏感，若较长时间内未收到响应，用户体验会明显下降。这对 LLM 在高并发场景下的吞吐能力，以及系统的实时响应通知能力，都提出了更高要求。 正因如此，很多系统在低并发时看起来“可以跑通”，但一旦进入高并发实时语音场景，底层消息架构中的问题就会被迅速放大。 传统消息架构在实时语音场景下面临哪些核心挑战？ 在智能语音交互业务的实际落地过程中，传统消息架构在支撑高并发、低延迟的实时语音场景时，往往会暴露出几类典型问题。 ▍1. 全链路 Session Sticky 的精准路由 语音交互的消息流转路径通常贯穿：APP Gateway BizProcessSystem (Route) LLM/ASR/TTS。其中，APP与 Gateway、BizProcessSystem与 LLM之间均维持着 WebSocket 长连接。 在这种架构下，整个链路必须严格保持会话粘滞（Session Sticky）。也就是说，某个用户的上行音频流和下行反馈结果，必须精准路由到其当前连接的特定网关节点，以及对应的后端处理实例。 问题在于，在分布式环境下，维护“Session ID 到物理节点 IP”的动态映射表本身就非常复杂。一旦网关节点扩容、重启，或者发生网络波动，路由表同步延迟极易导致消息被投递到错误节点，进而造成连接断裂、数据丢失，破坏交互连续性。 ▍2. 大模型异步结果的实时、精准回推 大模型（LLM）的推理过程通常耗时较长且波动明显（秒级甚至分钟级）。若采用同步等待模式，会长时间占用网关和业务线程，导致系统吞吐量急剧下降，甚至引发资源耗尽的雪崩效应。 因此，为了提升吞吐，LLM 调用通常需要改造成异步处理。但异步之后，新的难点也随之出现：最终计算结果（如 ASR 文本、TTS 音频）如何实时、准确地回推给发起请求的那个用户连接？ 如果依赖复杂的回调轮询或状态查询，不仅实现复杂，还会进一步增加延迟和维护成本。这也是语音交互架构设计中的核心难点之一。 ▍3. 海量临时通道带来的元数据爆炸 从业务逻辑上看，为每个独立语音会话（Session）建立隔离的通信通道，是避免数据串扰的理想方案。 但如果为每个 Session 都创建一个标准 RocketMQ Topic，就会带来明显的元数据（Metadata）爆炸问题。海量临时 Topic 会严重消耗 NameServer 和 Broker 的内存与 CPU 资源，导致集群性能急剧下降，甚至影响可用性。 此前，一些场景会采用广播消息的方案来规避这一问题。虽然实现简单，但也存在两个明显缺陷： + 消息会在所有节点重复投递和过滤，造成大量无效流量与计算资源浪费。 + 所有节点都需要处理全量消息，单节点处理能力容易成为整体容量上限，系统水平扩展受限。 因此，广播模式很难支撑持续增长的高并发语音业务。 ▍4. 会话生命周期的自动化管理缺失 语音会话通常具有很强的临时性，并不是永久存在的资源。它的生命周期可能只是一次几分钟的对话，也可能仅存在于一个业务周期之内。 但在传统架构下，会话结束后的路由记录、缓存状态、临时通道等资源，往往需要依赖定时任务扫描或手动清理。这会带来两个典型问题： + 清理不及时：无效资源长期堆积，占用系统内存和计算资源。 + 清理过早：可能切断仍在进行中的合法交互。 因此，系统需要一种真正适合临时会话场景的机制，能够实现通道的自动创建、自动过期和自动销毁。 基于 RocketMQ LiteTopic 的消息链路重构 针对上述问题，可以基于阿里云云消息队列 RocketMQ 的轻量主题（LiteTopic）模型，构建一套更适合高并发智能语音交互场景的消息中间件架构。 LiteTopic 支持动态创建海量轻量主题，天然具备会话隔离能力，并内置 TTL 自动清理机制。这些特性与 Agent 语音交互场景对“高并发、低延迟、强隔离、易回收”的要求高度契合。 ▍1. RocketMQ LiteTopic 方案设计 1.1 请求保序与响应隔离的核心架构 + 请求侧：分片音频包采用分区顺序 Topic 上传 同一个会话的语音包用 SessionID 作为分区顺序的 Key，发送到分区顺序 Topic 中，相同 Key 的消息会按照发送顺序投递给业务处理系统，从而保证同一会话内消息处理有序。 + 响应侧：模型结果通过 LiteTopic 异步通知 为每个会话创建一个 LiteTopic：直接使用 SessionID 作为 LiteTopic 名称。每个语音会话拥有独立的消息通道，天然实现消息隔离。 每个节点订阅不同的 LiteTopic 集合：应用服务端节点作为 Consumer，只订阅与当前节点相关会话的 LiteTopic 集合，确保消息“点对点”精准投递，无需维护复杂路由表。 节点动态订阅 LiteTopic：会话断连后，可以动态删除对应 SessionID 的 LiteTopic 订阅；新会话建立时，可以动态新增对应 SessionID 的 LiteTopic 订阅。当网络异常或者服务节点重启时，也可以利用动态订阅能力续订 LiteTopic 消息，从而保障会话内容连续性。 LiteTopic 自动创建：LiteTopic 无需预先创建。当生产者发送消息时，如果发现 LiteTopic 不存在，会自动创建，且不影响消息发送耗时。 配置合适的 TTL 时间：当 LiteTopic 距离最近一次消息写入超过设定时长后，会被自动删除。可结合实际业务特点，设置合适的 TTL 时长。 1.2 面向运维和问题定位的可观测性 RocketMQ LiteTopic 基于云监控建立了全方位、细粒度的监控、告警与排查体系： + 智能告警：配置 LiteTopic 的消息堆积量阈值。一旦某会话链路延迟超过预期，立即触发告警。 + 快速定位：收到告警后，运维人员可直接在控制台 Group 详情页查看堆积量最高的 LiteTopic 列表及对应的消费者 IP。借助这种细粒度透视能力，原本大海捞针般的故障定位变成了分钟级的精准修复。 ▍2. RocketMQ LiteTopic 方案优势 基于 RocketMQ LiteTopic 的轻量化、灵活订阅、自动创建及 TTL 自动删除等原生特性，这套方案在架构层面主要具备以下三方面优势： 2.1 保障长耗时会话的极致连续性 借助 LiteTopic 的自动创建、“一会话一通道”和动态订阅机制，系统可以为每个语音会话建立独立的响应通道。无论 LLM 推理耗时多久，消息都能在专属通道中有序流转，避免多会话间的消息串扰。 同时，即使后端服务发生扩缩容或网络波动，响应消息仍能回到用户当前连接的网关节点，保证长链路交互中的 Session Sticky 和会话连续性。 此外，通过 LiteTopic 的异步通知机制，系统可以避免长耗时线程阻塞，进一步提升整体吞吐能力，让用户在高峰期也能获得流畅的语音交互体验。 2.2 推动应用架构进一步无状态化 在传统方案中，应用通常需要维护“Session ID 到 Node IP”的路由映射，以及配套的心跳保活和异常清理逻辑，状态管理复杂、维护成本高。 引入 LiteTopic 后，路由逻辑下沉到消息中间件层，业务代码只需围绕 SessionID 发送和接收消息。应用节点因此更接近无状态计算单元，不再强依赖本地连接状态表。这样不仅能够降低状态管理复杂度，也让应用更容易进行弹性伸缩和故障恢复，从而提升整体可维护性与容灾能力。 2.3 降低无效调用带来的模型成本 在传统架构中，消息错投或超时导致的“无响应”往往会触发客户端重试，导致同一段音频被重复发送给 LLM 推理，带来额外的 Token 消耗。 通过更精准的会话路由和可靠的投递机制，这套方案能够更好地保障“一次请求，必达响应”，显著减少因链路问题导致的重复调用，从而直接降低 LLM 的无效 Token 成本。 消息链路优化后带来的业务价值 从业务效果来看，引入 RocketMQ LiteTopic 之后，高并发智能语音交互链路通常可以在以下几个方面获得明显提升： + 用户体验更稳定：可以显著减少因连接状态不一致导致的“无响应”问题，提升语音交互成功率。即使在网络波动场景下，也能更好保障无感知重连，确保交互连续性。 + 系统复杂度更低：不再需要维护复杂的自定义路由表和状态同步逻辑，而是借助 LiteTopic 的原生能力完成会话管理，整体架构更简单，也更易扩展。 + 运维定位更高效：借助细粒度监控与告警，潜在性能瓶颈可以在影响用户前被发现和处理，问题定位与修复效率明显提升。 + 资源成本更可控：借助云消息队列 RocketMQ 版的弹性能力，业务可以按量付费，无需提前预留峰值容量，同时减少重复调用带来的额外模型消耗。 + 业务扩展更从容：更轻量、可扩展的链路设计，也为后续拓展更多实时互动场景打下基础，能够更从容地应对流量增长。 结语 很多团队在构建 Agent 时，往往会优先关注模型能力、推理效果和成本控制。但在高并发实时语音场景中，想把这些能力稳定地交付给用户，消息链路的稳定性、精准性和可扩展性同样不可忽视。 基于 RocketMQ LiteTopic 的这套方案，本质上解决的是几个关键问题： + 海量会话如何隔离。 + 长链路如何保持 Session Sticky。 + 异步结果如何精准回推。 + 临时通道如何自动管理。 + 系统如何在高并发下保持可观测与可扩展。 RocketMQ LiteTopic 提供了一种更适合高并发实时交互场景的消息架构思路。对于正在推进 Agent、实时互动、大模型应用工程化落地的团队来说，尤其是需要支撑海量动态会话、低延迟响应和灵活扩展的业务，这类能力正在从“加分项”逐步变成“必选项”。 欢迎钉钉搜索_（__群号：__110085036316）_或扫码加入 RocketMQ for AI 用户交流群，与我们交流探讨。

引言 随着大模型推理服务成为主流，消息队列在 AI 场景下的精细化流量治理，正面临前所未有的挑战。 传统互联网应用的业务流程固定、请求耗时短，消息队列的限流机制已相对成熟。然而，在 AI 推理场景下，业务流程高度动态、单次任务可持续数分钟甚至更久。这让传统方法显得力不从心，并引发两大核心痛点： + 队列头部阻塞：单个用户的慢任务，会阻塞队列中其他用户的消息处理。 + 并发效率受损：简单粗暴的限流措施，会导致整个系统吞吐量急剧下降。 为解决这些问题，Apache RocketMQ 5.x 版本推出了专为 AI 场景设计的核心特性——轻量主题模型 LiteTopic。它支持百万级轻量主题的创建和高性能动态订阅。基于 LiteTopic 的精细化流量治理方案，既能实现毫秒级的实时限流，又能支持分钟级的忙闲调度，真正做到了“千人千面”的个性化流量治理。 AI 推理场景下的消息队列新挑战 AI 应用与传统互联网应用存在本质差异，在于其执行模式和任务耗时。传统应用流程固定可预测、耗时短（秒级）、多为单向一次性交互；AI 应用更偏主动执行，会自主拆解目标并动态调整策略，流程不确定，单次任务耗时长（分钟级且不可预测），还常伴随多轮对话交互。 这种差异，导致消息队列在 AI 推理场景下面临两大严峻挑战： 1. 队列头部阻塞 传统业务中，不同用户的请求耗时较均衡（通常为秒级），即便多租户共享队列，也不会长期占用队列头部，阻塞问题不明显。因此，只需设置几个队列即可满足需求。 但在 AI 推理场景下，不同用户的请求耗时差异巨大（几秒到几十分钟不等且不可预测）。多租户共享队列时，一条长耗时消息（如复杂推理任务）占据队列头部，会阻塞后续所有消息的处理，导致同队列其他用户的正常消息无法被及时处理。若某个用户密集提交慢任务，可能长期抢占全部队列头部位置，形成资源独占，导致其他用户延迟飙升，破坏系统公平性。 2. 并发效率受损 在 AI 推理场景中，当某个用户短时间内密集提交大量推理请求时，系统需要对该用户实施流量控制。然而，传统的限流措施（如 Thread.sleep()）会阻塞消费者线程，这会导致一个严重的问题： 即使队列中还有其他健康用户的消息等待处理，但由于所有消费线程都在处理限流用户的请求而被阻塞，这些健康用户的正常消息也无法得到处理。随着被限流的用户增多，大量线程陷入阻塞状态，整个系统的并发处理能力将急剧下降。 传统方案为何在 AI 推理场景中失效？ 面对 AI 推理场景的流量洪峰，业界通常采用两种“老套路”来限流，但都“治标不治本”。 ▍方案一：消费失败重试法 简单粗暴地让消息失败，并自动重回队列排队。这听起来似乎很取巧，实则埋下了“定时炸弹”： + 重试机制不可控：依赖中间件内置重试机制，缺乏时间精度控制，易造成延迟放大； + 服务质量不稳定：无法保证时效性，消息可能在队列里躺上好几轮才被处理，影响业务 SLA； + 资源浪费严重：失败重试会消耗额外的网络、磁盘和 CPU 资源，增加系统整体负载，降低系统稳定性。 ▍方案二：线程阻塞限流法 当检测到某个用户短时间内请求频率过高或资源消耗过大时，通过Thread.sleep()等同步阻塞 API 暂停消息处理线程，直接让处理线程“睡一会儿”。这看似控制住了消息处理频率，实则是在“饮鸩止渴”： + 资源利用率低：大量线程被无效阻塞，不仅占用内存，还增加调度开销，导致并发能力下降，长期运行有资源耗尽风险； + 租户隔离失效：在共享线程池中，对某个队列的限流会波及由同一线程处理的其他队列，从而破坏多用户间的隔离性； + 吞吐量受损：阻塞机制与高性能设计的初衷背道而驰，严重损害了系统整体的消息处理能力。 这两种传统方法，要么过度依赖中间件机制，要么牺牲系统性能，都无法从根本上解决多租户环境下的精细化流量控制难题。 RocketMQ LiteTopic 流量治理：千人千面，优雅调度 ▍1. 毫秒级实时限流：让每个用户都有“专属 VIP 通道” AI 推理请求可能在毫秒级内剧烈波动，需要毫秒级的精细化限流能力来应对瞬时流量洪峰。 RocketMQ 基于 LiteTopic 打造了一套精细化限流方案，通过构建完整的资源隔离与调度体系来实现高效的流量治理： + 物理隔离：为每个用户/会话创建独立 LiteTopic，从物理层面实现用户级资源隔离，彻底消除交叉干扰。 + 弹性扩容：LiteTopic 支持百万级规模的按需创建，无论是小批量测试还是大规模生产，都能从容应对。 + 精准流控：每个 LiteTopic 可独立执行限流策略，支持按用户配置差异化阈值，真正实现“千人千面”的个性化流量治理。 + 消费挂起：当检测到用户请求超限时，不是简单地拒绝（失败重试）或等待（阻塞线程），而是优雅地“请用户稍等片刻”（挂起），既保护了系统资源，又不影响用户体验。 在实际应用中，流量处理流程如下图所示： 1. 消息分流：上游业务消息根据用户标识（如 userId）分流到每个独立用户对应的专属 LiteTopic，实现物理隔离。 2. 并行拉取：消费者通过长轮询并行拉取各 LiteTopic 的消息，在限流窗口中对每个 LiteTopic 独立执行限流判断。 3. 限流判断： 未超限：当某用户请求未触发阈值时，正常消费并输出流量； 已超限：当检测到请求超限时，返回 Suspend 挂起状态。 4. 消费挂起：该 LiteTopic 立即挂起，消费者释放处理线程并暂停服务端对该用户的拉取，支持毫秒级精确控制挂起时长，确保限流策略的灵活性和响应速度。 5. 线程复用：释放的线程即时转交其他用户请求，实现资源的弹性调度与高效复用。 6. 自动恢复：挂起的 LiteTopic 将在指定时间后自动恢复消费。 以下消费代码示例展示了如何在实际业务中实现这套机制： ```plain LitePushConsumer litePushConsumer = PROVIDER.newLitePushConsumerBuilder() .setClientConfiguration(clientConfiguration) .bindTopic(TOPIC) .setConsumerGroup(GROUP) .setMessageListener(messageView { //【物理隔离】以userId作为liteTopic名称，实现用户级物理隔离 // 每个用户独享一个独立的物理队列，确保资源完全独立，避免相互干扰 String userId = messageView.getLiteTopic(); //【精准流控】根据业务规则判断是否需要触发限流 // 支持按用户配置差异化阈值，实现"千人千面"的个性化流量治理 if (shouldThrottle(userId)) { //【消费挂起】返回suspend，立即释放当前处理线程 // 服务端暂停对该用户的拉取，避免无效资源消耗 // 支持毫秒级精确控制，100ms后自动重投递，释放的线程可被重新分配给其他用户请求 return ConsumeResultSuspend.of(Duration.ofMillis(100)); } // 正常处理消息 processMessage(messageView); return ConsumeResult.SUCCESS; }) .build(); ``` 上述代码的核心是引入了“消费挂起”机制。 与传统消息队列仅支持“消费成功”与“消费失败”两种状态不同，这里新增了第三种消费状态——Suspend，实现了精准的时间窗口控制： + 状态扩展：消费者返回 ConsumeResultSuspend 状态时，可携带下次可见时间戳，指定消息在时间窗口内的不可见期； + 资源释放：系统立即释放处理线程，清理该队列的本地缓存，避免资源占用； + 自动恢复：服务端维护定时调度器，到达指定时间后自动唤醒队列，重新参与拉取消费。 这一机制让瞬时限流不再阻塞线程，既保护了系统资源，又确保了其他用户请求的正常处理，完美契合 AI 推理场景下的实时流量治理需求。 ▍2. 分钟级忙闲调度：让延迟任务“错峰出行” 除了毫秒级的瞬时流量控制，RocketMQ LiteTopic 的消费挂起机制同样适用于分钟级甚至小时级的长时间窗口调度，实现延迟不敏感任务的错峰调度。 在实际业务场景中，可能存在大量延迟不敏感的任务，如： + 跑批任务：数据统计、报表生成等批量处理作业； + 异步处理：非核心链路的异步通知、日志分析等； + 资源消耗型任务：模型训练、离线推理等计算密集型操作。 这类任务无需实时处理，但可能占用大量计算资源。通过消费挂起机制，我们可以将这些任务智能调度到业务空闲时段执行： 1. 长时间窗口挂起：设置秒级甚至分钟级的挂起时长（如 Duration.ofMinutes(30)），将任务延迟到低峰期处理； 2. 动态感知业务负载：实时监控系统负载，当检测到资源紧张时，主动挂起低优先级任务的消费； 3. 轻量级任务调度：在无需引入额外调度系统的情况下，通过消息队列本身实现任务的延迟执行和资源错峰，降低系统复杂度。 ```plain LitePushConsumer litePushConsumer = PROVIDER.newLitePushConsumerBuilder() .setClientConfiguration(clientConfiguration) .bindTopic(TOPIC) .setConsumerGroup(GROUP) .setMessageListener(messageView { String taskType = messageView.getUserProperty("taskType"); //【忙闲调度】识别延迟不敏感任务 if ("BATCH".equals(taskType) || "LOW_PRIORITY".equals(taskType)) { // 检测系统是否处于繁忙状态 if (isSystemBusy()) { //【长时间挂起】将任务延后到空闲时段处理 // 挂起30分钟后自动恢复，实现错峰调度 return ConsumeResultSuspend.of(Duration.ofMinutes(30)); } } // 正常处理消息 processMessage(messageView); return ConsumeResult.SUCCESS; }) .build(); ``` 这种忙闲调度能力，让 RocketMQ LiteTopic 在消息队列的基础上，扩展了延迟任务处理能力。无需引入额外的调度组件，即可在保障核心业务 SLA 的同时，最大化系统资源利用率。 RocketMQ LiteTopic 技术揭秘：如何实现百万级物理隔离？ LiteTopic 是 Apache RocketMQ 专为 AI 场景设计的轻量主题模型，具备轻量资源、自动化生命周期管理、高性能订阅和顺序性保障等特点。 其底层基于创新的存储架构和分发机制，支撑了百万级 LiteTopic 的高效管理，在不牺牲性能的前提下，实现了海量 LiteTopic 资源的物理隔离，为 AI 场景下的精细化流量治理提供了坚实的技术基础。 关键技术点包括： + 统一存储、多路分发：所有消息数据统一存储在底层 CommitLog 文件中且仅存储一份，采用追加写入模式避免磁盘碎片化，保障极致写入性能。同时，通过多路分发机制为不同 LiteTopic 生成独立的消费索引。 + RocksDB KV 存储引擎：摒弃传统文件型 CQ 结构，替换为高性能的 KV 存储引擎 RocksDB，将队列索引信息和消息物理偏移量作为键值对存储，充分发挥 RocksDB 顺序写入的高性能优势，实现对百万级元数据的高效管理。 + 订阅关系管理：Broker 负责管理消费者的订阅关系集，支持增量更新，能够实时、主动地感知消息与订阅的匹配状态。 + 事件驱动与就绪集维护：每当新消息写入时立即触发订阅匹配，将符合条件的消息聚合到就绪集中。 + 高效批量拉取：消费者只需一次 poll 请求即可批量拉取来自多个 LiteTopic 的消息，显著降低网络交互频率，确保在海量订阅场景下的低延迟与高吞吐。 _百万级 LiteTopic 高并发性能的发送和消费流程_ 结语 随着 AI 推理日益普及，传统消息队列限流方式已难以满足精细化流量控制需求。 基于 RocketMQ LiteTopic 的精细化流量治理方案，通过物理隔离、弹性扩容、精准流控和消费挂起四大核心特性，系统性解决了队列头部阻塞和并发效率受损两大痛点，为 AI 推理场景提供了从毫秒级实时限流到分钟级忙闲调度的全方位消息处理保障，实现了真正意义上的“千人千面”个性化流量治理。 值得一提的是，该方案已与阿里云大模型服务平台百炼网关达成深度合作，利用 RocketMQ LiteTopic 的精细化流控能力，帮助其更好地管理 AI 推理请求的流量峰值与资源调度。 目前，LiteTopic 的核心能力已在阿里云云消息队列 RocketMQ 版 5.x 系列实例中发布，若要在实际业务中使用，请点击下方阅读原文链接查看帮助文档。 未来，我们将继续探索更多创新技术，推动消息队列在 AI 时代的演进与发展。 欢迎钉钉搜索_（群号：110085036316）_或扫码加入 RocketMQ for AI 用户交流群，与我们交流探讨～

在智能汽车产业快速发展的背景下，车联网服务（TSP）已成为主机厂从“硬件制造”向“数据驱动服务”转型的关键引擎。长城汽车_（__https://www.gwm.com.cn__）_正加速从传统汽车制造商向“全球化智能科技公司”转型，以智能网联技术为核心，构建覆盖研发、生产、服务全链条的数字化生态。其云平台战略聚焦“软件定义汽车”，通过云原生技术、分布式架构与数据驱动能力，打造“车路云一体化”的智能出行解决方案。 消息总线作为云平台核心基础设施，承载跨业务异步集成与事件驱动，是支撑复杂业务流程自动化与实时数据交换的关键。随着业务规模与接入系统持续增长，长城汽车对消息总线提出更高的稳定性、可用性与扩展性要求。同时，在业务全球化与合规要求趋严的背景下，多云架构可增强运营韧性，实现资源优化与灵活调度，避免单点故障影响关键业务流程，保障业务连续性与体验一致性。 基于上述诉求，长城汽车对消息总线进行全面升级，核心目标是构建跨云双活能力：在故障场景下快速切换并保持业务连续，同时提升高并发接入下的稳定性与运维效率。本次升级引入阿里云云消息队列 RocketMQ 版的 Global Replicator，实现多云之间消息秒级同步，并结合 Serverless 弹性伸缩进一步增强系统可靠性，为全球车主“永远在线”的智能服务提供更稳固的消息底座。 长城汽车消息总线的核心特点 长城汽车消息总线的设计目标，是构建“消息、事件一体”、“中心、边缘一体”的事件总线平台，核心特点包括： 1. 标准化接入协议（HTTP）：采⽤ HTTP 协议作为统一接入协议，构建标准化的消息⼊⼝和出⼝接⼊点，降低系统接⼊门槛，便于精细化流量管理与控制。 2. 稳定可靠的消息存储组件：选用 Apache RocketMQ 作为消息存储组件，凭借其稳定可靠、高性能与功能丰富等优势，充分满足企业级消息服务需求。 3. 支持高级消息特性：支持顺序消息（按特定顺序消费）与定时/延时消息（按指定时间投递）等能力，满足时间敏感、流程复杂场景的精确控制需求。 4. 集成长城集团云平台周边系统：打通主题创建、消费组配置、权限分配等资源管理与现有工单系统，实现从请求提交到资源分配的全流程⾃动化；对接钉钉通知实现业务通知与告警；对接服务治理平台实现全链路灰度。 5. 跨云高可用部署架构：⽀持跨多云环境双活/多活部署，确保单数据中⼼故障时可⽆缝切换⾄备⽤节点继续运⾏，并通过一致性机制保障业务连续性与数据完整性。 构建跨云双活架构的关键挑战 作为云平台消息中枢，消息总线支撑跨业务实时数据流转，其可靠性直接影响业务连续性和用户体验。为满足核心业务的高可用诉求，跨公有云双活成为关键目标，但在设计与落地过程中主要面临以下挑战： ▍1. 跨云传输实时性与业务容限的权衡 + 网络延迟叠加：跨公有云通常依赖公网传输，端到端延迟显著增加；叠加多云环境下的跨地域距离与同步协议开销，总延迟可能突破业务容忍阈值。 + 一致性代价：为保障双活集群数据强一致性，需引入额外的同步机制，会进一步加剧延迟。 ▍2. 混合云环境的兼容性与安全性挑战 + 版本与协议兼容性：现有自建 RocketMQ 4.x 集群存在深度定制，引入云上托管 RocketMQ 5.x 服务以降低运维复杂度，需要兼容开源 Apache RocketMQ 4.x 和云上托管服务 RocketMQ 版本。 + 多云安全隔离：跨云消息同步链路需加密传输与访问鉴权（如基于 VPC 对等连接的流量隔离）。 ▍3. 特殊消息类型的跨云一致性保障 + 顺序消息：如流水单、订单状态变更等场景，要求消息严格按 Key 分组并有序消费。跨云同步需确保同一分组消息不乱序（如阿里云集群主节点故障时，其他云备节点接管且不破坏顺序）。 + 延时消息：如营销活动定时通知等场景，依赖精确的时间控制。跨云同步需保证延时触发时间在毫秒级误差范围内，避免业务逻辑错乱。 ▍4. 成本与高可用性的平衡难题 双活部署需要在两朵云上独立部署完整集群（包括 Broker、NameServer、存储节点等）来保障高可用性，基础资源与运维成本接近翻倍。 长城汽车消息总线跨云双活方案 长城汽车消息总线跨云双活架构要点如下： + 消息总线基于其他云和阿里云跨云部署，通过专线通信确保网络可靠性。 + 管理服务部署在其他云，与消息总线服务解耦，避免管理服务故障影响消息总线运行。 + 跨云消息同步采用云消息队列 RocketMQ 版的 Global Replicator 实现秒级数据同步。 + 基于动态 DNS 实现双活节点流量按自定义比例分配，并在单云故障时支持一键切流。 ▍1. 双活与容灾能力 采用其他云自建 RocketMQ 与 阿里云云消息队列 RocketMQ 版构建多云双活架构，云消息队列 RocketMQ 版提供全球消息备份的容灾能力。 + 消息数据一致性：两地消息全量互备，数据可靠性更高；重试策略可在⽹络分割等极端场景下确保数据⼀致性和完整性；同步策略与备份方式可灵活配置，降低开发成本；内置消息过滤机制，避免消息在跨云传输过程中重复复制。 + 服务可用性：消息服务提供两地容灾能力，服务可用性更高，业务恢复更快，延续性更强。 + 高级消息支持：顺序消息按顺序复制，保障顺序语义；延时消息在源集群对消费者可见后（已到延时时间）再复制到目标集群，保障延时语义，消费端可⽴即消费。 + 同步能力弹性可扩展：Global Replicator 同步链路可弹性扩展，以满足低延时同步要求。 + 流量自定义分配：动态 DNS 支持灵活分配双活节点流量，并可结合健康检测自动切换。 ▍2. 版本兼容 + 云消息队列 RocketMQ 版 5.x 系列兼容开源 RocketMQ 4.9 SDK，业务逻辑无需改造；在收发可靠性与多副本存储方面提供保障，并提供弹性规格以应对突发流量。 + 服务可用性：自建集群缺少 SLA 保障，故障恢复依赖自运维。而云消息队列 RocketMQ 版天然支持多可用区部署，具备同城容灾能力，服务可用性最高可达 99.99%。 + 管控适配：云消息队列 RocketMQ 版提供标准管控 API 与可观测数据，便于与消息总线进行管控与运维集成。 ▍3. 高级特性消息 云消息队列 RocketMQ 版全球消息备份能力，在传输过程中保障源集群数据语义。 + 顺序消息：同步到目标集群时保持与写入源集群的顺序一致。 + 定时消息：以“源集群消息对消费者可见”为同步触发条件。 ▍4. 降本增效 汽车行业流量波动明显，云消息队列 RocketMQ 版 5.x Serverless 系列可根据实时负载自动弹性伸缩、按量付费，无需预估和配置实例规格。相比“按峰值预留并叠加冗余”的方式，可显著降低资源闲置成本。 消息总线全面升级的关键价值 ▍1. 能力升级：面向全球业务的消息底座 + 技术领先性：依托云消息队列 RocketMQ 版千万级 TPS 吞吐与毫秒级低延迟，构建跨云多活架构的车联网消息平台。通过“多地域集群 + 逻辑 Topic 分区”实现车辆数据就近接入与跨云无缝路由，突破传统架构单云单点的瓶颈，支撑全球化业务布局。智能流量调度跨域传输延迟降低 30% 以上。 + 架构先进性：云消息队列 RocketMQ 版 5.0 采用云原生架构（计算存储分离、无状态代理层），实现资源弹性伸缩与故障秒级隔离。结合 Serverless 化部署，提升扩容效率与资源利用率，支撑突发流量场景（如大规模 OTA 推送）平稳运行。 ▍2. 稳定可靠：多云互联下的全链路容灾 面对服务商级网络中断等极端场景，基于云消息队列 RocketMQ 版的跨云、跨地域的多活容灾体系，通过三级容灾防护实现“零数据丢失、零感知切换”的高可用： + 同城双活：基于阿里云多可用区（AZ）部署，RPO=0、RTO + 跨云灾备：跨云异步复制，保障核心业务数据跨地域冗余； + 智能故障自愈：通过流量染色与灰度路由自动隔离异常节点，结合 AIOps 预测潜在风险，故障恢复时间缩短至分钟级。 ▍3. 弹性降本：Serverless 系列按需弹性 借助云消息队列 RocketMQ 版 Serverless 系列，实现“按量付费 + 弹性容量”的轻量化运维： + 成本直降 50%+：按实际吞吐计费，闲时资源自动释放，降低资源与运维成本； + 敏捷创新：开发人员通过 API 分钟级接入消息服务，无需关注底层基础设施，新功能上线周期缩短 20%。 重塑车联网服务边界，驱动产业智能升级 长城汽车车联网 TSP 平台的跨云多活升级，不仅是技术架构的迭代，更是对“用户价值优先”理念的践行。借助阿里云云消息队列 RocketMQ 版，长城汽车构建了高可靠、高性能、高性价比的全球车联网服务基座，为未来 V2X 协同与个性化用户服务奠定坚实基础。 面向智能汽车竞争的“下半场”，长城汽车将持续以技术领先定义行业标准，让每一辆车成为万物互联世界中最可靠的智能节点，与全球合作伙伴共建车联网新生态。

核桃编程是青少年编程教育行业的领先企业。自 2017 年 8 月成立以来，核桃编程通过打造智能实操产品与服务矩阵，发展成为了包含编程系列产品、编程硬件、赛级展服务、素质延展产品及数字出版物的多元化公司。在落实科学教育加法的实践之路上，核桃编程致力于提高青少年的科学素养，激发他们对学习的热爱，并以此培养未来科技创新人才。 随着业务规模快速增长，平台对“精准调度、金融级可靠性、极致并发”的要求显著提升。如何为千万学员提供稳定、流畅且公平的在线学习体验，成为技术团队的核心课题。 1. 在线考试的精准调度难 在线考试涉及组卷、开考、防作弊检测、阅卷、成绩发布等多环节，需严格按时间节点触发。传统调度方式在复杂场景下难以实现精准、高效触发，可能影响学员体验与教学公平性。 2. 交易链路的状态一致性风险 课程购买、退款等核心链路对请求处理顺序与状态一致性要求极高，需要确保交易过程可靠、安全。随着业务规模扩大，业务系统对交易处理的有序性和最终一致性要求进一步提高。 3. 直播互动流量洪峰难应对 在直播高峰期，直播课的答题、弹幕、课件同步等互动功能会产生瞬时激增的消息量。传统消息服务在弹性扩展与资源利用率方面仍有优化空间，难以高效应对流量洪峰。 技术破局：阿里云 RocketMQ 构建“数智底座”核心引擎 在对可靠性、性能、可扩展性等多个维度进行深度评估后，核桃编程选择阿里云云消息队列 RocketMQ 版作为核心消息中枢，并通过关键能力逐一破解难题： 1. 延迟消息：让考试流程拥有“智能时钟” 将“收卷后 5 分钟启动阅卷”等关键环节封装为延迟消息，由 RocketMQ 定时精准投递。由此告别轮询调度，实现全流程自动化与零人工干预，显著提升阅卷效率和考务处理效率。 2. 顺序消息：为交易链路加上金融级“原子锁” 在支付、退款等关键操作中启用 RocketMQ 顺序消息，确保同一用户请求严格串行处理。结合分布式事务能力，实现“扣款→开课→通知”链路最终一致性，保障交易过程安全可靠。 3. 广播消息 + 弹性架构：直播互动的“稳定器” 直播辅助指令通过广播消息触达网关实例，确保课件同步与互动指令全域生效。同时，依托 RocketMQ 的流量削峰能力，平滑承接瞬时百万级消息洪峰，保障直播体验稳定、流畅。 _云消息队列 RocketMQ 版产品架构图_ 方案优势：充分释放云原生红利，运维与成本双重优化 1. 弹性伸缩，优化资源使用效率 采用阿里云 RocketMQ 按需使用、按量付费的模式与自动扩缩容能力，课中高峰秒级扩容保障稳定性，课后低谷自动缩容避免资源浪费。相比常备冗余服务器的传统模式，有效提升资源利用率、降低消息服务成本。 2. 全托管服务，提升技术团队效能 阿里云 RocketMQ 全托管服务提供高可用保障、跨可用区容灾、实时监控告警，降低技术团队对基础设施运维工作的投入，更聚焦教学产品创新，提升技术团队效能。 3. 精细化成本管控，提升成本管理效率 通过消息类型智能选型（如非关键场景用普通消息替代广播消息）、流量分时调度等方式，进一步优化资源消耗，让消息服务支出更清晰可控，有效提升成本管理效率。 业务价值：技术驱动体验升级与创新加速 核桃编程与阿里云 RocketMQ 的深度合作，带来多维度价值提升： + 学员体验升级：考试流程零延迟精准触发、直播互动毫秒级响应，学习体验更稳定顺畅，用户满意度提升显著； + 业务安全加固：交易链路顺序与一致性更有保障，实现金融级别的安全可靠，进一步夯实业务安全与用户信任； + 业务创新加速：构建稳定可靠的消息底座，为 AI 学情分析、个性化推荐等新场景快速创新落地提供坚实支撑。 核桃编程与阿里云 RocketMQ 的合作，是教育科技与云原生技术深度融合、推动业务高质量发展的最佳实践。从考试自动化到直播高并发，从成本优化到运维提效，阿里云 RocketMQ 以“精准、可靠、弹性”的核心能力，为核桃编程业务稳定运行与持续创新提供有力支撑。 未来，双方将持续探索消息技术在实时学情反馈、AI 互动教学等场景的创新应用。阿里云亦将携手更多教育企业，以云原生基础设施助力教育数字化升级与高质量发展。

AI 助成长：「亲宝宝 APP」千万 MAU 下的架构挑战 亲宝宝是一家专注于家庭育儿领域的移动互联网公司，其核心产品「亲宝宝 APP」聚焦性化育儿服务，集成长记录、育儿知识、早教内容、家庭共享、智能推荐及 AI 育儿助手等功能于一体，致力于打造一个围绕儿童成长的家庭私密社交与育儿服务平台。 自 2012 年成立以来，亲宝宝注册用户总数已突破一亿，月活跃用户（MAU）超千万，日均上传照片/视频数量达数百万条，平台沉淀了海量的用户行为数据和成长内容数据。其技术架构需要支撑高并发写入、实时消息触达、个性化推荐、数据一致性保障等复杂场景，对底层中间件系统提出了极高要求。 高并发、强一致性与实时触达的三重压力 随着用户规模持续增长，亲宝宝面临三大核心挑战： 1. 高频写入与异步处理压力 用户每日上传海量成长影像，需在保证体验的同时完成缩略图生成、AI 标签识别、多端同步等后处理任务，传统同步调用链路难以支撑。 2. 跨设备实时通知的可靠性要求 家庭成员间的新动态（如“爸爸上传了宝宝照片”）需在秒级内精准触达所有关联成员，且不能丢失或重复。 3. 分布式事务场景下的数据一致性难题 如用户完成任务获得积分、兑换权益等操作，涉及账户、订单、通知等多个微服务，必须保障“操作成功则消息必发”，否则将导致用户权益异常。 面对上述挑战，亲宝宝亟需一个高吞吐、低延迟、支持事务语义、具备完善可观测性的消息基础设施。 为什么选择阿里云 RocketMQ 5.x？ 经过多轮技术评估，亲宝宝最终选择全面迁移至阿里云云消息队列 RocketMQ 版 5.x Serverless 系列。 核心原因如下： 1. Serverless 架构实现客户端轻量化 RocketMQ 5.x Serverless 通过引入独立的 Proxy 组件，将原本内嵌于客户端的路由、协议解析、重试等逻辑下沉至服务端，客户端仅需极简 SDK 即可完成消息收发。该架构不仅提升了系统的可维护性与安全性，也大幅降低了移动端的网络与内存开销，完美适配亲宝宝高并发、低功耗的终端环境。 2. 秒级精准延迟消息 RocketMQ 5.x Serverless 支持高精度延迟消息，通过秒级延迟消息实现“未读通知二次触达”、“临时草稿自动清理”、“成长里程碑倒计时提醒”等柔性业务逻辑，在提升用户体验的同时优化系统资源利用率。 3. 全链路可观测性 RocketMQ 5.x Serverless与阿里云 ARMS、SLS 等可观测产品深度集成，提供了从生产到消费的全链路消息轨迹追踪、消费延迟告警、堆积分析等运维闭环，极大简化运维工作，显著提升故障定位效率。 4. 云原生弹性伸缩与成本效益 亲宝宝的业务流量具有显著的“节日效应”，每逢春节、六一儿童节、开学季等高峰期，用户上传照片量可激增 3–5 倍，家庭通知消息峰值可达平日的 4 倍。过去自建 RocketMQ 集群需提前数周预估容量并手动扩容，成本高昂且难以精准预估偏差，导致资源浪费或服务降级。基于 RocketMQ 5.x Serverless，亲宝宝实现了真正的按需付费与秒级自动弹性伸缩，从容应对流量洪峰，同时大幅优化了资源成本。 核心应用场景与 RocketMQ 5.x 落地实践 ▍场景一：成长相册——高吞吐的异步处理流水线 当用户上传照片后，前端服务仅需完成元数据落库，并立即向 Topic_Photo_Process 发送一条普通消息。后端多个独立消费者组并行消费，分别执行各自负责的异步任务，如：图像压缩与多尺寸生成、AI 模型打标（如“笑脸”、“户外”等）、家庭成员推送通知、写入搜索索引等。得益于 RocketMQ 5.x Serverless 的百万级 TPS 吞吐能力与批量消费优化，整条处理流水线延迟稳定在 200ms 以内，系统资源开销降低 40%。 ▍场景二：成长印迹定时解锁——高精度的延迟消息应用 当用户为宝宝设置“时光信件”（如“18 岁生日开启”）或重要纪念日（如“百天纪念”）倒数提醒时，业务系统只需向 Topic_Growth_Reminder 发送一条延迟消息，延迟时间可精确到秒，跨度可从几分钟到数年。RocketMQ 5.x 服务端内置的高精度定时调度能力，确保消息在预定时刻被准时唤醒并投递。该方案极大简化了定时任务的实现，避免了传统数据库轮询带来的性能损耗与架构复杂性，为用户提供了温暖而可靠的长期约定功能。 ▍场景三：积分权益——强一致的事务消息保障 在用户完成“每日签到”等任务时，系统需同时完成“更新任务状态”和“发放积分/徽章”等操作。亲宝宝采用 RocketMQ 5.x 的事务消息机制来保障最终一致性，核心流程如下： 1. 应用发起本地事务（扣减任务状态）； 2. 若成功，则向 RocketMQ 提交一条“半消息”； 3. RocketMQ 回查本地状态，确认后将已提交的消息投递至 Topic_Reward_Delivery； 4. 下游服务消费消息，完成发放徽章并触发 Push 通知。 该方案在亲宝宝过去一年的生产环境中，实现了事务消息成功率高达 99.999%，达成了积分权益业务的“零资损”目标。 成效与价值 通过全面采用阿里云 RocketMQ 5.x Serverless，亲宝宝在技术与业务层面均获得了显著收益： 更重要的是，RocketMQ 5.x 的 Serverless 架构将复杂逻辑下沉至服务端 Proxy，提供的轻量化 SDK 显著降低了亲宝宝移动端的网络开销与内存占用，为亿级用户的流畅 App 体验提供了坚实保障。 未来展望 AI 时代下，亲宝宝与阿里云消息团队紧密合作，积极探索 RocketMQ 5.x 在 AI 场景下的更多前沿能力： + 使用 RocketMQ LiteTopic，打造 AI 场景下 MultiAgent 的异步通信，解决长耗时调用阻塞痛点。 + 采用“会话即主题”——会话独占 LiteTopic，基于状态持久化机制，保障了会话的连续性和完整性，提升了会话用户体验，减少了会话需求重试成本。 + 利用 RocketMQ 优先级消息，实现算力资源最大价值分配，保障高优先级任务的资源分配。

最近，“千问请全国人民喝奶茶”活动火爆全网，这种瞬时爆发的流量洪峰已成为新茶饮行业的常态化挑战。新茶饮行业的数字化演进已从最初的基础设施上云，演进为深度的云原生架构共创与能力共建，再到为 AI 原生提供确定性基座，古茗奶茶在阿里云云原生上的深度实践，正是这种演进的代表。 在新茶饮行业，每一次刷屏级的营销活动，每一杯奶茶的“丝滑”下单，背后都是对数字化基座的严峻考验，是一场应对瞬时高并发流量的技术硬仗。 作为拥有超万家门店的行业头部品牌，古茗不仅要支撑海量日常订单，更需在“周三会员日”等大促时刻，从容应对流量陡增，确保系统稳如磐石。面对高并发下的极速响应与弹性需求，古茗如何实现“大促自由”？ 本期《云故事探索》栏目走进古茗，揭秘支撑新茶饮“万店时代”的云原生力量。 从口味之争到体验之战，技术成为新茶饮竞争力 “如今，一杯奶茶的竞争已不仅限于口味。”古茗科技技术运维负责人刘星光表示，在新茶饮这条日趋激烈的赛道上，“口味决定品牌的记忆度，但真正拉开差距的，是门店高峰期的稳定体验、新品迭代的速度，以及消费者触达的精准度。” 对于古茗而言，数字化的核心价值并非上线了多少系统，而是打通了供应链、门店与营销等环节，以数据驱动决策，让成功的运营模式能在全国范围内快速复制。 这意味着技术团队的角色已从“系统维护者”升级为“业务赋能者”，不仅要保障系统稳定运行，更要支撑业务的高速增长与敏捷创新。 _古茗科技 技术运维负责人 刘星光_ 架构升级：微服务+DevOps，实现业务敏捷与体验统一 为支撑万店扩张与高频营销，古茗构建了以“微服务 + DevOps”为核心的云原生架构。订单、会员、库存、营销等核心业务被拆分为独立微服务，可独立开发、部署与扩缩容。其中，阿里云微服务引擎 MSE 作为服务注册与配置中心，在保障系统高可用的同时，也让古茗更聚焦业务研发。 架构升级带来的直接收益是迭代速度显著提升。刘星光表示：“一个新的优惠策略，如今可在数天内完成验证并上线，实现快速试错、快速复制。”2025 年，古茗完成底层架构的全面云原生升级，确保全国用户下单体验的一致性。 但微服务化也带来了调用链路复杂、峰值压力集中等挑战。要在流量洪峰下保持系统稳定，“异步解耦”与“流量削峰”成为关键，这正是消息队列的核心价值。 大促自由：RocketMQ Serverless 稳定可靠、弹性降本 每周三“会员日”，古茗中午 12 点的瞬时订单量可达平日数倍。传统架构下，需提前数天甚至数周预估流量、规划资源并手动扩容，不仅耗时费力，还伴随着稳定性风险与资源浪费。 在支付、营销、库存等核心链路中，古茗引入了阿里云云消息队列 RocketMQ 版 Serverless 系列，精准解决了三大痛点： 1. 极致弹性，告别容量焦虑与资源浪费 面对十万级 TPS 的瞬时并发请求，RocketMQ Serverless 无需人工干预即可秒级自动扩容，保障消息高吞吐、低延迟、不丢失、不积压，并在峰值结束后自动释放资源，真正实现按需使用、按量付费。据测算，该方案帮助古茗节省超 40% 成本。 2. 事务消息，保障业务数据最终一致性 在“支付成功后扣减库存并发放优惠券”等场景，数据一致性至关重要。RocketMQ 事务消息确保支付主流程与下游操作的最终一致性。即使下游服务短暂异常，可靠的重试机制也能保证业务最终成功，从根源上避免因数据不一致导致的资损与客诉风险。 3. 稳定可靠，让技术团队聚焦业务创新 RocketMQ 历经阿里巴巴十余年“双十一”万亿级数据洪峰验证，具备稳定可靠的 SLA 保障，并提供消息过滤、顺序消息等功能及完善的可观测工具，帮助古茗技术团队从繁琐的维稳工作中解放出来，更专注于业务创新。会员日由此成为业务增长的“加速器”，而非技术压力的“爆发点”。 _RocketMQ Serverless 架构及弹性示意图_ “拥抱云原生后，我们终于可以放手策划大规模活动了。”刘星光的话语中透露出十足的底气，“以前最怕系统崩溃，现在我们只需关心活动玩法能否打动用户。”这份底气，正源于以 RocketMQ Serverless 为代表的阿里云原生技术栈。 稳定第一：全链路可观测，让风险“可见可控” “稳定，永远是第一位的。”刘星光反复强调，“第一是稳定，第二是效率，第三是成本。” 为保障稳定性，古茗基于阿里云日志服务 SLS、应用实时监控服务 ARMS 等产品，构建了覆盖底层基础设施到上层业务逻辑的全链路可观测体系，实现多维度监控与实时告警，全面掌握系统状态。 刘星光表示：“任何一笔异常订单（如支付或领券失败），我们都能通过全链路追踪，在分钟乃至秒级内定位根因，从而快速修复，保障用户体验。” 从工具采纳到能力共建，从云原生迈向 AI 原生 古茗与阿里云的合作，已从工具采纳深化为场景共创（如优化事务消息延迟）与能力共建（如增强消息轨迹）。古茗真实的业务场景（如节假日大促、爆款联名发布）成为 RocketMQ Serverless 等阿里云产品的“极限压测场景”与“最佳实践样板”；阿里云则将经过古茗验证的架构模式产品化，赋能更多零售客户，形成相互成就、共同成长的深度伙伴关系。 面向未来，古茗正积极探索 AI 与业务的深度融合，包括智能推荐、经营分析、AIGC 营销等。他们的思路清晰而坚定：并非“从云原生切换到 AI 原生”，而是在云原生基础上，将 AI 能力逐步叠加，让技术架构与业务共同演进。 “云原生解决了弹性、稳定和标准化的问题，这恰恰是 AI 大规模落地的前提。”刘星光总结道，“只有底座足够稳，AI 才能真正服务于业务，而不是制造新的复杂性。” 一杯奶茶，一场深刻的技术革命 从一杯奶茶的“丝滑”下单，到一场大促的从容应对，古茗的故事是新茶饮数字化转型的缩影，也是云原生技术释放业务潜能的证明：新消费品牌的护城河，正在从产品和供应链向技术深度延伸。 以云消息队列 RocketMQ 版为代表的阿里云云原生产品，正凭借其极致弹性、高稳定性和领先技术，帮助像古茗这类高速发展的企业卸下技术包袱，在激烈的市场竞争中轻装上阵，将更多精力聚焦于业务创新，让“下单丝滑，大促自由”成为新常态。 未来，随着云原生与 AI 的进一步融合，每一杯奶茶的背后，都将蕴藏着一个更智能、更高效、更稳定的数字世界。

12 月 19 日，由 InfoQ 极客传媒与模力工场联合发起的“2025 中国技术力量榜单”评选结果正式揭晓，Apache RocketMQ 凭借其在 AI 时代的创新性突破——面向 AI 应用的事件驱动架构解决方案，从众多参选项目中脱颖而出，成功斩获“AI 开源明星项目”权威奖项。该奖项标志着业界对 Apache RocketMQ 从传统消息中间件向 AI 时代消息引擎演进的技术领导力与行业影响力的高度认可。 随着 AI 技术重塑应用架构，传统的“服务连接”模式正向“智能协同”跃迁，对底层通信基础设施提出了前所未有的挑战。为精准应对这一范式转变，Apache RocketMQ 前瞻性地完成了战略升级，进化为专为 AI 时代打造的消息引擎。其以轻量级通信模型 LiteTopic 为核心的创新特性，为海量长时会话（Session）、多智能体（MultiAgent）系统及大规模 AI 任务调度等场景提供了高效、可靠的事件驱动架构解决方案。 Apache RocketMQ for AI 核心价值解读 1. 多智能体异步通信，破解协同难题 针对多智能体应用中普遍存在的长耗时调用阻塞和协作扩展性问题，RocketMQ 的 LiteTopic 模型以其百万级轻量资源创建、自动化生命周期管理、细粒度订阅管理及顺序性保障，为 Agent 之间提供了高效、有序的异步通信机制。 2. 智能任务调度，最大化 AI 算力价值 面对稀缺的 AI 算力，Apache RocketMQ 作为前端请求与后端算力服务之间的缓冲层，通过流量整形平滑请求洪峰，通过消息优先级将宝贵算力优先分配给高价值任务，并通过消费者限流保障核心服务的稳定性，实现算力价值最大化。 3. 无状态、高可靠的分布式会话管理 Apache RocketMQ 动态为每个会话创建专属队列（LiteTopic），以连续消息流完整保存上下文，从而实现上层应用的“无状态化”，极大简化开发。通过顺序保障与排他消费机制，它能严格确保会话上下文的完整性与一致性，并以极低成本实现了生产级的会话续传与恢复，同时原生支持 AI 场景下的大规模数据负载传输。 目前，Apache RocketMQ for AI 的核心特性已在阿里云云消息队列 RocketMQ 版产品中发布，并在阿里巴巴集团内部，以及阿里云大模型服务平台百炼、通义灵码等产品中经过了大规模生产环境验证，展现出卓越的成熟度与可靠性。 值得一提的是，Apache RocketMQ 与本次同获“AI 开源明星项目”的阿里巴巴开源智能体开发框架 AgentScope 深度集成，联合打造企业级、高可靠的 A2A（AgenttoAgent）智能体通信基座，为开发者构建复杂多智能体应用提供了开箱即用的解决方案。 我们相信，开放与协作是推动 AI 技术普惠的基石。Apache RocketMQ for AI 的部分核心代码已在社区开源，我们诚邀全球开发者体验、交流与共建。 + 项目地址：基于 RocketMQ 实现的 A2A 通信 RocketMQTransport 部分代码现已开源 https://github.com/apache/rocketmqa2a + 免费体验：“通过 RocketMQ 实现多智能体异步通信” https://www.aliyun.com/solution/techsolution/rocketmqformultiagentcommunication + 社区交流：欢迎钉钉扫码或搜索群号 110085036316，加入 RocketMQ for AI 用户交流群。 展望未来，Apache RocketMQ 社区将持续深耕 AI 领域，与更多生态伙伴携手，共建智能时代的数字新基建，并将更多经过验证的优秀方案回馈给开源社区。

引言 Agentic AI 时代已至，在智能客服、代码生成、流程自动化等场景中，多智能体（MultiAgent）协作正从构想走向落地。然而，当多个 Agent 需要像一个团队那样高效协作时，脆弱的通信机制可能因网络抖动或服务宕机，就让整个系统瞬间瘫痪，导致昂贵的计算任务失败、会话状态丢失。 如何为这些聪明的“数字员工”们构建一个真正可靠、高效的通信基座？ 本文将为您介绍 Apache RocketMQ 全新推出的轻量级通信模型 LiteTopic，如何在 AI 应用场景中有效简化系统架构、提升稳定性与可靠性，并结合 A2A（AgenttoAgent）协议与阿里巴巴 AgentScope 框架的生产实践案例，深入剖析面向智能体通信的落地实践与技术实现。 RocketMQ for AI：重新定义 AI 应用通信范式 1.1 传统应用：单向、无反馈的事件驱动模式 在传统应用的事件驱动场景中，业务逻辑编排通常由人工预先约定，消息生产方成功发送消息后，便无需关注后续的处理逻辑。 下图以注册系统为例：用户发起账户注册请求后，注册系统向 RocketMQ 发送“新用户注册”的消息后便立即返回，无需关心下游的邮件或短信通知系统如何处理。邮件或短信通知系统再分别从 RocketMQ 拉取消息，驱动各自的发送流程。整条业务链路为单向、无反馈的事件驱动模式。 1.2 从单向事件到双向交互：AI 应用对通信提出新挑战 在 AI 应用场景中，业务逻辑编排通常由大模型动态生成，消息生产方需等待并处理响应结果，才能驱动后续的逻辑执行。 下图以典型的 AI 会话场景为例：用户所连接的 Gateway 不仅需要发送请求，还需要处理推理响应结果，并将结果推送给浏览器，形成完整的交互闭环。 结合真实 AI 应用场景的深度调研，我们发现 AI 场景具有四个显著特征，对底层通信模式提出了全新且严苛的挑战： + 更长的响应时间：传统互联网应用追求毫秒级响应延时，而 AI 应用的响应时长普遍达到分钟级以上。更关键的是，AI 应用单次业务的运行时间具有高度不可预测性。 + 更复杂的交互：AI 应用的多轮对话持续时间长，对话历史可达数十轮甚至更多。单次上下文传输可能达到几十甚至上百 MB，上下文管理难度高。多 Agent 之间的协同编排逻辑更加复杂，需要精确的状态同步。 + 更昂贵的计算资源：AI 推理依赖昂贵的 GPU 资源，瞬时高并发流量可能冲击推理服务稳定性，导致算力资源浪费，并且任务失败重试的成本极高。 + 更精细化的事件驱动：由于计算能力有限，异步事件驱动需要更精准的消费速度控制。同时，必须实现分级的事件驱动策略，以确保高优先级任务优先获得宝贵的计算资源。 1.3 RocketMQ LiteTopic：专为 AI 场景设计的通信模型 为了应对上述挑战，Apache RocketMQ 推出了以轻量级通信模型 LiteTopic 为核心的一系列新特性： + 轻量级通信模型 —— 为海量会话而生 其核心是百万级轻量资源管理能力。基于极低的资源动态创建开销，可轻松支持海量会话（Session）场景，并提供更细粒度的订阅管理，适用于长时 Session、AI 工作流和 AgenttoAgent 交互等场景。 + 企业级上下文管理 —— 让会话状态可靠持久 以连续的消息流完整保存 Session 上下文，通过顺序保障、排他消费等机制严格确保上下文的完整性与一致性。同时原生支持大消息体（数十 MB 甚至更大），轻松满足 AI 场景下庞大数据负载的传输需求。 1.4 LiteTopic 技术解析：百万队列支撑海量并发会话 LiteTopic 基于 RocketMQ 业界领先的百万队列核心技术构建，其底层本质是独立的 Queue。 + 它为每个独立会话（Session）创建一个专属的、低成本的“私有通道”——即轻量主题（LiteTopic），从而能够以极低的资源开销支撑海量并发会话的需求。 + 轻量级的 LiteTopic 在消息分配与发送行为上与顺序 Topic 一致（其所属 Queue 由单一 Broker 独占，消息始终路由至该 Broker，而非在多个 Broker 间轮询发送），这种设计天然确保了消息的严格顺序性，并极大降低了资源管理和路由的复杂度。 1.4.1 LiteConsumer 支持单节点粒度的订阅关系管理 与传统消息队列中“同一 Consumer Group ID（CID）必须全局一致订阅相同 Topic”的强约束不同，LiteConsumer 创新性地支持 CID 内各节点按需进行差异化订阅。每个节点可根据实际负载、业务场景或运行时需求，独立订阅不同的 LiteTopic，从而构建更加灵活、弹性的消费拓扑。 这一机制从根本上规避了因订阅关系不一致所引发的消费异常、重复消费或 Rebalance 风暴等问题，显著提升了系统的灵活性、可扩展性与稳定性。同时，它更契合 AI 时代轻量、动态、点对点的交互模式，为构建轻量级请求响应式消息收发模型提供了原生支持。 1.4.2 LiteConsumer 的核心能力 + 多节点差异化订阅：同一 CID 下的不同节点可独立订阅各自的 LiteTopic，实现细粒度、个性化的订阅策略。 + 动态订阅扩展：支持在运行时实时为单个节点新增 LiteTopic 订阅，无需重启服务或影响其他节点的正常消费。 + 动态退订能力：支持在运行时实时取消单个节点对特定 LiteTopic 的订阅，实现精准的资源释放与流量治理。 1.5 生产案例：RocketMQ LiteTopic 如何重塑 AI 应用架构？ 以下案例基于某客户真实的 AI 应用场景，通过架构对比直观展示采用传统 RocketMQ 通信模型与引入 LiteTopic 轻量级通信模型前后的显著差异。 采用 RocketMQ LiteTopic 轻量级通信模型后，客户架构实现了质的提升：不仅彻底移除了对 Redis 的依赖，还避免了广播推送带来的带宽与计算资源浪费。整体架构更轻量，系统稳定性与可靠性也得到显著提升。 1.5.1 改造前：依赖 Redis + 广播的臃肿架构 整体的业务流程步骤如下： 1. 任务提交：用户请求到达后，应用接入层节点将推理任务写入 Redis。 2. 任务处理：Worker 集群扫描 Redis 并处理推理任务，将推理过程中的中间结果以多条顺序消息的形式发送至 RocketMQ。 3. 结果持久化与通知：Consumer 集群顺序消费 RocketMQ 消息，将最终推理结果存入 Redis，并基于 RocketMQ 广播通知所有应用接入层节点。 4. 结果推送：应用接入层节点收到广播消息后，仅当结果归属于自身连接时，才从 Redis 获取完整结果并推送给客户端；否则直接忽略该消息。 传统架构采用“先存储、再广播、后过滤”的模式，在高并发 AI 场景下效率低下且成本高昂： + 架构臃肿且脆弱：强依赖外部组件 Redis，增加了系统的复杂度和潜在故障点，运维成本高，可用性受限。 + 资源浪费严重：无效的广播机制导致大量带宽被占用，且每个应用接入层节点都需进行计算密集型的过滤操作。 + 链路冗长低效：数据流转需多次读写 Redis，通信链路长、延迟高，应用接入层节点宕机后会话状态将全部丢失，严重影响用户体验。 1.5.2 改造后：基于 RocketMQ LiteTopic 的极简可靠架构 引入 LiteTopic 后，业务流程被大幅简化，实现了端到端的可靠、高效通信： 1. 会话绑定与动态订阅：应用接入层节点在发起推理请求时携带唯一身份标识（如 Session ID），并立即订阅该标识对应的 LiteTopic（无需预创建 consumer group、topic）。 2. 结果持久化发送：智能应用（Worker）根据请求中的身份标识，将推理结果直接发送至对应的 LiteTopic（同样无需预创建）。 3. 精准接收消费：应用接入层节点各自精准接收属于自己的 response 消息，无需过滤，无任何冗余消费。 1.5.3 核心价值：为 AI 会话注入“记忆”，实现断点续传与恢复 客户接入 LiteTopic 轻量级通信模型后，通过将 LiteTopic 与 Session 维度进行细粒度绑定，以极低成本实现了生产级的会话续传与恢复能力。 在按照上一小节的流程实现端到端的可靠通信后，在网关机器下线/宕机时： + 自动重连：客户端检测到连接断开后，自动发起重连请求。 + 动态订阅：新接管的应用接入层节点实例根据 Session ID，动态订阅原 session 对应的 LiteTopic（无需预创建）。 + 断点续传：新应用接入层节点从上次成功消费的 Offset 位点开始拉取消息，精准恢复到故障前的状态（不会丢消息，也不会重复消费已处理的消息）。 + 恢复会话：自动恢复 Session 的完整上下文，用户完全无感知，业务流程无缝衔接。 基于 RocketMQ LiteTopic 打造企业级 Session 管理 2.1 AI 场景下 Session 的四大核心要求 在 AI 应用场景下，业界对 Session 的特性提出了以下四项核心要求： + 低延迟：面向实时交互场景，要求快速响应。 + 时序性：必须严格按对话时间顺序组织内容，确保上下文的连续性与逻辑一致性。 + 单会话隔离：保障不同用户/会话间的数据隔离，避免消息串话或状态混淆。 + 上下文压缩：支持通过截断或摘要控制上下文长度，避免超出模型窗口限制导致溢出。 2.2 RocketMQ LiteTopic 实现 Session 的四大优势 基于 RocketMQ LiteTopic 实现 Session 的核心价值，在于将“Session”从内存易失状态转化为可持久、可追溯、可恢复的事件流，为多智能体系统提供企业级会话韧性，彻底解决传统架构中会话状态丢失、无法恢复等痛点。 1. 会话状态持久化 —— 进程重启不丢会话 消息天然持久化存储于 CommitLog，即使应用宕机或网络中断，也能通过消息重放完整重建会话上下文（如对话历史、任务状态、中间结果）。 如下图所示，应用 A 将响应输出的 TaskEvent / TaskUpdateEvent 转换为 RocketMQ LiteTopic 中存储的消息（Message）。当应用 A 重启后，可从 CommitLog 中重放所有消息，完整恢复会话状态。 2. 消息回溯与重放 —— 断点精准恢复 支持按时间 / Offset 回溯消费，应用重启后可从断点精确恢复会话，实现无缝续聊与任务接力，避免重复推理带来的算力浪费。 当应用宕机后重新启动，可以指定某个 Session（LiteTopic）中的具体位点开始继续消费，或从上次消费成功的位点开始消费。 3. Session 隔离与路由 —— 多会话并行无干扰 通过轻量级 LiteTopic 实现会话级隔离（如 Session ID 作为 LiteTopic 的唯一标识），确保多用户/多会话并行运行时互不干扰。 多用户多 Session 的消息存储于不同的 LiteTopic，在数据存储维度实现天然隔离，无需应用层手动过滤。 4. 流量削峰与缓冲 —— 保护下游应用稳定性 高并发会话请求被缓冲至 Broker，避免下游 Agent 瞬时过载崩溃，提升系统整体稳定性。下游应用根据自身处理能力按需消费消息，实现“削峰填谷”。 如下图所示，应用 A 发出的任务请求可在 Broker 中持久化堆积，下游应用 B 根据自身消费能力按需拉取并处理，有效保障系统稳定性。 基于 RocketMQ 构建高可靠 A2A 通信通道 在上一章，我们解决了单个会话的持久化与恢复问题。现在，让我们将视野放大：当成百上千个功能各异的 Agent 需要协作时，它们之间如何建立标准化的通信？这正是 A2A 协议诞生的意义所在。 3.1 A2A 协议 AgenttoAgent（简称 A2A）是一项由 Google 于 2025 年发起，并贡献至 Linux 基金会的开源通信协议。其核心目标是建立跨厂商、跨框架的标准化互操作机制，使异构 AI 智能体（Agents）能够自动发现、可靠通信并高效协作，从而构建开放、可组合、可扩展的多智能体系统生态。 3.2 单智能体 vs. 多智能体架构：能力边界与协同范式的演进 在深入探讨如何构建 A2A 通信之前，我们首先需要理解，为什么多智能体协同是必然趋势。我们从六个维度对比单智能体与多智能体的能力差异： 3.3 同步 RPC 与 RocketMQ 异步通信的对比 明确了多智能体架构的优势后，下一个关键问题是：如何实现 Agent 之间的通信？ A2A 协议原生支持的同步 RPC 协议包括 JSONRPC、gRPC 和 REST。然而，在企业级的复杂场景下，这些同步协议面临诸多挑战。下表从多个维度对比同步 RPC 与 RocketMQ 异步通信模型的差异： 3.4 开箱即用：基于 RocketMQ 的 A2A 协议实现 为加速 A2A 协议在异步通信场景的落地，我们基于 RocketMQ SDK 实现了 A2A 协议的 ClientTransport 接口。该实现旨在帮助用户在搭建多智能体应用时，能够专注于自身业务逻辑，快速构建高可靠、开箱即用的 A2A 通信方案。 发送普通同步请求：EventKind sendMessage(MessageSendParams request, @Nullable ClientCallContext context)发送Stream请求：void sendMessageStreaming(MessageSendParams request, Consumer eventConsumer…)重订订阅任务数据：void resubscribe(TaskIdParams request, Consumer eventConsumer, Consumer errorConsumer查询任务完成状态：Task getTask(TaskQueryParams request, @Nullable ClientCallContext context)取消任务执行Task cancelTask(TaskIdParams request, @Nullable ClientCallContext context)以及其他方法 开源项目地址 基于 RocketMQ 实现的 A2A 通信 RocketMQTransport 部分代码现已开源，欢迎关注！ 项目地址：_https://github.com/apache/rocketmqa2a_ 3.5 架构解析：如何通过 RocketMQ 实现 Agent 间通信？ 在一个典型的多智能体协作架构中，通信流程如下： + 应用 A 扮演 Supervisor 角色，负责对用户输入的需求进行任务分解，并将拆分后的子任务分别发送至应用 B 的业务 Topic（Normal Topic1）和应用 C 的业务 Topic（Normal Topic2）。 + 应用 B 集群从 Normal Topic1 拉取消息并执行相应逻辑处理，随后将结果发布到应用 A 订阅的 LiteTopic。 + 应用 C 集群则从 Normal Topic2 拉取消息进行处理，并同样将结果写入该 LiteTopic。 + 应用 A 集群通过拉取 LiteTopic 中的消息，汇聚各子任务的响应结果，进而驱动后续的业务逻辑编排。 AgentScope × RocketMQ：构建多智能体应用的最佳组合 理论与架构已经铺垫完毕，接下来，让我们结合一个完整的实战案例，看看如何将这套强大的通信基座，与顶尖的智能体开发框架 AgentScope 相结合，构建一个真正可用的多智能体应用。 4.1 AgentScope：面向多智能体的开发者友好框架 AgentScope 是阿里巴巴继 AI 模型社区 ModelScope 后，在 Agent 领域的又一战略级开源力作。它以“开发者为中心”，专注于提供智能体开发的开源框架，为构建复杂的多智能体应用提供了从设计、开发到调试的全套解决方案。它具备以下核心优势： + 对开发者透明：拒绝隐式魔法，所有环节（提示、API、智能体、工作流）可见、可控。 + 实时可介入：原生支持运行时中断与自定义处理。 + 更智能：内置工具管理、长期记忆、智能 RAG 等能力。 + 模型无关：一次编写，无缝适配各类大模型。 + 乐高式构建：模块化设计，组件解耦、自由组合。 + 面向多智能体：显式消息传递与工作流编排，专为协作场景打造。 + 高度可定制：全面开放工具、提示、智能体、工作流及可视化扩展，鼓励深度定制。 4.2 AgentScope x RocketMQ 的集成架构与合作展望 在明确了 AgentScope 的功能定位与应用价值之后，我们将进一步探讨其通信层与 RocketMQ 的现有集成机制，并展望双方在技术协同与生态共建方面的未来合作方向。 4.2.1 AgentScope 与 RocketMQ 集成架构 当 AgentScope 作为 Agent 应用服务提供者时，其内部支持符合 A2A（AgenttoAgent）协议的多种通信方式，包括基于 JSONRPC 的 WebService 和 RocketMQ Service，用于接收并处理来自其他 Agent 的 A2A 协议请求。同时，AgentScope 通过 wellknown 服务接口向外标准化地透出其所承载 Agent 的核心能力信息，包括但不限于： name（名称） description（描述） capabilities（能力列表） additionalInterfaces（额外支持的接口或协议） 这些元数据使调用方能够清晰识别该 Agent 提供的主要功能、所支持的通信协议及其对应的接入方式。 当 AgentScope 作为 Agent 应用服务的调用者时，它首先通过访问目标 Agent 暴露的 wellknown 服务，动态获取其详细的能力描述、支持的协议类型及对应的服务接入点（如 JSONRPC 端点或 RocketMQ Topic 信息）。随后，在通信层，AgentScope 利用 A2A 协议定义的传输客户端（如 JSONRPCTransport 或 RocketMQTransport）发起请求，并对返回的响应结果进行统一解析与处理，从而实现跨 Agent 的标准化、可互操作的协同调用。 1. 基于 RocketMQ 协议通信架构图 2. 基于 JSONRPC 协议通信架构图 4.2.2 合作展望 随着人工智能与分布式系统技术的深度融合，消息中间件与智能体（Agent）平台的协同正成为构建下一代智能分布式应用的关键路径。作为 Apache 软件基金会顶级项目，RocketMQ 凭借高吞吐、低延迟和高可靠等核心特性，已成为全球广泛采用的分布式消息队列，在金融、电商、物联网等关键领域积累了深厚的技术积淀，并于近期推出了轻量级通信模型 LiteTopic，进一步拓展了其在 AI 应用场景中的适用性。与此同时，AgentScope 作为新兴的智能体编排与运行平台，专注于为多智能体系统提供统一的调度、通信与治理能力。二者在技术理念与应用场景上高度契合，展现出广阔的合作前景与协同创新潜力。 1. 技术互补：构建“消息驱动 + 智能决策”的新型架构 RocketMQ 提供了强大的异步通信、事件驱动和流式处理能力。AgentScope 则聚焦于智能体生命周期管理、任务分解、上下文感知与自主协作。未来，二者可深度融合，形成“消息即事件、事件触发智能体行为”的新型架构： + 智能体间通信的标准化通道：利用 RocketMQ 作为 AgentScope 内部或跨集群智能体之间的可靠通信总线，保障高并发、有序、可追溯的消息传递，提升多智能体系统的鲁棒性与扩展性。 2. 生态共建：推动开源与标准协同发展 双方可基于开源社区开展深度合作： + 集成适配器开发：共同开发 RocketMQ 与 AgentScope 的官方集成插件，简化开发者接入流程。 + 联合参考架构发布：推出面向典型场景（如智能客服等场景）的联合解决方案模板，加速行业落地。 + 参与标准制定：在事件驱动架构（EDA）、智能体通信协议等领域协同推进开放标准，促进生态互操作性。 4.3 场景案例：用 AgentScope 与 RocketMQ 打造“智能旅行助手” 本案例以 AgentScope 作为 AI 智能体应用开发框架，构建了三个智能体： + SupervisorAgent（总控）：负责与用户交互，任务分解与逻辑编排。 + WeatherAgent（天气专家）：负责查询天气信息。 + TravelAgent（旅行专家）：负责依据天气进行用户的行程规划。 SupervisorAgent 应用具有如下逻辑： + 如果用户只查询天气情况，则直接请求 WeatherAgent 进行天气信息查询； + 如果用户希望做出行程规划，则先向 WeatherAgent 发出天气查询请求，获取对应天气信息后，再带着天气信息向 TravelAgent 发出行程规划请求，TravelAgent 对行程结果进行规划后将响应结果发送至 SupervisorAgent 订阅的 LiteTopic，SupervisorAgent 应用将结果发送至用户侧。 实战演练：三步构建高可靠多智能体应用 阿里云官网现已提供免费试用、一键部署的《》，带您亲手搭建并运行一个多智能体应用，并基于 RocketMQ LiteTopic 实现多智能体异步通信能力——具备持久化、高可靠、可追溯等特性，显著提升 AI 应用交互的稳定性与可观测性。 5.1 方案概览：技术架构与云资源 本方案将带领您搭建一个多智能体（MultiAgent）系统，能够根据用户的需求查询天气信息并制定行程规划。 为简化部署过程，我们将在 1 台云服务器 ECS 上部署 3 个独立的 Agent（SupervisorAgent，WeatherAgent 和 TravelAgent，具体功能可参考 4.3），并且通过 RocketMQ 消息服务实现 Agent 之间的异步通信。 本方案的技术架构包含构建一个完整多智能体应用所需的所有云资源： 5.2 三步体验：从创建资源到部署 Agent 1. 免费一键部署资源 访问体验方案页面，点击“免费试用”，进入实验操作界面后，点击“立即试用”即可领取免费试用点，自动开始创建资源。 2. 创建 Topic 和 Group 共创建 3 个 Topic，配置参数见下表，其余参数保持默认。 共创建 3 个 Group，配置参数见下表，其余参数保持默认。 3. 创建部署智能体应用 在阿里云百炼的应用管理页面，根据示例文档中提供的模型参数和提示词，分别创建并发布两个智能体应用（天气助手 Agent、行程助手 Agent）。 远程连接云服务器 ECS 根据提供的执行脚本部署示例应用程序。等待应用启动完毕，大约需要 3~5 分钟，直到终端显示 You 提示符，便可直接在终端中输入信息与智能体交互。 5.3 结果验证：任务执行与消息轨迹追踪 1. 在 You 提示符后，输入 帮我做一个下周三到下周日杭州周边自驾游方案 并回车。 2. 等待智能体执行任务，最终会返回结合天气信息的行程规划内容，过程如下： a. SupervisorAgent 接收用户输入，向消息队列发送一条消息 杭州下周三到周日的天气情况怎么样?。 b. WeatherAgent 监听到上述消息，执行天气查询，并将结果发往消息队列。 c. SupervisorAgent 监听到上述消息，获取了天气查询结果，然后向消息队列发送一条消息 杭州下周三至周日天气已知，天气为，请基于此制定一份从杭州出发的周边2人3天4晚自驾游行程规划（下周三出发，周日返回），包含住宿、餐饮与景点推荐。 d. TravelAgent 监听到上述消息，执行行程规划，并将结果发往消息队列。 3. 查看消息轨迹：在云消息队列 RocketMQ 版实例详情页，可以按 Topic 或按 LiteTopic 查询到相关的消息轨迹。 目前，该解决方案已在阿里云官网上线，欢迎点击即可部署体验～ 邀请您钉钉扫码加入 RocketMQ for AI 用户交流群，探索更多产品功能与应用场景，与我们共建 AI MQ 的未来！

前言 在现代 AI 应用中，多智能体（MultiAgent）系统已成为解决复杂问题的关键架构。然而，随着智能体数量增多和任务复杂度提升，传统的同步通信模式逐渐暴露出级联阻塞、资源利用率低和可扩展性差等瓶颈。为应对这些挑战，RocketMQ for AI 提供了面向 AI 场景的异步通信解决方案，通过事件驱动架构实现智能体间的高效协作。本文将探讨和演示如何利用 RocketMQ 构建一个高效、可靠且可扩展的多智能体系统，以解决企业级 AI 应用中的核心通信难题。 多智能体系统的通信需求与核心挑战 随着 AI 应用的复杂度不断提升，单智能体（AI Agent）因其能力边界和知识局限，已难以独立胜任动态、多维度的决策任务。因此，多智能体（MultiAgent）系统正迅速成为构建复杂 AI 应用的核心范式。MultiAgent 系统通常由一个主智能体（Supervisor Agent）负责将复杂任务分解，并分发给多个具备特定领域能力的子智能体（Sub Agent）并行执行，最终汇聚结果以达成共同目标。 整个系统的智能与效能，高度依赖于智能体间通信的效率与可靠性。为了实现不同厂商、不同技术栈开发的智能体高效协作，行业需要为它们建立一套标准化的“交互协议”与“工作流程”，例如 Google 提出的 A2A（AgenttoAgent）协议。然而，底层的通信模式仍是决定系统性能、可靠性和成本效率的关键。传统的同步调用模式在简单的“一对一”交互中尚可应对，但在 MultiAgent 系统这种涉及多个长周期任务并行协作的复杂场景下，其弊端逐渐凸显，主要体现为三大核心挑战： 1. 同步阻塞与性能瓶颈：在同步调用模式下，主智能体分发任务后必须等待子智能体返回执行结果，才能继续下一步规划。在包含多个长耗时任务的复杂链路中，这极易引发“级联阻塞”，严重限制了系统的并发处理能力和整体吞吐量，导致协作效率低下，系统难以扩展。 2. 系统可用性挑战：同步通信的强依赖特性，使得智能体间的调用关系如同“串联电路”，且通常缺乏可靠的重试与容错机制。任何一个智能体节点的故障或超时，都可能导致整个任务链路中断。任务失败不仅影响用户体验，还会造成中间过程消耗的宝贵算力资源被浪费。 3. 消费调度与成本效率困境：MultiAgent 系统中，上下游智能体的吞吐量差异巨大，任务负载也常出现波峰波谷。若缺乏精细化的流量控制与差异化调度能力，流量洪峰可能导致部分智能体服务过载甚至“雪崩”。同时，在算力资源有限的情况下，系统无法保证高优任务被优先处理，难以实现算力利用率的最大化，最终陷入“忙时过载、闲时浪费”的资源困境。 这些挑战共同制约了多智能体系统的性能、可靠性与成本效率，成为阻碍复杂 AI 应用规模化落地的重要因素。 RocketMQ for AI：构建智能体高效协作的异步通信引擎 要解决上述挑战，核心在于将系统架构从“请求响应（RequestReply）”的同步调用模式，转变为基于事件驱动的异步通信模式。RocketMQ for AI 通过一系列专为 AI 场景设计的特性，为多智能体系统的可靠通信与高效协作构建了一个强大的异步通信引擎。 1. 异步通信，提升协作扩展性：在异步通信模式下，主智能体将任务作为“消息”发送至消息队列后，便可立即返回处理其他工作，无需等待子智能体处理和反馈；子智能体作为“消费者”独立地从队列中获取任务并进行处理。这种“发布订阅”模式彻底消除了级联阻塞，使主智能体可以轻松地向多个子智能体并发分发任务，极大提升了协作效率与系统吞吐量，缩短了复杂任务的端到端时长。RocketMQ 专为 AI 场景推出的轻量主题模型（LiteTopic），支持百万级轻量资源与高性能动态订阅，为系统的动态扩展提供了坚实基础。 2. 持久化与重试机制，提升系统可用性：异步解耦打破了智能体间的调用强依赖，显著提升了系统整体可用性。RocketMQ 将智能体通信的请求和结果均持久化到消息队列，这相当于为任务处理流程提供了 checkpoint 能力。即使某个智能体服务短暂宕机或网络故障，任务消息也不会丢失，待服务恢复后可继续处理。结合 RocketMQ 内置的可靠重试与死信队列机制，可以确保任务最终成功交付，避免因瞬时故障导致整个任务链路失败和算力资源浪费，极大提升了系统的韧性和可用性。 3. 精细化调度，保障稳定性与优化成本效率：面对稀缺且昂贵的 AI 算力资源，RocketMQ 提供了丰富的消息调度策略，以实现成本与效率的最优平衡。通过控制消息的消费速率，可以对任务请求进行缓冲，起到“削峰填谷”的作用，防止下游智能体被突发流量冲垮，保护服务稳定性。通过优先级队列，可以确保在有限的算力资源下，高优先级任务能够被智能体优先处理，实现资源利用率的最大化。 场景实践：通过 RocketMQ 实现 MultiAgent 系统异步通信 下图展示了一个基于 RocketMQ LiteTopic 实现的多智能体异步通信的典型流程，包含一个主智能体（Supervisor Agent）和两个子智能体（SubAgent）。 1. 接收请求阶段：为每个 Sub Agent 创建一个 Topic 作为请求任务的缓冲队列。 2. 返回结果阶段： a. 为 Supervisor Agent 创建一个用于接收响应结果的 Topic，并让其订阅这个 Response Topic。该 Topic 可采用 RocketMQ 专为 AI 场景新发布的 Lite Topic 类型； b. 当 SubAgent 完成任务后，它会将结果发送至该 Response Topic，可以为每个独立任务动态创建一个专属的子 LiteTopic（例如，以任务 ID 或问题 ID 命名）； c. Supervisor Agent 通过 MQ 的异步通知机制实时获取这些子 LiteTopic 中的结果，并可通过 HTTP SSE（ServerSent Events）等协议推送给 Web 端。 场景示例： 现在，我们通过一个具体的天气查询与行程规划 MultiAgent 系统实例，展示如何利用 RocketMQ 实现智能体间的异步通信与高效协作。 1. 方案架构 为简化 MultiAgent 系统的部署过程，我们将在 1 台云服务器 ECS 上部署 3 个独立的 Agent—— 1 个主智能体（Supervisor Agent）、一个负责天气查询的子智能体（Weather Agent） 和一个负责行程规划的子智能体（TravelAgent），并且通过云消息队列 RocketMQ 版实现 Agent 之间的异步通信。 2. 实施步骤 a. 创建资源： i. 创建专有网络 VPC（为云服务器 ECS 等云资源构建云上私有网络）、云服务器 ECS（用于部署 MultiAgent 系统）、云消息队列 RocketMQ 版（提供消息队列服务，实现 Agent 之间的异步通信）。 ii. 在云消息队列 RocketMQ 版实例下创建 3 个 Topic：WeatherAgentTask（普通消息，用于 WeatherAgent 接收任务消息）、TravelAgentTask（普通消息，用于 TravelAgent 接收任务消息），WorkerAgentResponse（轻量消息，用于 SupervisorAgent 接收各个子 Agent 返回的任务结果）。 iii. 在云消息队列 RocketMQ 版实例下创建 3 个 Group：WeatherAgentTaskConsumerGroup（消费模式 CLUSTERING，并发投递，用于消费 WeatherAgentTask 的普通消息）、TravelAgentTaskConsumerGroup（消费模式 CLUSTERING，并发投递，用于消费 TravelAgentTask 的普通消息）、WorkerAgentResponseConsumerGroup（消费模式 LITE_SELECTIVE，顺序投递，用于消费 WorkerAgentResponse 的轻量消息）。 b. 创建智能体应用： i. 开通大模型服务平台百炼（用于调用模型服务），并获取百炼 API Key。 ii. 在百炼的应用管理页面，根据示例文档中（在此不详细展开）提供的模型参数和提示词，分别创建并发布两个智能体应用（天气助手 Agent、行程助手 Agent）。 c. 部署智能体应用：远程连接云服务器 ECS 根据提供的执行脚本部署示例应用程序。等待应用启动完毕，大约需要 3~5 分钟，直到终端显示 You 提示符，便可直接在终端中输入信息与智能体交互。 3. 效果验证 a. 输入帮我做一个下周三到下周日杭州周边自驾游方案。 b. 等待智能体执行任务，最终会返回结合天气信息的行程规划内容，过程如下： i. Supervisor Agent 接收用户输入，向消息队列发送一条消息杭州下周三到周日的天气情况怎么样?。 ii. Weather Agent 监听到上述消息，执行天气查询，并将结果发往消息队列。 iii.Supervisor Agent 监听到上述消息，获取了天气查询结果，然后向消息队列发送一条消息杭州下周三至周日天气已知，天气为，请基于此制定一份从杭州出发的周边2人3天4晚自驾游行程规划（下周三出发，周日返回），包含住宿、餐饮与景点推荐。 iv. Travel Agent 监听到上述消息，执行行程规划，并将结果发往消息队列。 v. Supervisor Agent 监听到上述消息，获取了行程规划结果并返回给用户。 c. 查看消息轨迹：在云消息队列 RocketMQ 版实例详情页，可以按 Topic 或按 LiteTopic 查询到相关的消息轨迹。 目前，该解决方案已在阿里云官网上线，欢迎点击即可部署体验～ 邀请您钉钉扫码加入 RocketMQ for AI 用户交流群，探索更多产品功能与应用场景，与我们共建 AI MQ 的未来！