作者 | 阿里云消息团队文婷、不铭、墨岭、稚柳前言随着 AIGC生成式人工智能浪潮席卷全球大语言模型LLM正在深刻重塑千行百业、重构应用开发范式。这场由模型与算法驱动的技术革命带来了前所未有的机遇也为开发者构建 AI 应用带来了全新而严峻的工程挑战如何保障长耗时对话的连续性如何公平高效地调度有限的算力资源如何避免多 AI Agent 或复杂工作流的级联阻塞问题…这些挑战的核心诉求在于我们需要一种可靠且高效的异步通信机制来支撑应用、数据与模型之间的协同交互。作为分布式系统不可或缺的基础组件Apache RocketMQ 在微服务异步解耦与数据流处理等方面表现出色。在 AI 时代如何应对复杂多变的业务场景、满足更高的性能与体验要求已成为 Apache RocketMQ 演进过程中的关键课题。挑战显现传统消息队列在 AI 场景中的局限性在传统分布式架构中消息队列作为实现异步解耦、流量削峰及数据流处理的成熟方案其可靠性已得到广泛验证。然而随着 AI 应用在交互模式、资源形态和应用架构上的根本性变革如果客户采用同步阻塞架构、或者基于传统消息队列的异步化架构都会面临很多新挑战。交互模式从“请求 - 响应”到“长时会话”传统应用的交互模式一般是无状态短平快的请求 - 响应模式一个用户请求会在毫秒级返回结果如收藏商品、加购物车、下单等场景。而 AI 应用交互如多轮对话多模态具有持续时间长单次推理可达数秒至分钟级、多轮次上下文依赖对话历史可达数十轮、计算资源消耗大等特征。现有的 AI 应用若采用 HTTP 长连接、 WebSocket 等协议结合后端同步阻塞架构极易因为网络抖动、网关重启或连接超时等偶发问题导致上下文丢失、推理任务中断造成不可逆的算力浪费和用户体验的损害。资源形态从“通用服务器”到“稀缺算力”AI 推理依赖昂贵的 GPU 资源瞬时高并发流量可能冲击推理服务稳定性导致算力资源浪费。传统消息队列虽能实现流量削峰填谷但在多租户共享资源池场景下由于缺乏精细的消费流量控制机制难以实现精细化、差异化的资源调度导致资源利用率低下。应用架构从“服务调用”到“智能体协作”AI Agent 或多步工作流本质上是长周期任务的协同。若采用同步调用机制任何单节点阻塞都可能引发整个任务链级联失败。因此需要一个高效、可靠的异步通信枢纽来连接这些独立且长时间运行的智能体或任务节点实现非阻塞协同保障分布式智能系统稳定运行。此外传统消息队列还面临其他挑战如在处理 AI 多模态等大负载时因传统消息队列对消息大小有更严格的限制需要采取繁琐的变通方案从而增加了系统复杂度和故障风险传统消息队列通常需要手动配置或复杂脚本进行 Topic 管理会带来运维成本攀升与资源泄漏隐患等。破局之道Apache RocketMQ 进化为 AI 消息引擎系化重构采用存算分离架构实现资源弹性、通过存储层多副本机制保障高可用性、引入轻量级 SDK 提升客户端灵活性等等最终达成了高弹性、高可用、低成本的核心目标也为解决 AI 时代的工程难题打下了坚实的基础。面对 AI 时代带来的全新挑战Apache RocketMQ 进行了前瞻性战略升级从传统消息中间件进化为专为 AI 时代打造的消息引擎成为构建下一代 AI 应用不可或缺的关键基础设施。这一演进的核心在于两大“颠覆性创新”轻量化通信模型支持动态创建百万级 Lite-Topic特别适用于长时会话、AI 工作流和 Agent-to-Agent 交互等场景。显著提升系统的扩展性与灵活性满足 AI 应用复杂的通信需求。智能化资源调度通过削峰填谷、定速消费、自适应负载均衡和优先级队列等功能实现对稀缺算力资源的精细化管理和平稳高效调度确保在高并发和多租户环境下高效利用资源。这些创新使 Apache RocketMQ 成功突破了传统消息队列的局限精准匹配 AI 应用的独特需求为现代 AI 系统提供稳定且高效的消息中枢服务。场景实践RocketMQ for AI 如何破解 AI 工程挑战“会话即主题”用 Lite-Topic 终结长会话状态管理难题AI 应用的交互模式具有特殊性即长耗时、多轮次且高度依赖高成本计算的会话。当应用依赖 SSE 或WebSocket 等长连接时一旦连接中断如网关重启、链接超时、网络不稳定触发不仅会导致当前会话上下文的丢失更会直接造成已投入的 AI 任务作废从而浪费宝贵的算力资源。因此构建一个健壮的会话管理机制实现在长耗时的对话过程中保障会话上下文的连续性和完整性减少重试带来的算力资源浪费同时降低应用程序代码的复杂度是该场景的核心技术攻坚点。为解决长会话状态管理难题RocketMQ for AI 提出了一种革命性的轻量化解决方案——“会话即主题”系统可为每个独立会话Session或问题Question动态创建一个专属的轻量级主题Lite-Topic。当客户端与 AI 服务建立会话时系统将动态创建一个以 SessionID 命名的专属队列例如 chatbot/{sessionID}或 chatbot/{questionID}。该会话的所有交互历史和中间结果均以消息形式在该主题中有序传递 。即使客户端断连重连后只需继续订阅原主题 Lite-Topic chatbot/{sessionID}即可无缝恢复上下文实现断点续传继续推送响应结果。该模型有效解决了“无状态后端”与“有状态体验”之间的矛盾将开发者从繁琐的会话状态保持、重连处理与数据一致性校验中彻底解放出来。不仅大幅简化了工程实现也从根本上避免了因任务中断重试造成的算力资源浪费为用户带来流畅、连续、稳定的 AI 交互体验。图 1这一创新模式的实现得益于 RocketMQ 专为 AI 场景设计的强大特性百万级队列支持RocketMQ 支持在单个集群中高效管理百万级 Lite-Topic能够为海量并发会话或任务提供独立 Topic并且保障性能无损。轻量化资源管理RocketMQ 队列的创建和销毁极其轻量和自动化系统可按需自动创建与回收 Lite-Topic如客户端连接断开或 TTL 到期时避免资源泄漏和手动干预显著降低运维复杂度和成本。大消息体传输RocketMQ 可处理数十 MB 甚至更大的消息体充分满足 AIGC 场景中常见的庞大数据负载的传输需求如大量上下文的 Prompt、高清图像或长篇文档等。顺序消息保障在单个会话队列中通常采用 LLM 的流式输出模式以降低问答延迟RocketMQ 原生支持顺序消息确保推理结果流式输出到客户端的顺序性保障会话体验连贯流畅。全面可观测性RocketMQ 全面支持 OpenTelemetry 标准的 Metrics 和 Tracing可实时监控消息收发量、消息堆积等关键指标查询消息收发轨迹详情为多 Agent 系统的调试与优化提供有力支撑。应用案例阿里巴巴安全团队“安全小蜜”智能助手阿里巴巴安全团队推出的“安全小蜜”智能助手在应对大规模并发会话时曾面临会话上下文丢失、任务中断导致资源浪费等挑战。通过引入 RocketMQ 的 Lite-Topic 能力重构会话保持机制“安全小蜜”成功实现了会话状态的自动持久化与快速恢复。这不仅能够在多轮对话中对用户的安全问题进行快速、精准的理解和响应还大幅简化了工程实现复杂度有效降低了因任务中断引发的资源浪费整体提升了用户体验与业务处理效率。目前阿里云多个产品线的 AI 答疑机器人也已采用该方案完成升级进一步验证了该架构在多样化 AI 场景下的通用性与有效性。智能算力编排不止于负载均衡构建可控算力调度中枢大模型服务在资源调度上普遍面临两大核心挑战负载不匹配前端请求突发性强而后端算力资源有限且相对稳定直接对接易导致服务过载崩溃或算力资源浪费。无差别分配在实现流量平稳后如何确保高优先级任务优先获得宝贵的计算资源成为提升整体服务价值的关键。在此背景下Apache RocketMQ 发挥了关键作用不仅作为前端请求与后端算力服务之间的缓冲调度层将不规则的流量“整形”为平稳、可控的请求流还通过定速消费、优先级队列等能力提供“可控的算力调度中枢” 实现对请求流量的细粒度控制大幅提升资源利用效率与服务质量。图 2RocketMQ 所具备的一系列核心特性为实现智能算力调度提供了坚实的基础天然削峰填谷保护核心 AI 算力RocketMQ 天然具备“流量水库”的作用能缓存突发请求使后端 AI 模型服务根据自身处理能力基于类似滑动窗口模式自适应消费负载均衡避免系统过载或资源浪费。定速消费最大化 AI 算力利用率RocketMQ 支持定速消费能力可为消费者组 ConsumerGroup 设置消费 quota。开发者可灵活定义 AI 算力的每秒调用量在保障核心 AI 算力不过载的前提下最大限度提升吞吐量。优先级队列智能调度与分配算力资源再进一步RocketMQ 的消息优先级机制还为复杂的业务场景提供了灵活优雅的资源调度方案抢占式分配当高价值任务如 VIP 用户请求、关键系统分析进入系统时可将其标记为高优先级消息。RocketMQ 确保这些消息被优先消费让宝贵的算力资源优先服务于最关键的任务。按权重分配在共享算力池场景下可依据各业务请求的实时执行状态设置请求消息优先级调整请求执行的先后顺序既保障整体吞吐效率又防止个别租户因资源饥饿而无法获得算力。应用案例阿里云大模型服务平台百炼、通义灵码阿里云大模型服务平台百炼的网关系统通过引入 RocketMQ 实现了对请求流量的削峰填谷有效将前端不规则的访问压力转化为平稳、可控的后端算力调度。同时借助 RocketMQ 的消息优先级功能根据用户的请求流量设置合理的优先级避免了大流量用户请求导致小流量用户分配不到算力资源显著提升了资源利用率和服务公平性。通义灵码通过 RocketMQ 将其 codebase RAG 架构从原有的同步流程升级为异步流程实现代码向量化与流量削峰填谷保障了系统全链路的稳定性。异步通信枢纽Lite-Topic 让 A2A 与 AI 工作流彻底告别同步阻塞Google 提出的 A2A 协议推荐采用异步通信机制来解决 AI 任务长耗时带来的同步阻塞问题。其核心机制是将一次请求 - 响应Request-Reply调用解耦为一个初始请求和一个异步通知pushNotificationConfig。在各类 Agentic AI 平台的工作流中每个节点执行完任务后都需要向下游节点通知执行结果而异步通信正是支撑这种复杂协作的关键。由于 AI 任务普遍运行时间长工作流场景同样需要解决“同步调用导致级联阻塞”的问题。无论是 Agent 之间的外部通信还是工作流内部的任务流转都面临一个共同挑战如何优雅地处理长耗时任务避免系统阻塞核心解决方案是采用统一的架构模式——将长耗时、有状态的交互转化为由无状态、事件驱动的可靠异步通知机制来连接。前文提到Apache RocketMQ 全新推出的Lite-Topic 机制凭借其轻量化、自动化的动态管理能力可高效实现 Request-Reply 模式的异步通信。核心流程如下动态创建回复通道当 Agent A 向 Agent B 发起请求时如 message/send无需同步等待响应。而是在请求中嵌入唯一的动态回复地址例如 a2a-topic/{taskID}。同时Agent A 订阅该地址RocketMQ 会在首次连接时自动创建这个轻量化的 Sub-Topic相当于为本次任务开辟了一个专属的异步通信通道。异步投递执行结果Agent B 按照自己的节奏处理任务。在任务完成后它将结果封装为消息直接发布到请求中指定的回复地址 a2a-topic/{taskID}。自动回收通信资源当 Agent A 成功接收并处理完结果后会断开与该 Lite-Topic 的连接。RocketMQ 的智能资源管理机制会检测到该 Topic 已无消费者并在设定的 TTLTime-To-Live后自动清理该 Topic 资源。整个过程完全自动化无需人工干预杜绝了资源泄露的风险。RocketMQ 的 Lite-Topic 方案优势在于其系统性的设计百万级 Lite-Topic 的海量并发能力结合按需创建、用后即焚的零开销资源管理从根本上解决了大规模 Agent 协作场景下的扩展性与易用性问题。同时顺序消息保障机制确保了流式或多步任务的逻辑正确而内置的持久化与高可用机制则保障了异步通信的最终一致性与可靠性。这些能力共同为 A2A 场景构建了一个真正健壮、高效且可扩展的异步通信基础设施。应用案例阿里 AI 实验室阿里 AI 实验室在其多 AI Agent 工作流中基于 RocketMQ 构建了一套高效、可靠的 Agent 编排体系。工作流中的每个节点均采用事件驱动架构实现可靠、持久化的通信。借助 Lite-Topic 机制还能实现 Agent 之间的节点级通信从而实现任务流程的精细化编排。在多 Agent 协同执行 AI 任务的过程中即使遇到 Agent 发布重启、调用超时等情况导致完整任务链中断也能通过持久化事件流的可靠重试继续推进中断的 AI 任务既有效避免了资源浪费又显著提升了用户体验。架构解析RocketMQ for AI 的关键技术升级为实现前文所述的创新模型Apache RocketMQ 需具备在单个集群中高效管理百万级 Lite-Topic 的能力但原有架构在支持该能力时面临两大核心挑战在存储层面原先基于文件的索引和元数据管理机制已难以支撑如此量级的 Topic在消息分发投递过程中当单个消费者订阅大量的 Lite-Topic 时旧有的长轮询通知机制在延迟和并发性能上也显得捉襟见肘。因此要实现海量 Lite-Topic 的高效管理必须攻克以下两个关键技术难题百万级 Lite-Topic 的元数据存储与索引结构的技术方案面向海量 Lite-Topic 订阅场景的高效消息分发与投递机制。图 3百万级 Lite-Topic 的数量级跃升意味着索引和元数据无法沿用之前的模型。若为每个主题维护一个或者多个基于物理文件的索引结构将带来巨大的系统开销和运维负担。为此Apache RocketMQ 基于其 LMQ 存储引擎 和 KV Store 能力重新设计了元数据管理和索引存储统一存储、多路分发所有消息在底层的 CommitLog 文件中仅存储一份但通过多路分发机制可以为不同的 Lite-Topic 生成各自的消费索引ConsumerQueue简称 CQ。索引存储引擎升级摒弃了传统的文件型 CQ 结构替换为高性能的 KV 存储引擎 RocksDB。通过将队列索引信息和消息物理偏移量Physical Offset作为键值对存储充分发挥 RocksDB 在顺序写入方面的高性能优势从而实现对百万级队列的高效管理。在 Lite-Topic 存储模型的基础上RocketMQ 进一步对消息分发与投递机制进行优化针对单个消费者订阅上万个 Lite-Topic 的场景重新设计了一套创新的事件驱动拉取Event-Driven Pull机制如图 3 所示订阅关系Subscription Set管理Broker 负责管理消费者订阅关系 Subscription 的 Lite-Topic Set并支持增量更新从而能够实时、主动地感知消息与订阅的匹配状态。事件驱动与就绪集Ready Set维护每当有新消息写入Broker 会立即根据其维护的 Subscription Set 进行匹配并将符合条件的消息或其索引添加到为消费者维护的 Ready Set 中。高效 Poll Ready Set消费者只需对 Ready Set 发起 poll 请求即可从 Ready Set 中获取所有匹配的消息。这种方式允许 Broker 将来自不同主题、不同流量的消息进行合并与攒批在一次响应中高效地返回给消费者显著降低了网络交互频率从而提升整体性能。通过在存储层与分发机制的创新升级Apache RocketMQ 有效解决了 Lite-Topic 模型的关键挑战在存储层面采用高性能的 RocksDB 替代传统文件索引实现了对百万级元数据的高效管理在消息分发层面通过创新的“事件驱动拉取”模型由 Broker 主动维护订阅集与就绪集将消费者的海量轮询转变为对聚合消息的单次高效拉取确保了在海量订阅场景下的低延迟与高吞吐。展望未来开启 AI MQ 新时代RocketMQ for AI 持续演进Apache RocketMQ for AI 的演进标志着其已从传统消息中间件全面升级为专为 AI 时代打造的消息引擎。通过在轻量化通信模型与智能化资源调度方面的“颠覆性创新”Apache RocketMQ 突破了传统消息中间件的能力边界成为构建高可用、可扩展 AI 应用的关键基础设施展现出其在 AI 工程化体系中的核心价值。Apache RocketMQ for AI 的增强能力已在阿里巴巴集团内部以及阿里云大模型服务平台百炼、通义灵码等产品中经过大规模生产环境的验证充分证明了其在高并发、复杂的 AI 场景下的成熟度与可靠性。当然这只是一个开始。AI 工程化仍处于快速发展阶段Apache RocketMQ 作为核心基础设施仍有广阔的优化与创新空间。未来阿里云消息团队将持续围绕用户 AI 场景迭代升级协同 Apache RocketMQ 开源社区的贡献者们打磨核心 AI 能力并逐步将经过阿里集团 AI 业务验证过的方案与特性持续反馈到开源社区。我们坚信通过持续的技术探索与开放共建Apache RocketMQ for AI 将推动“AI 原生消息队列”AI MQ成为行业标准助力全球开发者更轻松、更高效地构建下一代智能应用共同推动 AI 工程实践的标准化、普及化与生态繁荣。本文已同步收录至「RocketMQ 中文社区」 面向 RocketMQ 中文开发者的一站式学习社区