随着大语言模型（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 的未来！

前言 当企业想用大模型和内部非公开信息打造智能问答系统时，RAG（RetrievalAugmented Generation，检索增强生成）已成为必备技术。然而，在实际落地中，构建 RAG 应用的数据准备过程繁琐复杂且充满挑战，让很多企业和开发者望而却步。本文将介绍构建 RAG 的最佳实践：通过阿里云事件总线 EventBridge 提供的多源 RAG 处理方案，基于事件驱动架构为企业 AI 应用打造高效、可靠、自动化的数据管道，轻松解决 RAG 数据处理难题。 为什么 RAG 是治愈模型幻觉的“良方”？ 大语言模型（LLM）就像一个博览群书、记忆力超群的“学霸”，尽管文采斐然、对答如流，但偶尔也会犯一些令人啼笑皆非的错误，比如凭空编造事实或提供过时信息，这就是我们常说的“模型幻觉”。 这背后的原因很简单：这位“学霸”的知识完全来自于“毕业前”学过的海量教材（即训练数据），尽管覆盖了维基百科、新闻、书籍等通用知识和各领域的专业知识，但存在两个天然局限： + 知识领域局限：它对企业内部、垂直领域等私域知识知之甚少。比如，它不了解你公司内部的规章制度，也无法接触电商平台的用户数据等非公开信息。 + 知识时效局限：它的知识更新停留在训练数据截止的那个时间点，无法获取实时信息，比如股票行情、时事新闻等不断更新的动态数据。 为了治好大语言模型“一本正经胡说八道”的毛病，我们必须让它从“闭卷考试”升级为“开卷考试”，RAG（RetrievalAugmented Generation，检索增强生成）技术应运而生。 RAG 的核心理念，可以通俗地理解为“先查找资料，再生成答案”。当收到一个问题时，它不会让大模型直接凭记忆回答问题，而是分两步走： 1. 检索（Retrieval）：从一个可随时更新的外部知识库（如企业内部文档、产品手册等）中，快速检索出与问题最相关的信息片段。 2. 生成（Generation）：将检索到的信息片段连同用户问题一起作为上下文提供给大模型，引导它基于这些可靠的“证据”生成准确、有理有据且可追溯来源的回答。 _（来自阿里云大模型平台服务百炼 知识库功能 文档示例）_ 通过这种方式，不仅能有效减少模型幻觉，大幅提升生成答案的准确性与时效性，还让模型在无需耗费巨资和时间进行重新训练的情况下，就能轻松扩展知识边界。凭借这些显著优势，RAG 已成为企业构建可靠、智能 AI 应用的首选方案。 RAG 落地挑战：数据处理的“三重困境” 尽管 RAG 的原理听起来简单明了，但在实际落地时，无数企业和开发者却深陷数据处理的泥潭。 AI 时代的数据处理，与过去以结构化数据为主的传统数据处理模式截然不同。我们面对的是由海量、异构、多模态数据构成的洪流，数据处理的复杂度和挑战呈指数级增长。企业对实时性要求也不断提高，任何数据延迟都可能影响模型效果。 企业和开发者在落地 RAG 时，普遍会陷入数据处理的“三重困境”： 1. 扩展之困——异构化数据源的“接入鸿沟” 现代企业的数据通常散落在 ERP、CRM、OA、IoT 设备、社交媒体等数十个系统中，涵盖结构化数据（如数据库、表格）、非结构化数据（如 PDF、网页、图片、音视频）和半结构化数据（如 JSON、XML）。若采用传统点对点连接的数据集成方式，每接入一个新数据源，都需要复杂的定制化开发，扩展性极差，响应速度慢，严重拖慢 AI 应用的迭代速度。 2. 运维之难——脆弱数据管道的“运维噩梦” RAG 的数据处理链路漫长且复杂，涉及数据采集、清洗、切块、向量化、入库、检索等多个环节。整条链路如同一个脆弱的“黑箱”，任何一个环节的微小故障都可能导致全链路瘫痪。在实际运维过程中，数据源接口变更、数据质量问题、系统负载突增等突发状况层出不穷，数据管道的问题排查、修复和系统更新，都极其耗时耗力，让运维团队疲于奔命。 3. 稳定之痛——数据管道的“可靠性危机” 数据管道的稳定性是 AI 应用落地的基石。数据丢失、重复、延迟、质量下降以及系统故障等数据处理链路中的任何问题，都可能直接导致模型推理结果的偏差甚至错误，进而影响业务决策和用户体验。传统数据处理架构的紧耦合设计，导致任何一个组件故障都可能影响整个系统运行，并且缺乏有效的监控和告警机制，往往在造成严重影响后才发现问题。 因此，我们迫切需要一种全新的数据处理范式，来构建一个灵活、可扩展、实时、智能的数据处理管道。 破局之道：事件架构驱动重塑 AI 数据管道 事件驱动架构（EventDriven Architecture，EDA）为应对 AI 数据处理的复杂性挑战，提供了坚实的技术基础。在事件驱动架构中，“事件（Event）”是核心概念，它本质上是一次状态变化的数字化表达。在 AI 数据处理场景中，数据的产生、变更、处理、存储等各个环节都可以被抽象为事件。例如，当新的训练数据上传到系统时，产生数据接收事件；当数据经过清洗和转换后，产生数据处理完成事件；当向量化处理完成后，产生向量生成事件；当数据成功存储到向量数据库后，产生数据入库事件。 这种“事件化”的处理方式，使整个 AI 数据处理流程变得标准化、清晰、可控且可追溯，带来三大优势： 1. 松耦合 数据处理流程被分解为独立的事件和处理单元。数据工程、算法、平台等团队可以独立开发、部署和迭代各自负责的组件，无需关心对方的内部实现。一个组件的变更不会影响其他部分，系统容错能力和迭代效率更高。 2. 可扩展性与稳定性 每个组件都可以根据实际负载独立扩展，当某个组件成为瓶颈时，只需增加该组件的实例数量，而无需对整个系统进行扩容。同时，通过引入智能监控和自动恢复机制，系统能够及时发现和处理各种异常情况，保证数据链路稳定运行。 3. 端到端实时性 在智能客服、实时推荐等场景中，毫秒级的响应至关重要。事件驱动架构可以确保事件一旦发生，便能被立即捕获并触发后续处理。这使得 RAG 的知识库能够近乎实时地吸收新信息，让大模型始终掌握着最新“情报”。 综上所述，采用事件驱动架构的系统在敏捷性、可扩展性和可靠性方面实现了质的飞跃，这正是 AI 应用规模化落地的基石。 EventBridge 多源 RAG 处理方案：为 AI 场景提供高效数据管道 阿里云事件总线 EventBridge 基于事件驱动架构，将 AI 能力深度融入数据处理全链路，为企业和开发者提供专为 AI 应用设计的、端到端的、智能化的数据处理中间件。 EventBridge 通过一系列 ETL for AI Data 的全新能力，提供多源 RAG 处理方案：将 RAG 数据准备的全流程（从多源异构数据提取、清洗、切块、向量化再到入库）彻底实现自动化。 开发者现在可以通过 EventBridge 简单的“白屏化”配置，轻松实现： 1. 无缝对接多源数据 轻松接入主流的对象存储（OSS）、消息队列（如 Kafka、RocketMQ、MQTT）、日志服务（如 SLS）、数据库服务（如 MySQL）等多种数据源，覆盖结构化数据（如数据库、表格）、非结构化数据（如 PDF、网页、图片、音视频）和半结构化数据（如 JSON、XML）。 2. 智能化的数据处理 自动完成文档解析（Loader）、文本切分（Chunking）和向量化（Embedding）的完整数据转换流程，内置多种核心技术，支持多种非结构化数据（如 TEXT、JSON、XML、YAML、CSV）的智能解析和处理，提供完整的 Loader 技术体系，包括多种分块策略、单文档加载、批量数据加载，确保大规模数据的可靠处理；对结构化数据采用流式处理架构，能够实时处理高吞吐量的数据流，可实现复杂的流式数据转换和聚合操作。 3. 一键式向量入库 提供统一的向量数据库接入接口，支持将处理好的向量数据直接加载到主流向量数据库（如 DashVector、Milvus）中，也兼容传统数据库的向量扩展插件。只需简单的图形界面配置（拖拽方式配置数据源、处理逻辑、目标数据库等），系统会自动生成复杂的向量数据处理和入库流程。提供丰富的预置模板，可基于模板快速搭建数据处理流程。提供完善的监控仪表板和告警机制，可实时查看数据处理的状态、性能指标、错误信息等，及时发现和解决问题。 场景实践：从 0 到 1 构建基于事件驱动架构的实时 RAG 应用 接下来，我们将通过一个完整的实战场景，带你从零开始，利用阿里云事件总线 EventBridge、对象存储 OSS、函数计算 FC、向量检索服务 DashVector 和大模型服务平台百炼，快速构建一个实时的 RAG 应用。 方案概览 + 首先，通过 EventBridge 构建一个高效的 ETL 数据管道：能够自动从数据源（对象存储 OSS）中实时提取数据，通过函数计算 FC 灵活定义数据转换的逻辑，进行清洗、切块和向量化，并将处理结果持续加载到目标（向量检索服务 DashVector），形成一个动态更新的知识库。 + 然后，通过函数计算（FC）的 Web 函数构建一个简单的 RAG 应用，调用大模型服务平台百炼进行推理，以 DashVector 中的向量数据作为知识库。 + 最后，我们通过输入与知识库相关的用户问题，测试 RAG 应用的回答效果。 方案架构 方案提供的默认设置完成部署后，在阿里云上搭建的系统如下图所示。实际部署时您可以根据资源规划修改部分设置，但最终形成的运行环境与下图相似。 实施步骤 1. 构建自动化数据管道： 1. 创建事件流：在事件总线 EventBridge 控制台创建并配置一个事件流，作为数据处理管道的核心。 2. 配置数据源与目标：创建并配置对象存储 OSS Bucket 作为数据源（Source），创建并配置向量检索服务 DashVector 作为数据投递的目标（Sink）。 3. 配置数据转换逻辑（Transform）：选择“内容向量化”的函数模板创建一个函数，并在函数代码中填写获取的百炼 APIKEY，这个函数将负责对数据进行切块和向量化。 2. 构建 RAG 应用： 1. 创建 Web 函数：创建一个 Web 函数（注意和之前创建的用于处理数据流的事件函数区分）。 2. 编写应用代码：这个函数将作为 RAG 应用的后端，负责接收用户查询，从 DashVector 检索知识，并调用百炼大模型生成回答。需要在函数代码中配置百炼和向量检索服务 DashVector 的相关访问凭证（如 APIKEY、Endpoint 等）。 3. 部署应用：部署代码成功后，RAG 应用即构建完成并可供访问。 效果验证 1. 更新知识库：将包含私有数据的文件（例如，一份名为百炼系列手机产品介绍.txt 的文档，包含了虚拟手机厂商的商品数据）上传到 OSS Bucket 中。 2. 查看向量生成：文件上传成功后，EventBridge 会自动捕获这一事件并触发数据处理流程。稍等片刻，即可在 DashVector 控制台查看已生成的向量。 3. 测试问答效果：通过创建的 RAG 应用发起访问，输入一个与你上传文档相关的问题，例如：“百炼 X1 手机的分辨率是多少？”。 4. 获取精准回答：RAG 应用会自动检索知识库，并将相关信息连同问题一起发送给百炼大模型。很快就会收到一个基于私有数据生成的精准回答。在函数的执行日志中，还可以看到向量检索召回的具体原文片段，从而验证整个 RAG 链路的有效性。 目前，该解决方案已在阿里云官网上线，欢迎点击即可部署体验～ 邀请您钉钉扫码加入 EventBridge 用户交流群，探索更多产品功能，与我们共同定义和构建 AI 数据处理的未来！