Apache RocketMQ x AlAl Agent的事件驱动架构实践邹星宇阿里云智能云原生解决方案架构师C)阿里云01传统消息队列在A1场景中的局限性O2Lite-Topic:面向Al领域的新范式Contents目录O3优先级Topic:面向A1场景的分级消费策略04面向A1场景的应用案例1、传统消息队列在A1场景中的局限性AI业务场景的新特性更长的响应时间更复杂的交互和传统互联网应用毫秒级的延时比，A!·A1应用的的多轮对话持续时间长，历史可应用的响应时常基本是分钟级以上A1应用一次业务的运行时间不可预测性·上下文的传输可达到几十至上百M,多Agen之间的协同更加复杂更昂贵的计算资源更精细化的事件驱动·A推理依赖昂贵的GPU资源，瞬时高·因为计算能力有限，异步事件驱动需要并发流量可能冲击推理服务稳定性，导更精准的消费速度控制，致算力资源浪费。分级的事件驱动策路，确保高优先级任·任务失败重试的成本更高。务优洗获得宝贵的计算资源，AI时代RocketMQ新特性轻量化通信模型智能化资源调度大上下文管理轻量级的资源管理，低定速消费以连续的消息保开销的资源创建支持海优先级消费存上下文量会话场景优先级修改支持超大消息体更细粒度的的订阅管理确保在高并发和多租顺序、互斥消费适用于长时会话、A!工户环境下高效利用资保证上下文的完作流和Agent-to-源整Agent交互等场景2、Lite-Topic:面向AI领域的新范式轻量化通信模型：·百万级队列支持：RocketMQ支持在单个集群中高效管理百万级Lite-Topic,,能够为海量并发会话或任务提供独立Topic,并且保障性能无损。轻量化资源管理：RocketMQ队列的创建和销毁极其轻量和自动化，系统可按需自动创建与回收Lit-Topic(如客户端连接断开或TTL到期时)，避免资源泄漏和手动干预，显著降低使用复杂度和成本。大消息体传输：RocketMQ可处理数十MB甚至更大的消息体，充分满足AIGC场景中常见的庞大数据负载的传输需求，如大量上下文的Prompt、高清图像或长篇文档等。顺序消息保障：通过顺序消费，确保推理结果流式输出到客户端的顺序性，保障会话体验连贯流畅。QueueGroup-1轻量级Topic:MessageQueu-1QuqueGmoup-2·基于百万队列的方案，本质上是一个个queueMossagoOucue-2QuquoGmup-3·从全局上来看，一个轻量级Topic不会存在于QueueGroup-4每一个broker.上，在分配和发送时像顺序Topic的发送一样要做queue的hashLiteTopie-1Queue的消息是某个broker专属的，一个轻LieTopic-2量级topic的发送在只会到一台broker,而不轻量是轮询发送LneTopic-3I brokar-2LieTopie-4轻量级订阅：·消费组group的概念被弱化订阅关系粒度更细，以client ID维度维护Broker Registry Table新增互斥(Exclusive)消费模式group,topic,version,[ite topics]没有订阅关系不一致导致的各种消费问题Broker Version Tablec3group,topic,version,[ite topics]B1·TTL到期后自动删除订阅关系c2 versionProxyBroker消费分发策略：·客户端发起读请求不再指定topic,而是broker根据client_ID识别订阅关系，并返回多个topic的多条消息commt·引入类似epoll机制的topic ready set,在pop请求处理时直接访问就绪的topic·当订阅上线、新消息发送、消息ACK后仍有消息、order lock释放时往topic readyset进行add操作传统订阅模式新的模式消息客户端数量随应用负载和规模而定无差别订阅关系所有机器订阅关系一致不同机器订阅关系不同订阅Topic数量很少，几个到十几个较多，几千到1w级别console预创建动态生成，不适合预创建Topic:生命周期很长。如用于订单处理的Orderi消短。息，只要订单系统还在，主题就还在使用主题下消息数量多较少，几条到几万条消费过滤可过滤不过滤（因为粒度够细）Group用于共享消费的组织形式，因此需要订阅关系一致应用场景传统，共同完成某项任务，身份在特殊的场景下，各自接收属于无差别自己的消息3、优先级Topic:面向AI场景的分级消费策略大模型服务在资源调度上，普遍面临两大核心挑战负载不匹配：前端请求突发性强，而后端算力资源有限且相对稳定，直接对接易导致服务过载崩溃或算力资源浪费。无差别分配：在实现流量平稳后，如何确保高优先级任务优先获得宝贵的计算资源，成为提升整体服务价值的关键。C)阿里云优先级Topic为实现智能算力调度提供了坚实的基础削峰填谷保护A!算力：RocketMQ天然具备“流量水库"的作用，能缓存突发请求，使后端A!模型服务根据自身处理能力，基于类似滑动窗口模式自适应消费负载均衡，避免系统过载或资源浪费。·定速消费，最大化Al算力利用率：RocketMQ支持定速消费能力，可为消费者组group设置消费quota。开发者可灵活定义Al算力的每秒调用量，在保障核心A!算力不过载的前提下，最大限度提升吞吐量。·抢占式分配：当高价值任务（如VⅥP用户请求、关键系统分析）进入系统时，可将其标记为高优先级消息。RocketMQ确保这些消息被优先消费，让宝贵的算力资源优先服务于最关键的任务。按权重分配：在共享算力池场景下，可依据各业务请求的实时执行状态设置请求消息优先级，调整请求执行的先后顺序，既保障整体吞吐效率，又防止个别租户因资源饥饿而无法获得算力。优先级Topic:·多队列：对应优先级定义，不同优先级对应不同队列Head队列选择：对应调度（存取策略），优先从高P队列取数，依次进行ProducerConsumerget(max 3)队列选择策略：·永远从高到低依次选择◆高优先级消息·概率模式，当前普通消息的pop其实也在按概率选择队列，只不过每个队列概率相等，对于优先级队列来说，高优先级的被筛选到的概率更高