引言 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 数据处理的未来！

作者简介：艾阳坤，Apache RocketMQ PMC Member/Committer，CNCF OpenTelemetry Member，CNCF Envoy contributor。 在分布式系统中，多个服务之间的交互涉及到复杂的网络通信和数据传输，其中每个服务可能由不同的团队或组织负责维护和开发。因此，在这样的环境下，当一个请求被发出并经过多个服务的处理后，如果出现了问题或错误，很难快速定位到根因。分布式全链路追踪技术则可以帮助我们解决这个问题，它能够跟踪和记录请求在系统中的传输过程，并提供详细的性能和日志信息，使得开发人员能够快速诊断和定位问题。对于分布式系统的可靠性、性能和可维护性起到了非常重要的作用。 RocketMQ 5.0 与分布式全链路追踪 Apache RocketMQ 5.0 版本作为近几年来最大的一次迭代，在整个可观测性上也进行了诸多改进。其中，支持标准化的分布式全链路追踪就是一个重要的特性。 RocketMQ 5.0 可观测 而由 Google、Microsoft、Uber 和 LightStep 联合发起的 CNCF OpenTelemetry 作为 OpenTracing 和 OpenCensus 的官方继任者，已经成为可观测领域的事实标准，RocketMQ 的分布式全链路追踪也围绕 OpenTelemetry 进行展开。 分布式链路追踪系统的起源可以追溯到 2007 年 Google 发布的论文。这篇论文详细介绍了 Google 内部使用的链路追踪系统 Dapper，其中使用的 span 概念被广泛采用，并成为后来开源链路追踪系统中的基础概念之一。 Dapper Trace Tree 在 Dapper 中，每个请求或事务被追踪时都会创建一个 span，记录整个请求或事务处理过程中的各个组件和操作的时间和状态信息。这些 span 可以嵌套，形成一个树形结构，用于表示整个请求或事务处理过程中各个组件之间的依赖关系和调用关系。后来，很多开源链路追踪系统，如 Zipkin 和 OpenTracing，也采用了类似的 span 概念来描述分布式系统中的链路追踪信息。现在，合并了 OpenTracing 和 OpenCensus 的 CNCF OpenTelemetry 自然也一样采用了 span 概念，并在此基础上进行了进一步发展。 OpenTelemetry 为 messaging 相关的 span 定义了，旨在制定一套与特定消息系统无关的 specification，而 OpenTelmetry 自身的开发其实也都是由 specification 驱动进行展开。 Specification Driven Development Messaging Span 定义 Specifaition 中描述了 messaging span 的拓扑关系，包括代表消息发送、接收和处理的不同 span 之间的父子和链接关系。关于具体的定义可以参考：。对应到 RocketMQ 中，有三种不同的 span： | Span | Description | | | | | send | 消息的发送过程。span 以一次发送行为开始，成功或者失败/抛异常结束。消息发送的内部重试会被记录成多条 span。 | | receive | 消费者中接收消息的长轮询过程，与长轮询的生命周期保持一致。 | | process | 对应 PushConsumer 里 MessageListener 中对消息的处理过程，span 以进入 MessageListener 为开始，离开 MessageListener 为结束。 | 特别地，默认情况下，receive span 是不启用的。在 receive span 启用和不启用的两种情况下，span 之间的组织关系是不同的： 启用 receive span 前后的 span 关系 在没有启用 receive span 的情况下，process span 会作为 send span 的 child；而当 receive span 启用的情况下，process span 会作为 receive span 的 child，同时 link 到 send span。 Messaging Attributes 定义 语义约定中规定了随 span 携带的通用属性的统一名称，这包括但不限于： messaging.message.id: 消息的唯一标识符。 messaging.destination：消息发送的目的地，通常是一个队列或主题名称。 messaging.operation：对消息的操作类型，例如发送、接收、确认等。 具体可以查看 。 特别地，不同的消息系统可能会有自己特定的行为和属性，，这包括： | Attribute | Type | Description | | | | | | messaging.rocketmq.namespace | string | RocketMQ 资源命名空间，暂未启用 | | messaging.rocketmq.client_group | string | RocketMQ producer/consumer 负载均衡组，5.0 只对 consumer 生效 | | messaging.rocketmq.client_id | string | 客户端唯一标识符 | | messaging.rocketmq.message.delivery_timestamp | int | 定时消息定时时间，只对 5.0 生效 | | messaging.rocketmq.message.delay_time_level | int | 定时消息定时级别，只对 4.0 生效 | | messaging.rocketmq.message.group | string | 顺序消息分组，只对 5.0 生效 | | messaging.rocketmq.message.type | string | 消息类型，可能为 normal/fifo/delay/transaction，只对 5.0 生效 | | messaging.rocketmq.message.tag | string | 消息 tag | | messaging.rocketmq.message.keys | string[] | 消息 keys，可以有多个 | | messaging.rocketmq.consumption_model | string | 消息消费模型，可能为 clustering/broadcasting，5.0 broadcasting 被废弃 | 快速开始 在 OpenTelemetry 中有两种不同的方式可以为应用程序添加可观测信息： Automatic Instrumentation：无需编写任何代码，只需进行简单的配置即可自动生成可观测信息，包括应用程序中使用的类库和框架，这样可以更方便地获取基本的性能和行为数据。 Manual Instrumentation：需要编写代码来创建和管理可观测数据，并通过 exporter 导出到指定的目标。这样可以更灵活自由地控制用户想要观测的逻辑和功能。 在 Java 类库中，前者是一种更为常见的使用形式。RocketMQ 5.0 客户端的 trace 也依托于 automatic instrumentation 进行实现。在 Java 程序中，automatic instrumentation 的表现形式为挂载 Java agent。在过去的一年里，我们将 推入了 OpenTelemetry 官方社区。现在，只需要在 Java 程序运行时挂载上 OpenTelemetry agent，即可实现对应用程序透明的分布式全链路追踪。 除此之外，Automatic Instrumentation 和 Manual Instrumentation 并不冲突，Automatic Instrumentation 中所使用的关键对象会被注册成全局对象，在 Manual Instrumentation 的使用方式中也可以非常方便的获取。实现两个 Instrumentation 共用一套配置，非常灵活和方便。 首先准备好 RocketMQ 5.0 Java 客户端，可以参考 进行消息的收发。关于 RocketMQ 5.0 的更多细节，欢迎大家参考和关注 和 。 然后准备好 OpenTelemetry agent jar，可以从 OpenTelemetry 官方，在应用程序启动时增加 javaagent:yourpath/opentelemetryjavaagent.jar 即可。可以通过设置 OTEL_EXPORTER_OTLP_ENDPOINT 环境变量来设置 OpenTelemetry collector 的接入点。 默认情况下，按照 OpenTelemetry 中关于 messaging 的规范，只有 send 和 process 的 span 会被启用，receive 的 span 是默认不启用的，如果想要启用 receive span，需要手动设置 Dotel.instrumentation.messaging.experimental.receivetelemetry.enabled=true。 场景最佳实践 目前，主流的云服务供应商都为 OpenTelemetry 提供了良好的支持，阿里云上的 SLS 和 ARMS 两款可观测产品都提供了基于 OpenTelemetry 的分布式全链路追踪服务。 为了更好地展示分布式全链路追踪的过程，这里提供了一个代码示例： 。在这个代码示例中，会启动三个不同的进程，涉及三种不同类库和业务逻辑之间的相互调用，展示了一个在分布式环境较复杂中间件之间进行交互的典型案例。 请求首先会从 gRPC 客户端发往 gRPC 服务端，在 gRPC 服务端收到请求之后，会向 RocketMQ 5.0 的 Producer 往服务端发送一条消息，然后再回复对应的 response 给客户端。在 RocketMQ 5.0 的 PushConsumer 接受到消息之后，会在 MessageListener 中使用 Apache HttpClient 往淘宝网发送一条 GET 请求。 示例代码调用链路 特别地，gRPC 客户端在发起具体的调用是在一个上游业务 span 的生命周期之内进行的，这个 span 我们称之为 ExampleUpstreamSpan，RocketMQ 5.0 PushConsumer 在收到消息之后，也会在 MessageListener 里执行其他的业务操作，也会有对应的 span，我们称之为 ExampleDownstreamSpan。那么默认在 receive span 没有启用的情况下，按照开始时间的顺序，会先后存在 7 个 span。分别是： ExampleUpstreamSpan。 gRPC 客户端请求 span。 gRPC 服务端响应 span。 RocketMQ 5.0 Producer 的 send span。 RocketMQ 5.0 Producer 的 process span。 HTTP 请求 span。 ExampleDownstreamSpan。 RocketMQ 5.0 对接 SLS Trace 服务 首先在阿里云日志服务中创建 Trace 服务。然后获取接入点，项目和实例名称等信息，具体可以参考。 在补充好信息之后完成接入之后，稍等一会就可以看到对应的 Trace 信息已经被上传到 SLS trace 服务中： SLS Trace 服务分布式全链路展示 Trace 服务其实是将相关数据存储到日志中，因此这些数据也可以通过 SLS 的 SQL 语法查询得到。 通过 Trace 数据，我们可以很方便知道用户的操作系统环境，Java 版本等一系列基础信息。消息的发送延时，失败与否，消息是否准时投递到了客户端，以及客户端本地消费耗时，消费失败与否等一系列有效信息，可以帮助我们十分有效地进行问题排查。 除此之外，SLS Trace 服务的 demo 页也提供了基于 RocketMQ 5.0 定制的消息中间件大盘，生动展示了利用 Trace 数据得到的发送成功率，端到端延时等一系列指标。 ：展示利用 Trace 数据得到的包括发送延时、发送成功率、消费成功率、端到端延时在内的一系列指标。 ：可以根据上一步得到的差错长 message id 进行进一步的细粒度查询。 消息中间件分析 RocketMQ 5.0 对接应用实时监控服务（ARMS） 首先进入应用实时监控服务 ARMS 控制台，点击接入中心中的 OpenTelemetry，选择 java 应用程序下的自动探测，获取启动参数并修改至自己的 java 应用程序，具体可以参考。 配置好参数之后，启动自己的相关应用程序，稍等一会儿，就可以在 ARMS Trace Explorer 里看到对应的数据了。 Trace Explorer 还可以查看 span 之间的时序关系。 ARMS Trace Explorer 分布式全链路追踪展示 具体地，可以点进每个 span 查看详细的 attributes/resources/events 等信息。除此之外，ARMS 还支持通过使用 OpenTelemetry Collector 转发的形式来收集应用程序的 Trace 数据。 趋势与思考 随着现代应用程序架构的不断演进，可观测性的重要性日益凸显。它不仅可以帮助我们快速发现和解决系统中的问题，还提高应用程序的可靠性和性能，同时也是实现 DevOps 的关键部分。在相关领域，也陆续诞生了像 DataDog 和 Dynatrace 这样的明星公司。 近年来涌现了一些新兴技术，如 eBPF（Extended Berkeley Packet Filter）和 Service Mesh 也为可观测领域提供了一些新的思路： eBPF 可以在内核层面运行，通过动态注入代码来监控系统的行为。它被广泛应用于实时网络和系统性能监控、安全审计和调试等任务，并且性能影响很小，未来也可以作为 continuous profiling 的一种选择。Service Mesh 则通过在应用程序之间注入代理层实现流量管理、安全和可观测性等功能。代理层可以收集和报告有关流量的各种指标和元数据，从而帮助我们了解系统中各个组件的行为和性能。 Service Mesh 中反映出的技术趋势很大一部分已经在 RocketMQ 5.0 proxy 中得到了应用，我们也在更多地将可观测指标往 proxy 进行收敛。当前的 Trace 链路未来也在考虑和服务端一起进行关联，并打造用户侧，运维侧，跨多应用的全方位链路追踪体系。除此之外还可以将 Trace 数据与 Metrics 数据通过 Exemplars 等技术进行联动。实现面到线，线到点的终极排查效果。 在可观测领域，RocketMQ 也在不断探索和摸索更加领先的可观测手段，以帮助开发者和客户更快更省心地发现系统中的隐患。 特别感谢阿里云 SLS 团队的千乘同学和 ARMS 团队的垆皓同学在接入过程提供的帮助和支持！ 相关链接 RocketMQ 5.0 客户端： OpenTelemetry Instrumentation for RocketMQ 5.0： RocketMQ OpenTelemetry 示例：

消息典型应用场景 阿里云拥有丰富的消息产品家族，除了 RocketMQ 以外，还有大家熟知的对标开源的云 Kafka、支持 AMQP 协议的开源消息队列 RabbitMQ、物联网通信网关 MQTT、 对标 AWS SQS/SNS 的 Serverless 版消息 MNS（现在也是轻量版 RocketMQ）以及云上事件总线、事件中心 EeventBridge 。 阿里云所有消息产品均采用 RocketMQ 作为底层存储引擎，为用户提供稳定、可靠、高性能的消息通信能力，比如百万 TPS、百万队列、毫秒级通信延迟、分级存储、 Serverless 弹性等众多的消息产品。也带来了丰富的应用场景，分为应用集成和数据集成两大类。 应用集成以 RocketMQ 为主，应用最为广泛，本文也将分享 RocketMQ 在微服务解耦、电商交易、金融支付等场景下的最佳实践，比如银行的交易流水、保单的支付流转等。RabbitMQ 、MQTT 也主要用于应用集成场景，比如物联网、 IoT 双向通信、云产品的事件通知以及后处理等。对于新建的业务场景，一般首推 RocketMQ 作为消息选型，因为 RocketMQ 拥有最丰富的功能特性；而对于存量的业务迁移，则可以根据具体使用的消息产品来进行选择，以降低迁移成本。 数据集成以云 Kafka 为主，在大数据分析、日志采集分析等场景下应用最为广泛，比如游戏的玩家操作、广告埋点、数据分析、应用数据监控等。各种 SaaS 类的集成、聚石塔、电商数据打通等场景，则主要使用 EventBridge。更多的产品选型对比，可以参考专题页中的消息队列产品选型。 业务消息使用场景 RocketMQ 经过阿里集团内部十年锤炼，经过双 11 大促等万亿级规模的实时场景验证，SLA 最高可支持 4 个9，数据可靠性支持 9 个9。 以微服务解耦、订单交易、金融支付等场景为例，在互联网、电商、零售、金融等行业，消息的使用量非常大。尤其是在秒杀大促时，为了保障系统的稳定运行，需要 RocketMQ 进行削峰填谷。另外金融客户对每笔交易、每个订单也都要求数据不能丢失。因此在此类场景普遍对消息的可靠传输、海量并发、低延迟、稳定性与容灾等有着非常高的要求。RocketMQ 提供了丰富的消息类型，比如事务消息、定时消息、顺序消息等。 在交易系统里，为了简化交易流程，一般使用事务消息和定时消息。同时 RocketMQ 也提供了消息轨迹查询、消息 dashboard ，可以非常方便地对每个消息进行回溯，对每个 topic 或者 group 进行监测。RocketMQ 5.0 也提供了丰富的实例规格，从百级别 QPS 到百万级 QPS ，可以覆盖大部分应用场景。RocketMQ 默认提供多副本、多可用区部署，也提供了跨地域消息路由能力，支持客户构建高可用容灾或多活，且 RocketMQ 能够支持 99.9%的消息 RT 在 10ms 传输。 RocketMQ 事务消息举例 实现订单状态机异步流转 以订单状态机异步流转为例。此前，如果收到一笔订单交易，需要逐个通知下游的服务模块，比如需要更新购物车、更新积分等。每个业务模块耦合在一起会导致大促时的流量峰值非常大，需要每个服务模块保障其处理性能。而基于 RocketMQ 的事务消息能力，即可轻松实现订单子流程系统的异步解耦和流量的削峰填谷，事务消息可以确保数据库订单状态持久化和下游通知的事务性。 收到 LBL 订单交易时，可以先向 RocketMQ 发送一条半事务消息，此时 RocketMQ 会 hold 住消息，等核心交易事务完成后再向 MQ 提交确认半事务消息的状态，并执行下游服务模块的通知。假设核心交易模块失败，则会废弃之前提交的半事务消息，不通知下游。 对比此前的传统事务模块，使用 RocketMQ 可以大幅简化交易链路，缩短下单耗时。尤其是在大促场景下，可以解耦下游的服务模块，提供削峰填谷的能力。 超时中心 RocketMQ 的定时消息场景也是常见的使用方式。 比如双 11 等大促场景存在大量预售订单、定点尾款等，会带来大量定时任务。在电商交易过程中，订单流转也存在多个超时状态的任务，处理超时状态的任务需要确保可靠及时。以传统的方案进行构建分布式调度机制实现的时候，比如基于定时器调度延迟大,可能会存在性能瓶颈。 而采用 RocketMQ 的定时消息，实现将变得非常简单。定时任务只需提交一条延迟消息到 RocketMQ ，由 RocketMQ 保障定时消息达到秒级的精度，最高可支持百万级别的 TPS 能力，同时也能支持消息的消费重试，保障任务可靠触发，相比传统的使用方式大大简化了定时的复杂度。 RocketMQ 灰度策略举例 微服务全链路灰度 微服务场景下，精准地控制灰度流量并进行灰度版本验证，是保障线上业务稳定运行的关键。大部分情况下，用户通过划分不同的环境来进行灰度发布，对应 RocketMQ 的不同实例。但是很多用户希望能够简化环境管理，尽可能复用线上资源，结合消息来提供微服务全链路灰度能力。 如上图所示，线上已经在运行的微服务模块游 A、B、C，C 模块会产生消息，并由 A 模块进行消费。此时对服务模块 A 和 C 做灰度发布，则线上会存在两条泳道，一条是正常的业务流量，一条是灰度链路。我们希望线上版本 C 模块生产的消息能够被线上版本 A 模块进行消费，灰度版本 C 模块生产的消息能够被灰度版本 A 模块进行消费。 RocketMQ 支持透传环境标签，可在生产端给消息属性添加标签，然后开启 RocketMQ 的 SQL 92 语法过滤，服务端即可完成消息的过滤和路由，从而降低客户端的压力。 本文讲述了 RocketMQ 的业务消息场景、一些功能特性的使用方法，包括事务消息、定时消息、消息全链路灰度等，欢迎大家尝试使用。

阿里云消息队列 RocketMQ 商业化历程 RocketMQ 诞生于 2012 年，诞生即开源。2012～2015 年，RocketMQ 一直在通过内部电商业务打磨自身服务能力,并在 2015 年于阿里云上线公测。2016 年，阿里云 RocketMQ 完成商业化，同时被捐赠给 Apache 基金会，同年获得了年度受欢迎中国开源软件荣誉。 在 Apache 孵化期间，Apache RocketMQ 经历了快速发展，2017 年即毕业成为了 Apache 顶级项目。同年，Apache RocketMQ TLP RocketMQ 4.0 正式发布。此后，RocketMQ 4.0 经历了长足发展，期间阿里云商业和开源相辅相成、齐头并进，直到今天，共同迈入 RocketMQ 5.0 时代。 RocketMQ 5.0 发布后，阿里云商业会持续采取 OpenCore 的发展模式，秉承上游优先的社区发展原则，与社区一起将 RocketMQ 打造为一个超融合的数据处理平台。 阿里云消息队列产品矩阵 阿里云基于 RocketMQ 消息底座，构建了多元化的消息产品系列。 RocketMQ 是阿里云主打的消息品牌，互联网新兴业务领域首选的数据通道。消息队列 Kafka 是大数据的首选数据通道，微消息队列 MQTT 是移动互联网和物联网的数据通道，消息队列 RocketMQ 是传统业务领域的数据通道。消息服务 MNS 是 RocketMQ 轻量版，主要应用于应用集成领域，为平台型应用提供简单的队列服务。事件总线 Event Bridge 定位为云上事件枢纽，旨在阿里云上构建统一的事件中心。 阿里云消息队列产品矩阵完全构建在 RocketMQ 之上，基本实现了应用场景全覆盖，包括微服务解耦、SaaS 集成、物联网、大数据或日志收集生态，同时也在内部覆盖了阿里巴巴所有业务，在云上为数万阿里云企业提供了优质的消息服务。阿里云的消息产品矩阵涵盖了互联网、大数据、移动互联网等领域业务场景，为云原生客户提供不可或缺的一站式解决方案。 RocketMQ 在阿里云商业化历程中，一直致力于探索业务消息实践，也孵化了大量业务消息特性，并持续反哺到开源社区。 RocketMQ 4.0 业务消息探索之路 RocketMQ 在商业化过程中，陆续推出了四种消息类型来满足丰富的业务场景。 普通消息：普通消息提供极致弹性、海量堆积能力，内置重试与死信队列来满足业务对失败重试的需求，同时具备高吞吐、高可用、低延迟等特性，广泛应用于应用集成、异步解耦、削峰填谷等场景。 定时消息：提供秒级定时精度， 40 天超长定时，主要面向分布式定时调度、任务超时处理等场景，目前正在开源中。  顺序消息：支持全局与局部严格有序，从发送、存储到消费，保证端到端有序。面向有序事件处理、撮合交易、数据实时增量同步等场景。 事务消息：分布式、高性能、高可用的最终一致性事务解决方案，广泛应用于电商交易系统中服务的一致性协调场景并且已经开源。  RocketMQ 4.0 期间，商业和开源都致力于全方位拓展消息接入能力，使 RocketMQ 能够非常轻松地连接应用开源和云产品生态。比如商业上提供了多语言 SDK ，开源也有相应的 SDK 能够覆盖 Java、Go、Python 、C++使用 RocketMQ。同时支持 Spring 生态，能够通过 Spring Cloud 的方式使用 RocketMQ。商业上提供了一组非常简单易用的 HTTP API，提供了 67 种语言的实现。 除了 SDK 接入，RocketMQ 也在积极拥抱社区标准，在云产品侧提供了 AMQP 和 MQTT 的接入能力，其中 MQTT 已开源。 RocketMQ 也大力在发展 connector 生态，能够通过 RocketMQ connector 接入很多数据源，包括 Redis、MongoDB、Hudi 等大数据系统。 另外，阿里云构建的事件总线 EventBridge 也已开源，通过该产品能够将阿里云的云产品、SaaS 应用、自建数据平台的数据引入 RocketMQ。 RocketMQ 4.0 版本做了大量尝试，提供了全方位的消息接入能力。 RocketMQ 在服务阿里集团用户和商业化历程中，沉淀了大量领先的业务消息处理与服务能力。比如消息订阅方面，RocketMQ 支持集群分布式消费能力，也支持广播消费。在消息处理方面支持基于 Tag 和 SQL 做灵活过滤，其中基于 SQL 过滤是电商交易中非常重要的特性，能够支持在非常订阅比的情况下实现较低的投递比。 全球消息路由能力具备性能高、实时性强的特点。在云时代，数据中心天然分布在各个地域，各个地域之间还有 VPC 网络隔离。但是通过全球消息路由功能可以将地域与网络打通，能够满足更多业务场景。比如在阿里内部基于该能力实现了异地多活、异地容灾等企业级特性。 另外，全球消息路由具备非常高的易用性，提供了可视化任务管理界面，通过简单配置即可创建复制链路。 消息治理方面，RocketMQ 提供了访问控制、命名空间、实例限流、消息回放、重试消息、死信消息、堆积治理等能力。 服务能力方面，RocketMQ 经历了非常多沉淀，它在为交易链路服务了 12  年，参加了 10 年双 11，这也保证了 RocketMQ 能够在阿里云上提供非常高的可靠性。双 11 消息收发 TPS 峰值过亿，日消息收发总量超过 3 万亿。而即使在双十一万亿级数据洪峰下，消息也能做到 99.996% 毫秒级响应能力，消息发布平均响应时间不超过 3 毫秒，最大不超过 20 毫秒，真正实现了低延迟消息发布。 商业化初期，客户遇到最大难题是在分布式环境下如何完整地追踪异步消息链路。基于此背景，我们打造了可视化全生命周期消息轨迹追踪系统，能够提供丰富的消息查询、消息下载、定点重投、轨迹追踪能力，通过可观测系统帮助用户解决分布式环境中不可观测的问题。 如上图所示，一条消息从产生、发送至服务端存储到最终投递到消费者，整个发送和消费轨迹都有迹可循，包括投递给哪些消费者、哪些消费者在什么地方成功消费或者消费失败、何时进行重投，真正帮助客户解决了分布式观测难题。 除了功能特性，RocketMQ 在稳定性方面也做了很多建设。我们始终坚持，SLA 是云原生的根本，因此整个研发运维链路都有严格的稳定性保障措施： 架构开发：每个方案设计都会面向失败设计，代码开发阶段会有严格 Code Review 阶段，也会完整经历单元测试、集成测试、性能测试和容灾测试流程。 变更管理：有着非常严格的变更制度，要做到每个变更可灰度、可监控、可回滚、可降级。 稳定性防护：提供了限流、降级、容量评估、应急方案、大促保障等能力，会定期进行故障和预案演练，定期进行风险梳理。 体系化巡检：在云上有全方位的生产环境黑盒巡检。基于用户视角，会对全地域所有功能做全功能扫描，包含高达 50 多项检测项，任意项功能出问题都能立刻被监测到。在白盒巡检方面，会对 JVM 运行时指标、内核系统、集群指标进行巡检。 故障应急：有完整地故障应急流程，包括监控报警、故障发生、快速止血、排查根因、故障复盘。 RocketMQ 5.0 云原生架构升级之路 云原生时代，云上用户对云产品服务化程度、弹性能力、可控制性能力以及韧性都有了更高的要求。在此背景之下，我们对 RocketMQ 进行了云原生架构升级，这也是 RocketMQ 5.0 的诞生背景。 轻量级 SDK：基于云原生通信标准 gRPC 开发了一组轻量级 SDK，能够与当前富客户端优势互补。   无状态消息网关：在核心数据链路推出了无状态消息网关。通过搭建无状态服务节点Proxy，再通过 LB 进行服务暴露，将存储节点数据分离来独立负责核心消息存储和高可用。Proxy 与 Store 节点分离部署，独立弹性。  Leaderless 高可用架构：Store 节点身份完全对等，完全 Leaderless 化，去 ZK 和 HA 管控节点，能够做到非常高的可用性。同时相比传统的 Raft 一致性协议，该 Leaderless 架构能够做到副本数灵活选择，同步异步自动升降级，实现秒级故障转移。高可用架构目前已经完成开源并与 Dledger 进行了融合。  云原生基础设施：可观测验能力云原生化，OpenTelemetry 标准化。整体架构走向 Kubernetes 化，能够充分利用售卖区的资源弹性能力。 RocketMQ 4.0 推荐的接入方式主要是富客户端。富客户端提供了诸如客户端侧负载均衡、消息缓存、故障转移等一系列企业级特性。但在云原生时代，轻量级、高性能的客户端更容易被云原生技术栈所集成。 因此，RocketMQ 5.0 重磅推出了全新多语言轻量级 SDK，具有以下优势： 全新极简 API 设计：不可变 API，有完善的错误处理。多语言 SDK 保障 API 在 Native 层面对齐。同时引入了全新的 Simple Consumer，能够支持按消息模型进行消费，用户不再需要关心消息队列，只需要关注消息。  通信层采用 gRPC 协议：拥抱云原生通信标准，gRPC 能够使服务更易被集成。多语言 SDK 通信代码也可以通过 gRPC 快速生成，更 Native 。  轻量级实现：采用无状态消费模式，能够大幅降低客户端的实现复杂度。客户端更轻量，采用的应用也更容易被 Serverless化、Mesh 化。  云原生可观测性：客户端实现了 OpenTelemetry 标准，能够支持以 OpenTelemetry 形式导出 Metrics 与 Tracing。 RocketMQ 5.0 的另一个重大升级是引入了全新的无状态消费模型。该消费模型完全构建在原先的队列模型之上。队列模型是与存储模型一致的消费模型，消费者完全按照队列做负载均衡，也按照队列做消息拉取，非常适合批量高速拉取以及对单条消息状态不敏感的场景，比如流计算等。 RocketMQ 5.0 推出了 PoP 机制，巧妙地在队列模型之上构建了消息模型，实现了鱼与熊掌兼得。在此消息模型的设计上，业务可以只关心消息而无需关心队列，所有 API 都能够支持单条消息级别的消费、重试、修改不可见时间、删除。 在消息模型下，消息发送过来被存储后，即对消费者可见。消费者通过 Receive Message API 对消息进行消费后，消息进入定时不可见状态。消息超时过后又会重新处于可见状态，能被其他消费者继续消费。某消费者确认消息后，服务端会对该消息进行删除，随即不可见。 基于消息系模型的消费流程下，API 完全面向消息而不是面向队列。而当 PoP 机制遇见了无状态 Proxy，除了存储层，其他节点都是无状态的；客户端、连接和消费也是无状态的，可任意在 Proxy 节点上飘移，真正做到轻量级。 经过重构，RocketMQ 5.0 的可观测性也走向了云原生标准。 Metrics 侧： 指标涵盖丰富：设计了更丰富的指标，包含消息量、堆积量、各个阶段耗时等指标，每个指标从实例、Topic、消费 GroupID 多维度做聚合和展示。 消息团队实践模板：为用户提供实践模板，并持续迭代更新。 Prometheus + Grafana：Prometheus 标准数据格式，利用 Grafana 展示。除了模板，用户也可以自定义展示大盘。 Tracing 侧： OpenTelemetry Tracing 标准：RocketMQ Tracing 标准已经合并到 OpenTelemetry 开源标准，提供了规范和丰富的 messaging tracing 场景定义。 消息领域定制化展示：按照消息维度重新组织抽象的请求 span数据，展示一对多的消费，多次消费信息直观且方便理解。 可衔接 tracing 链路上下游：消息的 tracing 可继承调用上下文，补充到完整的调用链路中，消息链路信息串联了异步链路的上游和下游链路信息。 Logging 侧： Error Code 标准化：不同的错误有唯一的 Error Code。 Error Message 完整：包含完整的错误信息和排序所需要的资源信息。 Error Level 标准化：细化了各种不同错误信息的日志级别，用户可根据 Error、Warn 等级别配置更适合的监控告警。 弹性方面，RocketMQ 5.0 商业版能够充分撬动云的计算、存储和网络的池化资源。比如在计算方面，RocketMQ 5.0 所有工作负载完全部署在 ACK 之上，充分利用了 ACK 弹性能力，撬动 ACK 弹性资源。主要依赖 ACK 的两项技术，一是弹性资源池，另一个是 HPA 支持计算能力快速弹性。同时也会在 ACK 之上做跨可用区部署以提供高可用保障。 网络层面，RocketMQ 5.0 也会充分利用阿里云网络设施，为用户提供更便捷的网络访问能力。比如 RocketMQ 5.0 实例能够支持公网随开随用，需要依赖公网做测试的时候即开即用，测试完立即关闭，安全与方便兼具。同时支持多种私网类型的网络形态，包括 Single Tunnel、Private Link，另外也基于 CEN 构建了全球互通设计网络。 存储方面，RocketMQ 5.0 商业版率先引入多级存储概念，基于 OSS 构建二级存储，能够充分利用 OSS 存储的弹性能力，存储计费也转向了按量付费。而用户能够在 RocketMQ 之上自定义消息存储时长，比如将消息从 3 天有效时长延长至 30 天，能够真正将消息变为数据资产。同时利用二级存储能力，将冷热数据分离，为用户提供一致的冷读 SLA 。 RocketMQ 5.0 商业版发布预告 RocketMQ 4.0 历经了五年发展，开源和商业版本共同迈入了 5.0 时代。7 月底，阿里云消息队列将会基于开源版发布全新的 5.0 商业化版本。注：截止发稿前，RocketMQ 5.0 已经在阿里云消息队列 RocketMQ 产品上全新发布，目前支持国内主要地域。 RocketMQ 5.0 版相对于 4.0 版实例主要有以下几大改变： 第一，新版本、新售卖，更便宜。新版本采取了全新计量方式，有包年、包月型，也有按量付费和公网流量弹性计费。也有更全的售卖体系，比如新增专业版实例，能够满足部分用户需求。同时每个商品系列都新增了测试环境专用实例，能够方便用户以低成本的方式搭建自己的开发环境。 第二，更强弹性，降本提效利器。存储完全走向弹性，能够通过 Serverless 按需使用，按量付费。预留弹性，实例基础规格支持实时升降配，用户可以很方便地在流量到来之前做弹性。此外，专业版支持突发流量弹性，能够解决线上稳定性风险。 第三，全新架构，增强可观测运维。无状态消息消费模型能够解决一些老版本的痛点。同时在可观测上全面采取了云原生接入栈。 消息的全新形态：事件总线 EventBridge 事件总线 EventBridge 已经开源到 RocketMQ 社区中。云原生时代，事件无处不在，云计算资源散落在各地，各类生态孤岛随处可见。因此，以事件和事件驱动的方式来集成这一切是大势所趋。 基于此，阿里云推出了全新事件型产品 EventBridge。该产品构建在 RocketMQ 之上，是 RocketMQ 之上的一个事件驱动架构实践。 EventBridge 的事件源包括阿里云服务的管控事件比如资源变更事件、审计事件、配置变更事件，阿里云服务的数据事件，也包括自定义应用、SaaS 应用、自建数据平台、其他云厂商服务等。 事件经过 EventBridge 处理后会投递到事件目标，事件目标包括函数计算、消息服务、自建网关、HTTP(S)、短信、邮箱、钉钉等。 事件源到事件目标之间会经历完整的事件处理，包括事件源接入到 EB 后，可以对事件进行过滤、转换、归档、回放等。事件在 EventBridge 整个流程中也有完善的可观测性设计，包括事件查询、链路追踪。事件的接入方式非常丰富，可以通过 OpenAPI 来接入、7 种多语言 SDK、CloudEvents SDK、Web Console 和 Webhook 。 EventBridge 具有如下特点： 能够大幅度减少用户开发成本，用户无需额外开发，通过创建 EventBridge 源、事件目标、事件规则等资源即可实现事件架构。用户可以编写事件规则，对事件做过滤、转换。  提供原生 CloudEvents 支持，拥抱 CNCF 社区，能够无缝对接社区 SDK 。标准协议也能统一个阿里云事件规范。  事件 Schema 支持：能够支持事件 Schema 自动探测和校验，支持 Source 和 Target 的 Schema 绑定。  全球事件任意互通：组建了全球事件任意互通网络，组件了跨地域、跨账户的事件网络，能够支持跨云、跨数据中心的事件路由。 EventBridge在云上生态已经初具规模，已经集成了 255+ 云产品事件源和 1000+ 事件类型。 EventBridge率先对消息生态做了融合。阿里云的消息产品矩阵生态均通过 EventBridge 做了完全融合。任何一款消息产品与另一款消息产品的数据都能互通。同时，依靠 EventBridge 的全球事件网络，能够为所有消息产品赋予全球消息路由的能力。 EventBridge 目前已经在内部接入钉钉 ISV、聚石塔 ISV，外部也有 50+ SaaS 系统可以通过 Webhook 的方式接入。另外，海量事件源可以触达 10 多种事件目标，已经对接了全系云产品 API ，任何事件都可以驱动全量云产品 API。 作者介绍： 周新宇 Apache Member，Apache RocketMQ PMC Member，阿里云消息队列 RocketMQ 研发负责人。

2022年，RocketMQ5.0的正式版发布。相对于4.0版本而言，架构走向云原生化，并且覆盖了更多业务场景。 一、消息队列演进史 操作系统、数据库、中间件是基础软件的三驾马车，而消息队列属于最经典的中间件之一，已经有30多年的历史。消息队列的发展主要经历了以下几个阶段： 第一阶段（19802000年）：80年代诞生了第一款消息队列The Information Bus，第一次提出发布订阅模式来解决软件之间的通信问题；90年代是国际商业软件巨头的时代，IBM、Oracle、Microsoft纷纷推出自己的MQ，其中最具代表性的为IBM MQ，价格昂贵，面向高端企业，主要是大型金融、电信等企业。该类商业MQ一般采用高端硬件，软硬件一体机交付，MQ本身的软件架构为单机架构。 第二阶段（2000~2007年）：进入00年代后，初代开源消息队列崛起，诞生了JMS、AMQP两大标准，与之对应的两个实现分别为ActiveMQ、RabbitMQ，他们引领了初期的开源消息队列技术。开源极大促进了消息队列的流行，降低了使用门槛，技术普惠化，逐渐成为企业级架构的标配。相比于今天而言，这类MQ主要面向传统企业级应用和小流量场景，横向扩展能力较弱。 第三阶段（2007~2017年）：PC互联网、移动互联网爆发式发展。由于传统的消息队列无法承受亿级用户的访问流量与海量数据传输，诞生了互联网消息中间件，核心能力是全面采用分布式架构，具备很强的横向扩展能力，开源典型代表有Kafka、RocketMQ，闭源的有淘宝Notify。Kafka的诞生还将消息中间件从消息领域延伸到了流领域，从分布式应用的异步解耦场景延伸到大数据领域的流存储与流计算场景。 第四阶段（2014~至今）：云计算、IoT、大数据引领了新的浪潮。 二、互联网时代的RocketMQ 阿里的电商系统最初是个庞大的单体巨石应用，在研发效率、稳定性方面都无法满足淘宝和天猫飞速的发展。为了解决问题，2008年，淘宝与天猫发起了一次最大规模的架构升级，启动了“五彩石”项目，将单体应用拆分为分布式应用，同时抽象淘宝、天猫的共同底座——业务中台，包括交易中心、商品中心、买家中心等。在业务中台之下，同时诞生了阿里中间件（初期三大件包括消息、RPC、分布式数据层），RocketMQ是其中之一。 虽然在当时业界已经存在不少商业或开源的消息队列，比如IBMMQ、ActiveMQ、RabbitMQ，但无一例外，它们都诞生于传统企业级应用的场景，无法承受互联网对于高并发、无限扩展的苛刻要求。以RabbitMQ为例，RabbitMQ的队列流量与存储负载都为单机，无法满足业务横向扩展的需求。当时另一款具备无限横向扩展能力的消息队列是Kafka，但其主要用于日志类场景，未经过大规模核心业务稳定性验证，而且偏向于简单的log型消息队列，无法满足电商对于复杂消息功能特性的诉求，比如消息过滤、延迟消息等。 另一方面，传统的消息队列无法解决电商业务对于分布式一致性的要求。通过消息队列实现应用异步解耦后，电商业务还需要保障不同上下游应用对于订单状态要达成最终一致，否则会产生大量脏数据，造成业务错误。 大规模的电商系统，既要高性能又要一致性，传统的分布式事务技术束手无策。比如IBM MQ虽然可以使用XA事务来满足分布式一致性的功能诉求，但是XA带来的延迟与成本，对于海量的互联网流量难以承受。 为了解决电商业务对于消息队列的高性能、一致性、无限扩展等需求，自研消息队列成为了当时阿里唯一的出路，最终互联网消息队列RocketMQ应运而生。 为了支持超大规模的复杂电商业务，RocketMQ面向四个方面进行了重点建设，形成了四大优势能力。 ① 支撑超大规模复杂业务的能力，具备丰富的消息特性。 每一个大型互联网公司都会有主营业务（比如阿里是交易、蚂蚁是支付、饿了么是外卖），以主营业务为中心扩展业务能力，阿里电商是围绕交易事件建设的电商操作系统，每笔交易事件都会触发不同的业务，不同细分业务会关注不同类型的交易事件，比如垂直市场只关注某个类目的交易事件、天猫超市只关注某个卖家的交易事件、购物车只关注下单成功的交易事件等。 RocketMQ的SQL订阅提供灵活的消息过滤能力，能够满足下游消费者按照不同的业务维度进行消息过滤的诉求。 在大型互联网业务中，还会有各种定时事件触发场景，最典型的是交易超时关闭机制，阿里交易或者12306订票都有类似的机制。RocketMQ的定时消息能够很方便的满足这类诉求。 ② 一致性。 无论是阿里交易还是蚂蚁支付，都天然对数据一致性有着极高要求，RocketMQ在一致性方面也打造了多个关键特性。最具代表性的是分布式事务消息，RocketMQ是第一个实现该种特性的消息队列，能够保障交易的上下游对于订单状态达到最终一致。该方案也成为异步消息一致性方案的事实标准，被多个互联网公司所采纳，甚至也有公司将移植到定制版的Kafka种。除了分布式一致性之外，RocketMQ还提供了顺序消息的特性，满足顺序一致性的需求。 ③ 稳定性。 稳定性是交易与金融场景的基石特性，也是RocketMQ的根本。RocketMQ除了具备核心服务的HA之外，还具备了全局高可用能力，在阿里内部支持同城多活、异地多活、中心容灾等高阶HA能力。同时，稳定性也不局限于数据与服务的高可用，RocketMQ从产品层面对稳定性进行了全方位的建设，如消息轨迹、消息回溯、消息死信机制。 ④ 高性能。 在双十一的极限流量下，RocketMQ写消息延迟4个9在1ms内，100%在100ms内。RocketMQ采用sharednothing分布式架构，在吞吐量方面也具备无限扩展的能力，已经连续10年支持了双十一万亿级消息洪峰，为百万级的应用实例提供低延迟消息服务。互联网的故事还在进行，云计算规模化落地的时代悄然而来。 三、云计算时代的RocketMQ5.0 2015年，RocketMQ的首个云消息服务在阿里云上线，开启了大规模的云计算实践的序幕。同时RocketMQ也是业界第一个提供公有云服务的开源消息队列。 在大规模的云计算业务场景下，RocketMQ面临着全新的挑战与机遇。 多样性：它不再仅服务于某一家公司的内部业务，不再局限于互联网或金融企业，需要实现全行业、全场景的覆盖。 标准化：对于服务企业内部的自研消息队列而言，无需考虑协议或API的标准化。但是对于云消息服务而言，因为服务对象是外部企业客户，据信通院统计，80%以上的企业客户已经采纳开源技术和标准技术。因此，作为一款云消息服务，需要提供对业界的事实标准协议、接口、SDK的兼容，才能保证客户平滑上云，同时打消客户技术绑定的担忧。 云原生：云原生理念深入人心，消息队列要更好地帮助客户实现云原生应用架构，为业务降本提效。 新趋势：各种新技术的兴起，包括IoT、5G、边缘计算、事件驱动，还有事件流技术。面向技术的新趋势与多样化的业务需求，RocketMQ进行了自我进化，演进到5.0版本。 为了充分释放云的技术红利，RocketMQ5.0在技术架构上进行了云原生的演进。从客户端到服务端都进行了全方位的改造，更高弹性、可用性、更低成本。 客户端采用轻量SDK设计理念，将原来富客户端的逻辑下沉到Broker，满足现代化应用轻量化、Serverless的趋势。 Broker彻底进行弹性架构改造，分离RocketMQ Proxy与Store层，其中Proxy是完全无状态的计算节点，专注多协议、多领域场景覆盖，可以面向不同工作负载独立弹性，如物联网、微服务、大数据不同场景有不同的资源诉求。Store层则专注消息的高可用存储，包括副本复制、主备切换与云存储集成。同时对RocketMQ的Topic资源进行三层解耦，面向消息的Topic、面向流的Topic逻辑分片、面向底层存储的Topic物理分片，每一层都可以独立弹性。 在存储层引入了Leaderless的高可用架构，Store节点身份对等，Leaderless化，0外部依赖。多副本策略可定制，可用性+可靠性+成本灵活组合，面向多可用区、多region组建Geo高可用能力。 为了满足云时代多样化的用户需求，RocketMQ5.0从原来的互联网业务消息中间件扩展到"消息、事件、流"超融合处理平台，解锁更全面的能力。 在消息领域，全面拥抱云原生技术，更好的弹性架构与高可用能力。 在事件领域，支持CloudEvent规范，以事件为中心的产品新界面，助力客户建设跨业务、跨组织的数字化商业生态。 在流领域，流存储增强批量特性，大幅度提高数据吞吐量；新增逻辑队列能力，解耦逻辑资源与物理资源，在流场景也具备无缝伸缩能力；新增流数据库RSQLDB，提供实时事件流处理、流分析能力。 RocketMQ基于端云一体化架构实现了完整的物联网消息队列的能力，从原来的连接应用扩展到连接物联网设备。同时RocketMQ5.0也继续保持极简架构的原则，能够以最低的资源消耗、运维成本搭建服务，适合边缘计算。 除了的产品核心能力之外，RocketMQ5.0积极建设开源生态。 一方面是应用架构生态的建设，既有经典的开源项目、规范的集成，比如JMS、AMQP等，也有云原生技术生态的集成，比如CloudEvents、Dapr、Envoy。同时RocketMQ也会进一步发力数据架构生态，全链路集成大数据的摄入、数据存储、数据处理、数据分析组件，从离线大数据到实时大数据。 【活动】一键体验 RocketMQ 六大生产环境 免费试用+30秒一键体验，低门槛、快速、高效、易操作，带你了解“历经万亿级数据洪峰考验”的云消息队列RocketMQ！ 点击阅读原文，立即参与活动！

设计总览 RocketMQ 多级存储旨在不影响热数据读写的前提下将数据卸载到其他存储介质中，适用于两种场景： 1. 冷热数据分离：RocketMQ 新近产生的消息会缓存在 page cache 中，我们称之为热数据；当缓存超过了内存的容量就会有热数据被换出成为冷数据。如果有少许消费者尝试消费冷数据就会从硬盘中重新加载冷数据到 page cache，这会导致读写 IO 竞争并挤压 page cache 的空间。而将冷数据的读取链路切换为多级存储就可以避免这个问题； 2. 延长消息保留时间：将消息卸载到更大更便宜的存储介质中，可以用较低的成本实现更长的消息保存时间。同时多级存储支持为 topic 指定不同的消息保留时间，可以根据业务需要灵活配置消息 TTL。 RocketMQ 多级存储对比 Kafka 和 Pulsar 的实现最大的不同是我们使用准实时的方式上传消息，而不是等一个 CommitLog 写满后再上传，主要基于以下几点考虑： 1. 均摊成本：RocketMQ 多级存储需要将全局 CommitLog 转换为 topic 维度并重新构建消息索引，一次性处理整个 CommitLog 文件会带来性能毛刺； 2. 对小规格实例更友好：小规格实例往往配置较小的内存，这意味着热数据会更快换出成为冷数据，等待 CommitLog 写满再上传本身就有冷读风险。采取准实时上传的方式既能规避消息上传时的冷读风险，又能尽快使得冷数据可以从多级存储读取。 Quick Start 多级存储在设计上希望降低用户心智负担：用户无需变更客户端就能实现无感切换冷热数据读写链路，通过简单的修改服务端配置即可具备多级存储的能力，只需以下两步： 1. 修改 Broker 配置，指定使用 org.apache.rocketmq.tieredstore.TieredMessageStore 作为 messageStorePlugIn 2. 配置你想使用的储存介质，以卸载消息到其他硬盘为例：配置 tieredBackendServiceProvider 为 org.apache.rocketmq.tieredstore.provider.posix.PosixFileSegment，同时指定新储存的文件路径：tieredStoreFilepath 可选项：支持修改 tieredMetadataServiceProvider 切换元数据存储的实现，默认是基于 json 的文件存储 更多使用说明和配置项可以在 GitHub 上查看多级存储的 技术架构 architecture 接入层：TieredMessageStore/TieredDispatcher/TieredMessageFetcher 接入层实现 MessageStore 中的部分读写接口，并为他们增加了异步语意。TieredDispatcher 和 TieredMessageFetcher 分别实现了多级存储的上传/下载逻辑，相比于底层接口这里做了较多的性能优化：包括使用独立的线程池，避免慢 IO 阻塞访问热数据；使用预读缓存优化性能等。 容器层：TieredCommitLog/TieredConsumeQueue/TieredIndexFile/TieredFileQueue 容器层实现了和 DefaultMessageStore 类似的逻辑文件抽象，同样将文件划分为 CommitLog、ConsumeQueue、IndexFile，并且每种逻辑文件类型都通过 FileQueue 持有底层物理文件的引用。有所不同的是多级存储的 CommitLog 改为 queue 维度。 驱动层：TieredFileSegment 驱动层负责维护逻辑文件到物理文件的映射，通过实现 TieredStoreProvider 对接底层文件系统读写接口(Posix、S3、OSS、MinIO 等)。目前提供了 PosixFileSegment 的实现，可以将数据转移到其他硬盘或通过 fuse 挂载的对象存储上。 消息上传 RocketMQ 多级存储的消息上传是由 dispatch 机制触发的：初始化多级存储时会将 TieredDispatcher 注册为 CommitLog 的 dispacher。这样每当有消息发送到 Broker 会调用 TieredDispatcher 进行消息分发，TieredDispatcher 将该消息写入到 upload buffer 后立即返回成功。整个 dispatch 流程中不会有任何阻塞逻辑，确保不会影响本地 ConsumeQueue 的构建。 TieredDispatcher TieredDispatcher 写入 upload buffer 的内容仅为消息的引用，不会将消息的 body 读入内存。因为多级储存以 queue 维度构建 CommitLog，此时需要重新生成 commitLog offset 字段 upload buffer 触发 upload buffer 上传时读取到每条消息的 commitLog offset 字段时采用拼接的方式将新的 offset 嵌入到原消息中 上传进度控制 每个队列都会有两个关键位点控制上传进度： 1. dispatch offset：已经写入缓存但是未上传的消息位点 2. commit offset：已上传的消息位点 upload progress 类比消费者，dispatch offset 相当于拉取消息的位点，commit offset 相当于确认消费的位点。commit offset 到 dispatch offset 之间的部分相当于已拉取未消费的消息 消息读取 TieredMessageStore 实现了 MessageStore 中的消息读取相关接口，通过请求中的逻辑位点（queue offset）判断是否从多级存储中读取消息，根据配置（tieredStorageLevel）有四种策略： DISABLE：禁止从多级存储中读取消息； NOT_IN_DISK：不在 DefaultMessageStore 中的消息从多级存储中读取； NOT_IN_MEM：不在 page cache 中的消息即冷数据从多级存储读取； FORCE：强制所有消息从多级存储中读取，目前仅供测试使用。 ${a} 需要从多级存储中读取的消息会交由 TieredMessageFetcher 处理：首先校验参数是否合法，然后按照逻辑位点（queue offset）发起拉取请求。TieredConsumeQueue/TieredCommitLog 将逻辑位点换算为对应文件的物理位点从 TieredFileSegment 读取消息。 ${b} TieredFileSegment 维护每个储存在文件系统中的物理文件位点，并通过为不同存储介质实现的接口从中读取所需的数据。 ${c} 预读缓存 TieredMessageFetcher 读取消息时会预读一部分消息供下次使用，这些消息暂存在预读缓存中 ${d} 预读缓存的设计参考了 TCP Tahoe 拥塞控制算法，每次预读的消息量类似拥塞窗口采用加法增、乘法减的机制控制： 加法增：从最小窗口开始，每次增加等同于客户端 batchSize 的消息量。 乘法减：当缓存的消息超过了缓存过期时间仍未被全部拉取，在清理缓存的同时会将下次预读消息量减半。 预读缓存支持在读取消息量较大时分片并发请求，以取得更大带宽和更小的延迟。 某个 topic 消息的预读缓存由消费这个 topic 的所有 group 共享，缓存失效策略为： 1. 所有订阅这个 topic 的 group 都访问了缓存 2. 到达缓存过期时间 故障恢复 上文中我们介绍上传进度由 commit offset 和 dispatch offset 控制。多级存储会为每个 topic、queue、fileSegment 创建元数据并持久化这两种位点。当 Broker 重启后会从元数据中恢复，继续从 commit offset 开始上传消息，之前缓存的消息会重新上传并不会丢失。 开发计划 面向云原生的存储系统要最大化利用云上存储的价值，而对象存储正是云计算红利的体现。 RocketMQ 多级存储希望一方面利用对象存储低成本的优势延长消息存储时间、拓展数据的价值；另一方面利用其共享存储的特性在多副本架构中兼得成本和数据可靠性，以及未来向 Serverless 架构演进。 tag 过滤 多级存储拉取消息时没有计算消息的 tag 是否匹配，tag 过滤交给客户端处理。这样会带来额外的网络开销，计划后续在服务端增加 tag 过滤能力。 广播消费以及多个消费进度不同的消费者 预读缓存失效需要所有订阅这个 topic 的 group 都访问了缓存，这在多个 group 消费进度不一致的情况下很难触发，导致无用的消息在缓存中堆积。 需要计算出每个 group 的消费 qps 来估算某个 group 能否在缓存失效前用上缓存的消息。如果缓存的消息预期在失效前都不会被再次访问，那么它应该被立即过期。相应的对于广播消费，消息的过期策略应被优化为所有 Client 都读取这条消息后才失效。 和高可用架构的融合 目前主要面临以下三个问题： 1. 元数据同步：如何可靠的在多个节点间同步元数据，slave 晋升时如何校准和补全缺失的元数据； 2. 禁止上传超过 confirm offset 的消息：为了避免消息回退，上传的最大 offset 不能超过 confirm offset； 3. slave 晋升时快速启动多级存储：只有 master 节点具有写权限，在 slave 节点晋升后需要快速拉起多级存储断点续传。

高可用架构演进背景 在分布式系统中不可避免的会遇到网络故障，机器宕机，磁盘损坏等问题，为了向用户不中断且正确的提供服务，要求系统有一定的冗余与容错能力。RocketMQ 在日志，统计分析，在线交易，金融交易等丰富的生产场景中发挥着至关重要的作用，而不同环境对基础设施的成本与可靠性提出了不同的诉求。在 RocketMQ v4 版本中有两种主流高可用设计，分别是主备模式的无切换架构和基于 Raft 的多副本架构（图中左侧和右侧所示）。生产实践中我们发现，两副本的冷备模式下备节点资源利用率低，主宕机时特殊类型消息存在可用性问题；而 Raft 高度串行化，基于多数派的确认机制在扩展只读副本时不够灵活，无法很好的支持两机房对等部署，异地多中心等复杂场景。RocketMQ v5 版本融合了上述方案的优势，提出 DLedger Controller 作为管控节点（中间部分所示），将选举逻辑插件化并优化了数据复制的实现。 如何实现高可用系统 副本组与数据分片 在 PrimaryBackup 架构的分布式系统中，一份数据将被复制成多个副本来避免数据丢失。处理相同数据的一组节点被称为副本组（ReplicaSet），副本组的粒度可以是单个文件级别的（例如 HDFS），也可以是分区级 / 队列级的（例如 Kafka），每个真实存储节点上可以容纳若干个不同副本组的副本，也可以像 RocketMQ 一样粗粒度的独占节点。独占能够显著简化数据写入时确保持久化成功的复杂度，因为每个副本组上只有主副本会响应读写请求，备机一般配置只读来提供均衡读负载，选举这件事儿等价于让副本组内一个副本持有独占的写锁。 RocketMQ 为每个存储数据的 Broker 节点配置 ClusterName，BrokerName 标识来更好的进行资源管理。多个 BrokerName 相同的节点构成一个副本组。每个副本还拥有一个从 0 开始编号，不重复也不一定连续的 BrokerId 用来表示身份，编号为 0 的节点是这个副本组的 Leader / Primary / Master，故障时通过选举来重新对 Broker 编号标识新的身份。例如 BrokerId = {0, 1, 3}，则 0 为主，其他两个为备。 一个副本组内，节点间共享数据的方式有多种，资源的共享程度由低到高来说一般有 Shared Nothing，Shared Disk，Shared Memory，Shared EveryThing。典型的 Shared Nothing 架构是 TiDB 这类纯分布式的数据库，TiDB 在每个存储节点上使用基于 RocksDB 封装的 TiKV 进行数据存储，上层通过协议交互实现事务或者 MVCC。相比于传统的分库分表策略来说，TiKV 易用性和灵活程度很高，更容易解决数据热点与伸缩时数据打散的一系列问题，但实现跨多节点的事务就需要涉及到多次网络的通信。另一端 Shared EveryThing 的案例是 AWS 的 Aurora，Aliyun 的 PolarStore，旁路 Kernal 的方式使应用完全运行于用户态，以最大程度的存储复用来减少资源消耗，一主多备完全共用一份底层可靠的存储，实现一写多读，快速切换。 大多数 KV 操作都是通过关键字的一致性哈希来计算所分配的节点，当这个节点所在的主副本组产生存储抖动，主备切换，网络分区等情况下，这个分片所对应的所有键都无法更新，局部会有一些操作失败。消息系统的模型有所不同，流量大但跨副本组的数据交互极少，无序消息发送到预期分区失败时还可以向其他副本组（分片）写入，一个副本组的故障不影响全局，这在整体服务的层面上额外提供了跨副本组的可用性。此外，考虑到 MQ 作为 Paas 层产品，被广泛部署于 Windows，Linux on Arm 等各种环境，只有减少和 Iaas 层产品的深度绑定，才能提供更好的灵活性。这种局部故障隔离和轻依赖的特性是 RocketMQ 选则 Shared Nothing 模型重要原因。 副本组中，各个节点处理的速度不同，也就有了日志水位的概念。Master 和与其差距不大的 Slave 共同组成了同步副本集（SyncStateSet）。如何定义差距不大呢？衡量的指标可以是日志水位（文件大小）差距较小，也可以是备落后的时间在一定范围内。在主宕机时，同步副本集中的其余节点有机会被提升为主，有时需要对系统进行容灾演练，或者对某些机器进行维护或灰度升级时希望定向的切换某一个副本成为新主，这又产生了优先副本（PriorityReplica）的概念。选择优先副本的原则和策略很多，可以动态选择水位最高，加入时间最久或 CommitLog 最长的副本，也可以支持机架，可用区优先这类静态策略。 从模型的角度来看，RocketMQ 单节点上 Topic 数量较多，如果像 kafka 以 topic / partition 粒度维护状态机，节点宕机会导致上万个状态机切换，这种惊群效应会带来很多潜在风险，因此 v4 版本时 RocketMQ 选择以单个 Broker 作为切换的最小粒度来管理，相比于其他更细粒度的实现，副本身份切换时只需要重分配 Broker 编号，对元数据节点压力最小。由于通信的数据量少，可以加快主备切换的速度，单个副本下线的影响被限制在副本组内，减少管理和运维成本。这种实现也一些缺点，例如存储节点的负载无法以最佳状态在集群上进行负载均衡，Topic 与存储节点本身的耦合度较高，水平扩展一般会改变分区总数，这就需要在上层附加额外的处理逻辑。 为了更规范更准确的衡量副本组的可用性指标，学术上就引入了几个名词： RTO（Recovery Time Objective）恢复时间目标，一般表示业务中断到恢复的时间。 RPO（Recovery Point Object）恢复点目标，用于衡量业务连续性。例如某个硬盘每天备份，故障时丢失最近备份后的所有更新。 SLA（ServiceLevel Agreement）服务等级协议，厂商以合约的形式对用户进行服务质量承诺，SLA 越高通常成本也越高。 节点数量与可靠性关系密切，根据不同生产场景，RocketMQ 的一个副本组可能会有 1，2，3，5 个副本。 1. 单副本成本最低，维护最简单，宕机时其他副本组接管新消息的写入，但已写入的数据无法读取，造成部分消息消费延迟。底层硬件故障还可能导致数据永久丢失，一般用于非关键日志，数据采集等低可靠性成本诉求较强的场景。 2. 两副本较好的权衡了数据冗余的成本与性能，RocketMQ 跨副本组容灾的特性使得两副本模式适用于绝大部分 IOPS 比较高的场景。此时备机可以分摊一定的读压力（尤其是主副本由于内存紧张或者产生冷读时）。两副本由于不满足多数派（quorum）原则，没有外部系统的参与时，故障时无法进行选举切换。 3. 三副本和五副本是业界使用最为广泛的，精心设计的算法使得多数情况下系统可以自愈。基于 Paxos / Raft 属于牺牲高可用性来保证一致性的 CP 型设计，存储成本很高，容易受到 IO 分布不均匀和水桶效应的影响。每条数据都需要半数以上副本响应的设计在需要写透（write through）多副本的消息场景下不够灵活。 日志复制还是消息复制 如何保证副本组中数据的最终一致性？那肯定是通过数据复制的方式实现，我们该选择逻辑复制还是物理复制呢？ 逻辑复制：使用消息来进行同步的场景也很多，各种 connector 实现本质上就是把消息从一个系统挪到另外一个系统上，例如将数据导入导出到 ES，Flink 这样的系统上进行分析，根据业务需要选择特定 Topic / Tag 进行同步，灵活程度和可扩展性非常高。这种方案随着 Topic 增多，系统还会有服务发现，位点和心跳管理等上层实现造成的性能损失。因此对于消息同步的场景，RocketMQ 也支持以消息路由的形式进行数据转移，将消息复制作为业务消费的特例来看待。 物理复制：大名鼎鼎的 MySQL 对于操作会记录逻辑日志（bin log）和重做日志（redo log）两种日志。其中 bin log 记录了语句的原始逻辑，比如修改某一行某个字段，redo log 属于物理日志，记录了哪个表空间哪个数据页改了什么。在 RocketMQ 的场景下，存储层的 CommitLog 通过链表和内核的 MappedFile 机制抽象出一条 append only 的数据流。主副本将未提交的消息按序传输给其他副本（相当于 redo log），并根据一定规则计算确认位点（confirm offset）判断日志流是否被提交。这种方案仅使用一份日志和位点就可以保证主备之间预写日志的一致性，简化复制实现的同时也提高了性能。 为了可用性而设计的多副本结构，很明显是需要对所有需要持久化的数据进行复制的，选择物理复制更加节省资源。RocketMQ 在物理复制时又是如何保证数据的最终一致性呢？这就涉及到数据的水位对齐。对于消息和流这样近似 FIFO 的系统来说，越近期的消息价值越高，消息系统的副本组的单个节点不会像数据库系统一样，保留这个副本的全量数据，Broker 一方面不断的将冷数据规整并转入低频介质来节约成本，同时对热数据盘上的数据也会由远及近滚动删除。如果副本组中有副本宕机较久，或者在备份重建等场景下就会出现日志流的不对齐和分叉的复杂情况。在下图中我们将主节点的 CommitLog 的首尾位点作为参考点，这样就可以划分出三个区间。在下图中以蓝色箭头表示。排列组合一下就可以证明备机此时的 CommitLog 一定满足下列 6 种情况之一。 下面对每种情况进行讨论与分析： 11 情况下满足备 Max 主 Max，可能由于主异步写磁盘宕机后又成为主，或者网络分区时双主写入造成 CommitLog 分叉。由于新主落后于备，少量未确认的消息丢失，非正常模式的选举（RocketMQ 将这种情况称为 unclean 选举）是应该尽量避免的。 33 理论上不会出现，备的数据长于主，原因可能是主节点数据丢失又叠加了非正常选举，因此这种情况需要人工介入处理。 租约与节点身份变更 前文提到 RocketMQ 每个副本组的主副本才接受外部写请求，节点的身份又是如何决定的呢？ 分布式系统一般分为中心化架构和去中心化架构。对于 MultiRaft，每个副本组包含三个或者五个副本，副本组内可以通过 Paxos / Raft 这样的共识协议来进行选主。典型的中心化架构，为了节省数据面资源成本会部署两副本，此时依赖于外部 ZK，ETCD，或者 DLedger Controller 这样的组件作为中心节点进行选举。由外置组件裁决成员身份涉及到分布式中两个重要的问题：1. 如何判断节点的状态是否正常。2. 如何避免双主问题。 对于第一个问题，kubernetes 的解决方案相对优雅，k8s 对与 Pod 的健康检查包括存活检测（Liveness probes）和就绪检测（Readiness probes），Liveness probes 主要是探测应用是否还活着，失败时重启 Pod。Readiness probes 来判断探测应用是否接受流量。简单的心跳机制一般只能实现存活检测，来看一个例子：假设有副本组中有 A、B、C 三个副本，另有一个节点 Q（哨兵） 负责观测节点状态，同时承担了全局选举与状态维护的职责。节点 A、B、C 周期性的向 Q 发送心跳，如果 Q 超过一段时间（一般是两个心跳间隔 ）收不到某个节点的心跳则认为这个节点异常。如果异常的是主副本，Q 将副本组的其他副本提升为主并广播告知其他副本。 在工程实践中，节点下线的可能性一般要小于网络抖动的可能性。我们假设节点 A 是副本组的主，节点 Q 与节点 A 之间的网络中断。节点 Q 认为 A 异常。重新选择节点 B 作为新的 Master，并通知节点 A、B、C 新的 Master 是节点 B。节点 A 本身工作正常，与节点 B、C 之间的网络也正常。由于节点 Q 的通知事件到达节点 A、B、C 的顺序是未知的，假如先达到 B，在这一时刻，系统中同时存在两个工作的主，一个是 A，另一个是 B。假如此时 A、B 都接收外部请求并与 C 同步数据，会产生严重的数据错误。上述 "双主" 问题出现的原因在于虽然节点 Q 认为节点 A 异常，但节点 A 自己不认为自己异常，在旧主新主都接受写入的时候就产生了日志流的分叉，其问题的本质是由于网络分区造成的系统对于节点状态没有达成一致。 租约是一种避免双主的有效手段，租约的典型含义是现在中心节点承认哪个节点为主，并允许节点在租约有效期内正常工作。如果节点 Q 希望切换新的主，只需等待前一个主的租约过期，则就可以安全的颁发新租约给新 Master 节点，而不会出现双主问题。这种情况下系统对 Q 本身的可用性诉求非常高，可能会成为集群的性能瓶颈。生产中使用租约还有很多实现细节，例如依赖时钟同步需要颁发者的有效期设置的比接收者的略大，颁发者本身的切换也较为复杂。 在 RocketMQ 的设计中，希望以一种去中心化的设计降低中心节点宕机带来的全局风险，（这里认为中心化和是否存在中心节点是两件事）所以没有引入租约机制。在 Controller （对应于 Q ）崩溃恢复期间，由于 Broker 对自己身份会进行永久缓存，每个主副本会管理这个副本组的状态机，RocketMQ Dledger Controller 这种模式能够尽量保证在大部分副本组在哨兵组件不可用时仍然不影响收发消息的核心流程。而旧主由于永久缓存身份，无法降级导致了网络分区时系统必须容忍双主。产生了多种解决方案，用户可以通过预配置选择 AP 型可用性优先，即允许系统通过短时分叉来保障服务连续性（下文还会继续谈谈为什么消息系统中分叉很难避免），还是 CP 型一致性优先，通过配置最小副本 ack 数超过集群半数以上节点。此时发送到旧主的消息将因为无法通过 ha 链路将数据发送给备，向客户端返回超时，由客户端将发起重试到其他分片。客户端经历一个服务发现的周期之后，客户端就可以正确发现新主。 特别的，在网络分区的情况下，例如旧主和备，Controller 之间产生网络分区，此时由于没有引入租约机制，旧主不会自动降级，旧主可以配置为异步双写，每一条消息需要经过主备的双重确认才能向客户端返回成功。而备在切换为主时，会设置自己只需要单个副本确认的同步写盘模式。此时，客户端短时间内仍然可以向旧主发送消息，旧主需要两副本确认才能返回成功，因此发送到旧主的消息会返回 SLAVE_NOT_AVAILABLE 的超时响应，通过客户端重试将消息发往新的节点。几秒后，客户端从 NameServer / Controller 获取新的路由时，旧主从客户端缓存中移除，此时完成了备节点的提升。 外置的组件可以对节点身份进行分配，上图展示了一个两副本的副本组上线流程： 1. 多个 Controller 通过选举和对 Broker 的请求进行重定向，最终由一个 Controller 做为主节点进行身份分配。 2. 如果 RocketMQ 副本组存在多个副本且需要选主，节点默认以备的身份启动，备节点会将自己注册到 Controller。 3. 节点从 Controller 获取 BrokerMemberGroup，包含了这个副本组的描述和连接信息。 1. 若分配的身份为备，解析出主节点的对外服务的地址并连接，完成日志截断后进行 HA 同步。 2. 若分配的身份为主，等待备机连接到自身的 HA 端口，并向 NameServer 再次宣告自己是主节点。 4. 主节点维护整个副本组的信息，向备发起数据复制，周期性的向 Controller 汇报主备之间水位差距，复制速度等。 RocketMQ 弱依赖 Controller 的实现并不会打破 Raft 中每个 term 最多只有一个 leader 的假设，工程中一般会使用 Leader Lease 解决脏读的问题，配合 Leader Stickiness 解决频繁切换的问题，保证主的唯一性。 Leader Lease: 租约，上一任 Leader 的 Lease 过期后，等待一段时间再发起 Leader 选举。 Leader Stickiness：Leader Lease 未过期的 Follower 拒绝新的 Leader 选举请求。 _注：Raft 认为具有最新已提交的日志的节点才有资格成为 Leader，而 MultiPaxos 无此限制。_ 对于日志的连续性问题，Raft 在确认一条日志之前会通过位点检查日志连续性，若检查到日志不连续会拒绝此日志，保证日志连续性，MultiPaxos 允许日志中有空洞。Raft 在 AppendEntries 中会携带 Leader 的 commit index，一旦日志形成多数派，Leader 更新本地的 commit index（对应于 RocketMQ 的 confirm offset）即完成提交，下一条 AppendEntries 会携带新的 commit index 通知其它节点，MultiPaxos 没有日志连接性假设，需要额外的 commit 消息通知其它节点。 计算日志分叉位点 除了网络分区，很多情况导致日志数据流分叉。有如下案例：三副本采用异步复制，异步持久化，A 为旧主 B C 为备，切换瞬间 B 日志水位大于 C，此时 C 成为新主，B C 副本上的数据会产生分叉，因为 B 还多出了一段未确认的数据。那么 B 是如何以一个简单可靠的方法去判断自己和 C 数据分叉的位点？ 一个直观的想法就是，直接将主备的 CommitLog 从前向后逐渐字节比较，一般生产环境下，主备都有数百 GB 的日志文件流，读取和传输大量数据的方案费时费力。很快我们发现，确定两个大文件是否相同的一个好办法就是比较数据的哈希值，需要对比的数据量一下子就从数百 GB 降低为了几百个哈希值，对于第一个不相同的 CommitLog 文件，还可以采取局部哈希的方式对齐，这里仍然存在一些计算的代价。还有没有优化的空间呢，那就是利用任期 Epoch 和偏移量 StartOffset 实现一个新的截断算法。这种 EpochStartOffset 满足如下原则： 1. 通过共识协议保证给定的一个任期 Epoch 只有一个Leader。 2. 只有 Leader 可以写入新的数据流，满足一定条件才会被提交。 3. Follower 只能从 Leader 获取最新的数据流，Follower 上线时按照选举算法进行截断。 下面是一个选举截断的具体案例，选举截断算法思想和流程如下： 主 CommitLog Min = 300，Max = 2500，EpochMap = {, , }备 CommitLog Min = 300，Max = 2500，EpochMap = {, , } 1. 备节点连接到主节点进行 HA 协商，获取主节点的 EpochStartOffset 信息并比较 2. 备从后向前找到任期起始点相同的那个点作为分叉任期，在上述案例里是 3. 选择这个任期里主备结束位点的最小值（如果主副本没有切换且为最大任期，则主副本的结束位点是无穷大） 实现的代码如下： ${e} 数据回发与日志截断 故障发生后，系统将会对分叉数据进行修复，有很多小小细节值得深究与探讨。 在实现数据截断的过程中，有一个很特殊的动作，当备切主的时候要把 ConsumeQueue 的 Confirm Offset 提升到 CommitLog 的 MaxPhyOffset，即使这一部分数据在主上是否被提交是未知的。回想起几年前看 Raft 的时候，当一条日志被传输到 Follower，Follower 确认收到这条消息，主再把这条日志应用到自己的状态机时，通知客户端和通知所有的 follower 去 commit 这条日志这两件事是并行的，假如 leader 先回复 client 处理成功，此时 leader 挂了，由于其他 follower 的确认位点 confirm offset 一般会略低于 leader，中间这段未决日志还没应用到 follower 的状态机上，这时就出现了状态机不一致的情况，即已经写入 leader 的数据丢失了。让我们来举一个具体的案例，假设两副本一主一备： 1. 主的 max offset = 100，主向备发送当前 confirm offset = 40 和 message buffer = [40100] 的数据 2. 备向主回复 confirm offset = 100 后主需要同时做几件事 1. 本地提交（apply） [40100] 区间的数据，用后台的 dispatch 线程异步构建这段数据的索引 2. 向 producer 响应 [40100] 这段数据是发送成功的。 3. 向多个备机异步的提交，实际上是发送了 confirm offset = 100 3. 此时主突然宕机，备机的 confirm offset 可能是 [40100] 中的值 所以当备切换为主的时候，如果直接以 40 进行截断，意味着客户端已经发送到服务端的消息丢失了，正确的水位应该被提升至 100。但是备还没有收到 2.3 的 confirm = 100 的信息，这个行为相当于要提交了未决消息。事实上新 leader 会遵守 "Leader Completeness" 的约定，切换时任何副本都不会删除也不会更改旧 leader 未决的 entry。新 leader 在新的 term 下，会直接应用一个较大的版本将未决的 entry 一起提交，这里副本组主备节点的行为共同保证了复制状态机的安全性。 那么备切换成功的标志是什么，什么时候才能接收 producer 新的流量呢？对于 Raft 来说一旦切换就可以，对于 RocketMQ 来说这个阶段会被稍稍推迟，即索引已经完全构建结束的时候。RocketMQ 为了保证构建 consume queue 的一致性，会在 CommitLog 中记录 consume queue offset 的偏移量，此时 confirm offset 到 max offset 间的数据是副本作为备来接收的，这部分消息在 consume queue 中的偏移量已经固定下来了，而 producer 新的流量时由于 RocketMQ 预计算位点的优化，等到消息实际放入 CommitLog 的再真实的数据分发（dispatch）的时候就会发现对应位置的 consume queue 已经被占用了，此时就造成了主备索引数据不一致。本质原因是 RocketMQ 存储层预构建索引的优化对日志有一些侵入性，但切换时短暂等待的代价远远小于正常运行时提速的收益。 消息中间件场景 a. 元数据变更是否依赖于日志 目前 RocketMQ 对于元数据是在内存中单独管理的，备机间隔 5 秒向当前的主节点同步数据。例如当前主节点上创建了一个临时 Topic 并接受了一条消息，在一个同步周期内这个 Topic 又被删除了，此时主备节点元数据可能不一致。又比如位点更新的时候，对于单个队列而言，多副本架构中存在多条消费位点更新链路，Consumer 拉取消息时更新，Consumer 主动向 broker 更新，管控重置位点，HA 链路更新，当副本组发生主备切换时，consumer group 同时发生 consumer 上下线，由于路由发现的时间差，还可能造成同一个消费组两个不同 consumer 分别消费同一副本组主备上同一个队列的情况。 原因在于备机重做元数据更新和消息流这两件事是异步的，这有一定概率会造成脏数据。由于 RocketMQ 单个节点上 Topic / Group 数量较多，通过日志的实现会导致持久化的数据量很大，在复杂场景下基于日志做回滚依赖 snapshot 机制也会增加计算开销和恢复时间。这个问题和数据库很像，MySQL 在执行 DDL 修改元数据时通过会创建 MDL 锁，阻塞用户其他操作访问表空间的访问。备库同步主库也会加锁，元数据修改开始点和结束点所代表的两个日志并不是一个原子操作，这意味着主库上在修改元数据的过程中如果宕机了，备库上持有的 MDL 锁就无法释放。MySQL 的解决方案是在主库每次崩溃恢复后，都写一条特殊的日志，通知所有连接的备库释放其持有的所有 MDL 排他锁。对所有操作都走日志流进行状态机复制要求存储层有多种日志类型，实现也更加复杂。RocketMQ 选择以另一种同步的模式操作，即类似 ZAB 这样二阶段协议，例如位点更新时的可以选择配置 LockInStrictMode 让备都同步这条修改。事实上 RocketMQ 为了优化上述位点跳跃的现象，客户端在未重启时，遇到服务端主备切换还会用优先采纳本地位点的方式获取消息，进一步减少重复消费。 b. 同步复制与异步复制 同步复制的含义是用户的一个操作在多个副本上都已经提交。正常情况下，假设一个副本组中的 3 个副本都要对相同一个请求进行确认，相当于数据写透 3 个副本（简称 33 写），33 写提供了非常高的数据可靠性，但是把所有从节点都配置为同步复制时任何一个同步节点的中断都会导致整个副本组处理请求失败。当第三个副本是跨可用区时，长尾也会带来一定的延迟。 异步复制模式下，尚未复制到从节点的写请求都会丢失。向客户端确认的写操作也无法保证被持久化。异步复制是一种故障时 RPO 不为 0 的配置模式，由于不用考虑从节点上的状态，总是可以继续响应写请求，系统的延迟更低，吞吐性能更好。为了权衡两者，通常只有其中一个从节点是同步的，而其他节点是异步的模式。只要同步的从节点变得不可用或性能下降，则将另一个异步的从节点提升为同步模式。这样可以保证至少有两个节点（即主节点和一个同步从节点）拥有最新的数据副本。这种模式称为 23 写，能帮助避免抖动，提供更好的延迟稳定性，有时候也叫称为半同步。 在 RocketMQ 的场景中，异步复制也被广泛应用在消息读写比极高，从节点数量多或者异地多副本场景。同步复制和异步复制是通过 Broker 配置文件里的 brokerRole 参数进行设置的，这个参数可以被设置成 ASYNC_MASTER、SYNC_MASTER、SLAVE 三个值中的一个。实际应用中要结合业务场景合理设置持久化方式和主从复制方式，通常，由于网络的速度高于本地 IO 速度，采用异步持久化和同步复制是一个权衡性能与可靠性的设置。 c. 副本组自适应降级 同步复制的含义是一条数据同时被主备确认才返回用户操作成功，可以保证主宕机后消息还在备中，适合可靠性要求较高的场景，同步复制还可以限制未同步的数据量以减少 ha 链路的内存压力，缺点则是副本组中的某一个备出现假死就会影响写入。异步复制无需等待备确认，性能高于同步复制，切换时未提交的消息可能会丢失（参考前文的日志分叉）。在三副本甚至五副本且对可靠性要求高的场景中无法采用异步复制，采用同步复制需要每一个副本确认后才会返回，在副本数多的情况下严重影响效率。关于一条消息需要被多少副本确认这个问题，RocketMQ 服务端会有一些数量上的配置来进行灵活调整： TotalReplicas：全部副本数 InSyncReplicas：每条消息至少要被这个数量的 Broker 确认（如果主为 ASYNC_MASTER 或者 AllAck 模式则该参数不生效） MinInSyncReplicas：最小的同步副本数，如果 InSyncReplicas 。对于正常情况下，两个副本会处于同步复制，当备下线或假死时，会进行自适应降级，保证主节点还能正常收发消息，这个功能为用户提供了一个可用性优先的选择。 d. 轻量级心跳与快速隔离 在 RocketMQ v4.x 版本的实现中，Broker 周期性的（间隔 30 秒）将自身的所有 Topic 序列化并传输到 NameServer 注册进行保活。由于 Broker 上 Topic 的元数据规模较大，带来了较大的网络流量开销，Broker 的注册间隔不能设置的太短。同时 NameServer 对 Broker 是采取延迟隔离机制，防止 NameServer 网络抖动时可能瞬间移除所有 Broker 的注册信息，引发服务的雪崩。默认情况下异常主宕机时超过 2 分钟，或者备切换为主重新注册后才会替换。容错设计的同时导致 Broker 故障转移缓慢，RocketMQ v5.0 版本引入轻量级心跳（参数liteHeartBeat），将 Broker 的注册行为与 NameServer 的心跳进行了逻辑拆分，将心跳间隔减小到 1 秒。当 NameServer 间隔 5 秒（可配置）没有收到来自 Broker 的心跳请求就对 Broker 进行移除，使异常场景下自愈的时间从分钟级缩短到了秒级。 RocketMQ 高可用架构演进路线 无切换架构的演进 最早的时候，RocketMQ 基于 MasterSlave 模式提供了主备部署的架构，这种模式提供了一定的高可用能力，在 Master 节点负载较高情况下，读流量可以被重定向到备机。由于没有选主机制，在 Master 节点不可用时，这个副本组的消息发送将会完全中断，还会出现延迟消息、事务消息、Pop 消息等二级消息无法消费或者延迟。此外，备机在正常工作场景下资源使用率较低，造成一定的资源浪费。为了解决这些问题，社区提出了在一个 Broker 进程内运行多个 BrokerContainer，这个设计类似于 Flink 的 slot，让一个 Broker 进程上可以以 Container 的形式运行多个节点，复用传输层的连接，业务线程池等资源，通过单节点主备交叉部署来同时承担多份流量，无外部依赖，自愈能力强。这种方式下隔离性弱于使用原生容器方式进行隔离，同时由于架构的复杂度增加导致了自愈流程较为复杂。 切换架构的演进 另一条演进路线则是基于可切换的，RocketMQ 也尝试过依托于 Zookeeper 的分布式锁和通知机制进行 HA 状态的管理。引入外部依赖的同时给架构带来了复杂性，不容易做小型化部署，部署运维和诊断的成本较高。另一种方式就是基于 Raft 在集群内自动选主，Raft 中的副本身份被透出和复用到 Broker Role 层面去除外部依赖，然而强一致的 Raft 版本并未支持灵活的降级策略，无法在 C 和 A 之间灵活调整。两种切换方案都是 CP 设计，牺牲高可用优先保证一致性。主副本下线时选主和路由定时更新策略导致整个故障转移时间依然较长，Raft 本身对三副本的要求也会面临较大的成本压力，RocketMQ 原生的 TransientPool，零拷贝等一些用来避免减少 IO 压力的方案在 Raft 下无法有效使用。 RocketMQ DLedger 融合模式 RocketMQ DLedger 融合模式是 RocketMQ 5.0 演进中结合上述两条路线后的一个系统的解决方案。核心的特性有以下几点： 1. 利用可内嵌于 NameServer 的 Controller 进行选主，无外部依赖，对两副本支持友好。 2. 引入 EpochStartOffset 机制来计算日志分叉位点。 3. 消息在进行写入时，提供了灵活的配置来协调系统对于可用性还是一致性优先的诉求。 4. 简化日志复制协议使得日志复制为高效。 几种实现对比表如下： 与其他消息系统的对比 控制节点 1. 是否强制要求选主 Kafka 的 Controller 是 Broker 选举产生，这需要有一个存储节点间的服务发现机制。RocketMQ 的 Controller 可以作为管控节点单独存在。对 Kafka，Pulsar 而言必须选择主副本进行写入，随着时间的运行节点之间负载需要通过复杂的方案进行再均衡。对 RocketMQ 的融合架构而言，由于选主是可选的，静态布局的方案（例如无需依赖复杂的动态调度就可以较为均衡的实现跨机架跨可用区），并且无切换与切换架构可以相互转换。 2. Controller 的逻辑复杂度 RocketMQ Controller 相比 Kafka Controller 更加轻量，Kafka 的 Controller 承担 Partition 粒度的 ISR 维护和选举等功能，而RocketMQ 的 Controller 维护的数据是副本组粒度的，对于元数据只维护节点身份，更加简单。RocketMQ Controller 可以独立部署，也可以内嵌 NameServer 运行。 3. Controller 依赖程度 RocketMQ Broker 的同步副本集维护是 Master Broker 节点上报，由于不强依赖中心节点来提供租约，controller 宕机时虽然无法为同时有主故障的副本组选举，但不影响绝大部分副本组可用性。Pulsar 中通过 fencing 机制防止有多个 writer（pulsar 中的计算节点称为 broker）同时写同一个 partition，是对外部有依赖的。 数据节点 1. 副本存储结构的抽象与最小粒度不同，在这一点上其实三者的设计各有优势 Kafka 的存储抽象粒度是 Partition，对每个分区进行维护多副本，扩容需要进行数据复制，对于冷读支持更好。 RocketMQ 的日志流是 Broker 粒度的，顺序写盘效率更高，在磁盘空间不足时一般选择水平扩容，只需复制元数据。 Pulsar 其实抽象了一个分布式日志流 Journal，分区被进一步分成分片，根据配置的时间或大小进行滚动，扩容只需复制元数据。 2. 复杂的参数配置被收敛至服务端 Kafka 和 RocketMQ 都支持灵活的配置单条消息的 ack 数，即权衡数据写入灵活性与可靠性。RocketMQ 在向云原生演进的过程希望简化客户端 API 与配置，让业务方只需关心消息本身，选择在服务端配置统一配置这个值。 3. 副本数据的同步方式不同 Pulsar 采用星型写：数据直接从 writer 写到多个 bookeeper。适合客户端与存储节点混部场景。数据路径只需要 1 跳，延迟更低。缺点是当存储计算分离时，星型写需要更多的存储集群和计算集群间网络带宽。 RocketMQ 和 Kafka 采用 Y 型写：client 先写到一个主副本，由其再转发给另外 Broker 副本。虽然服务端内部带宽充裕，但需要 2 跳网络，会增加延迟。Y 型写利于解决文件多客户端写的问题，也更容易利用 23 写克服毛刺，提供更好的延迟稳定性。 高可用架构的未来 仔细阅读 RocketMQ 的源码，其实大家也会发现 RocketMQ 在各种边缘问题处理上细节满满，节点失效，网络抖动，副本一致性，持久化，可用性与延迟之间存在各种细微的权衡，这也是 RocketMQ 多年来在生产环境下所积累的核心竞争力之一。随着分布式技术的进一步发展，更多更有意思的技术，如基于 RDMA 网络的复制协议也呼之欲出。RocketMQ 将与社区协同进步，发展为 “消息，事件，流” 一体化的融合平台。 参考文档： 1. Paxos design： 2. SOFAJRaft： 3. Pulsar Geo Replication： 4. Pulsar Metadata： 5. Kafka Persistence： 6. Kafka Balancing leadership： 7. Windows Azure Storage: A Highly Available Cloud Storage Service with Strong Consistency： 8. PolarDB Serverless: A Cloud Native Database for Disaggregated Data Centers：

前言 回顾 RocketMQ 的发展历程，至今已十年有余。2022 年 RocketMQ 5.0 正式发布，全面迈进云原生时代。 11 月 5 日，2022 杭州 · 云栖大会上，阿里云智能高级产品专家杨秋弟在云原生峰会上发表主题演讲，发布消息队列 RocketMQ 5.0：从消息服务到云原生事件流处理平台。 阿里云智能高级产品专家&Apache RocketMQ 联合创始人 杨秋弟 Apache RocketMQ 发展史 回顾 Apache RocketMQ 过去十年的发展历程，可分为“诞生于互联网”与“成长于云计算”两大阶段。 第一个阶段是 RocketMQ 的从 0 到 1，在阿里内部规模化落地。2012 年，为了支撑超大规模电商互联网架构，阿里中间件研发了 RocketMQ，并在产品诞生初期开源，2017 年 RocketMQ 统一了阿里消息技术体系。 第二个阶段是云计算。2015 年 RocketMQ 上云，这也是业界首个提供公共云 SaaS 形态的开源消息队列；2016 年，阿里把 RocketMQ 捐赠给 Apache，2017 年孵化毕业，成为国内首个 TLP 的互联网中间件。 十年磨一剑，出鞘必锋芒。在这十年的过程中，通过集团打磨稳定性，依托云计算孵化创新，开源共建加速标准化建立与生态连接，RocketMQ 始终坚持开源、集团、商业三位一体的发展思路，内核演进和产品迭代齐头并进。2022 年 RocketMQ 5.0 正式发布宣告着全面迈进云原生时代。 RocketMQ 5.0：从消息服务到云原生事件流平台 回顾过去这十年，RocketMQ 服务于集团几乎所有的业务，在阿里云上更是累计服务了 10 万余家企业客户，覆盖互联网、零售、金融、汽车等 20 多个行业，大规模的生产实践持续累积产品的核心优势。 多样性。企业级应用复杂的业务诉求，催生 RocketMQ 提供丰富的消息类型，比如定时消息、事务消息、顺序消息等等。此外，也提供了像消息轨迹、消息回溯等一系列的消息治理能力。 一致性。无论是淘宝交易还是蚂蚁支付都天然对数据一致性有着极高的要求，RocketMQ 提供的分布式事务消息是业内第一个提供该特性的消息产品，将异步解耦与数据一致性完美融合，是金融客户中不可或缺的产品能力。 稳定性。稳定性是产品的根基，更是一个系统工程，RocketMQ 长期在电商和金融领域中打磨，不仅提供最高达 99.99% SLA，更是帮助客户提供全方位的健康巡检与故障排查能力，如消息轨迹、消息回溯、死信机制等等，提供多样化的稳定性兜底手段。 高性能。在双十一的极限流量下，RocketMQ 具备无限扩展能力，支持千万级并发与万亿级消息洪峰，P9999 写延迟在 1ms 内，100%在 100ms 内。 可以说，在消息领域，尤其在业务消息领域，RocketMQ 在国内已经做到顶尖，成为企业客户的首选。 而随着云原生以及数字化时代的到来，RocketMQ 也在快速的发生着变化，那么变化主要体现在哪些方面呢？ 首先，全面拥抱云原生。向下，消息系统自身实现云原生架构的演进，充分释放云基础设施的池化能力，全方位提高消息的核心技术指标。向上，消息产品形态持续演进，成为云原生应用架构的核心引擎。比如微服务、事件驱动、Serverless 等现代化应用架构。 其次，全面拥抱实时数据。企业的数字化转型从原来的业务数字化迈向了数字业务化。对业务数据的实时洞察、实时决策成为指导业务成功的关键要素。消息队列也将从在线业务架构的基础设施，延伸到实时数据架构的基础设施，从而实现事务分析一体化。 随着 5.0 的发布，RocketMQ也正式从消息服务升级到云原生事件流处理平台。 RocketMQ 5.0：云原生架构升级 首先来看 RocketMQ 自身的云原生架构演进。从下面的全景图可以看出，RocketMQ 从客户端到服务端都进行了全方位的改造，具体体现在以下几个方面： 轻量化。RocketMQ 4.0 诞生于 2012 年的淘宝电商，当时大部分的业务还跑在物理机上，单节点计算能力强，客户端节点数少且较为稳定，因此，富客户端的接入方式不仅更加高效，更可以提供诸如客户端侧负载均衡、消息缓存、故障转移等一系列企业级特性。但这种模式在云原生时代发生了改变，轻量客户端更容易被云原生技术栈所集成。因此，RocketMQ 5.0 客户端采用轻量 SDK 设计理念，将原来富客户端的逻辑下沉到服务端，满足现代化应用轻量化、Serverless 化以及 Mesh 化的趋势，更容易被集成；同时也正是因为轻量化，使得 SDK 多语言开发成本低了很多，快速覆盖当下主流的多语言版本。 弹性。存算分离架构让无状态计算节点可以快速伸缩，而分级存储以及冷热分离架构更是让消息存储具备更强的弹性能力。 高可用。基于全新的 Leaderless 架构，去 ZK 依赖的同时，可以做到副本数灵活选择，同步异步自动升降级，实现秒级故障转移；面向云的多可用区、多地域组建全局高可用能力。 基础设施云原生化。RocketMQ 整体架构走向 Kubernetes 化，拥抱 OpenTelemetry，依托于阿里云提供的 ARMS、Prometheus 以及Grafana 实现可观测能力的云原生化。 而 RocketMQ 5.0 本次的升级，除了在技术架构云原生化之外，在产品能力以及成本优化方面都有着重大的提升，我们来逐一分解。 轻量无状态消费模型 RocketMQ 4.0 采用按队列消费模型，消费者完全按照队列负载均衡，非常适合批量拉取快速消费，对单一消息状态不敏感的场景，比如流计算。然而在业务消息领域，尤其是金融场景以及事件驱动架构之下，每一条消息状态都是极为重要的。再加上不同业务类型的消息处理耗时也是各不相同，从毫秒级、到秒级甚至到分钟级，队列的负载不均衡或者短暂的 Block 都可能会引发消息的局部堆积，从而影响到最终用户的体验。因此，RocketMQ 5.0 全新推出按消息负载的轻量无状态消费模型，通过 PoP 机制巧妙地在队列模型之上构建了消息模型，业务只需要关心消息而无需关心队列，所有 API 都能够支持单条消息级别控制，如消息的消费、重试、删除等。而基于消息消费模型，客户端、连接和消费都是无状态的，可在任意 Proxy 节点上飘移，真正做到轻量化。 RocketMQ 5.0 提供按队列消费模型以及按消息消费模型，更好的满足事件与流的业务场景，正可谓鱼与熊掌兼得。 海量消息分级存储 RocketMQ 5.0 的另一个重大升级则是海量消息的分级存储。对消息队列了解的同学都知道，消息通常都是流动的短时间的存储，业内大部分消息产品对消息的保留时间都比较优先，3 天，7 天，最长 15 天不等。有限的存储空间使不仅限制了消息的保留时长，更在某些场景下可能会导致业务资损，比如在消息还没有被消费的时候，因为磁盘空间不足或者消息过期而被清除，这在金融等领域都是不可接受的。所以，RocketMQ 一直想要解决这样的问题，让存储变得更有弹性。 RocketMQ 5.0 基于ESSD、对象存储打造冷热分离以及分级存储架构，提供低成本的无限存储能力，确保消息不会因为本地磁盘空间不足而提前被清除，造成业务资损。我们提供消息存储的 Serverless，客户只需按实际存储使用量付费，而无需预购存储空间。 此外，流量削峰是消息队列极为常见的场景，而由此带来的海量消息堆积能力以及在堆积情况下的性能稳定性成为衡量产品性能的核心指标。RocketMQ 5.0 基于冷热数据分离架构进一步做到读写隔离，避免在堆积的场景下影响热数据的写入性能。分级存储的冷数据碎片规整能力更是提高了冷数据的读取性能，为用户提供一致的冷读 SLA。 售卖系列全线升级，最高降本 50% 从前面的介绍我们已经了解到，RocketMQ 5.0 在技术架构以及产品能力上都有着明显提升。 而 5.0 推出全新的售卖形态与计费策略相比 4.0 更简单、更灵活也更为普惠。实例的综合成本最高可降低 50%。接入门槛最低可降至 390 元/月，远低于自建成本。消息存储支持 Serverless 弹性能力，按需付费可大幅降低闲置成本。结合冷热分离的多级存储能力，相比开源自建可降低 67%，大幅降低消息队列的使用成本。 EventBridge：云上事件枢纽 事件驱动是一个起源很早的概念，早在几十年前，无论是操作系统内核的设计、还是客户端编程框架都大量采用了事件驱动技术。RocketMQ PushConsumer 提供的 API 其实就是一种事件驱动的编程范式，但在微服务应用架构当中，集成与通信才是刚需，事件驱动的价值并没有那么明显的体现。 而随着云原生时代的到来，计算力的构成越来越多样化。作为云原生的代表技术，Serverless 架构范式也是事件驱动的。无论是阿里云的函数计算、还是 AWS 的 Lambda，它们的主要触发源都是各种形态的事件，云产品事件，如 OSS 文件上传触发用户基于函数进行文件加工处理；用户业务事件，如 RocketMQ 触发函数运行消费逻辑处理等等。 以事件驱动为核心理念，阿里云推出了 EventBridge 产品，其使命就是打造云上的事件枢纽。通过EventBridge 可以兑现四大业务价值： 统一事件枢纽。阿里云从 IaaS、PaaS到第三方的 SaaS，每天都有数以亿计的事件产生，但却没有一种简单和统一的方式来触达这些事件；这些事件散落在各个地方形成『事件孤岛』，很难挖掘出有用的业务价值。只有充分发挥数据的规模效应，建立起数据之间的血缘关系，我们才能更好的发掘出数据的价值；所以 EventBridge 首要任务便是统一阿里云上的事件规范，拥抱CloudEvents 事件标准，打造阿里云统一的事件枢纽。 事件驱动引擎。当 EventBridge 连接了海量的事件源后，基于 RocketMQ 毫秒级的事件触发能力，必将加速企业 EDA/Serverless 的架构升级。 开放与集成。EventBridge 提供丰富的跨云、跨平台、跨产品、跨地域以及跨账号的连接能力，能够促进云产品、应用程序、SaaS 服务的相互集成。 低代码。EventBridge 借助Serverless 的应用中心，通过简单的规则匹配以及丰富的模板，即可实现事件的分发、过滤、转换等处理，进一步提升事件驱动的效率。 让消息无处不在，让事件无所不及 依托于 EventBridge、RocketMQ 以及函数计算 FC 的强强联合，目前 EventBridge 的事件生态已初具规模。 在云产品事件集成方面，目前已经集成 200+云产品事件源，3000 多种事件类型。 在数据集成与处理方面，EventBridge 与微服务应用、大数据、IoT 等场景深度集成。比如与消息生态的融合，阿里云 6 款消息队列产品通过 EventBridge 实现消息数据的互联互通，并且通过 EventBridge 的全球事件网络赋予消息全球消息路由的能力，同时也可以通过EventBridge 提供的丰富的模板，实现 ETL 数据处理能力。 在 SaaS 应用集成方面，包括钉钉、聚石塔以及云上 50 多个 SaaS 服务都可以通过 EventBridgehook 方式连接到 EventBridge。 在事件触发方面，EventBridge 目前已经触达 10 多个事件目标，海量的事件源都可以通过 EventBridge 触发包括 FC/SAE 等在内的 10 多款事件目标云产品。除此之外，目前 EventBridge 已经对接了阿里云全量的云产品 API，任何一个事件都可以通过云产品 API 的方式进行触达。 未来还有会更多的事件源以及事件目标接入到 EventBridge 上来。 RocketMQ Streams：轻量级计算的新选择 正如开篇所提到的，基于云原生架构的全面升级，RocketMQ 5.0 也将从在线业务架构的基础设施，延伸到实时数据架构的基础设施，实现事务分析一体化。将 RocketMQ Streams 打造成为轻量级计算的新选择。 业内最常见如 Flink、Spark 等大数据框架大多采用中心化的 MasterSlave 架构，依赖和部署比较重，每个任务的执行都需要很大的开销，有较高的使用成本。而与之不同的是，RocketMQ Streams 着重打造轻资源，高性能的轻量计算引擎，无额外依赖，最低1core，1g 即可完成部署，适用于大数据量、高过滤、轻窗口计算的场景，在资源敏感型场景，如消息队列流计算、安全风控，边缘计算等，RocketMQ Streams 具有有很大优势。阿里云消息团队与安全团队合作，通过对过滤场景做大量优化，性能提升 35 倍，资源节省 50%80%。 目前，RocketMQ Streams 已经在开源社区发布，未来计划在 2023 年 4 月在阿里云完成商业化。 RocketMQ 这十年，我们一同向前 RocketMQ历经十余年的打磨，已经取得了众多成果。全球拥有 700+的贡献者，1.8w Star 数，超过 80%的主流云厂商提供了 RocketMQ 的商业托管服务，Apache RocketMQ 社区始终保持着极高的活跃度，因此，也荣获了科创中国“开源创新榜”，中日韩开源软件优秀技术奖等十多个国内外开源奖项。 而阿里云作为 RocketMQ 的起源和核心贡献者，不仅 100%覆盖全集团业务，承载千万级并发万亿级消息洪峰。十多年以来更是累计服务 10w+万企业客户，覆盖互联网、零售、汽车等 20 多个行业，超过 75%的头部企业选择使用 RocketMQ。期望阿里云的消息队列 RocketMQ 可以成为广大企业客户的心之所选。也诚邀更广大的开发者来积极参与RocketMQ 的开源社区建设，一起将 RocketMQ 打造为消息领域的引领者。

本文目录 背景知识 大数据时代的构架演进 RocketMQ Connector&Stream Apache Hudi 构建Lakehouse实操 本文标题包含三个关键词：Lakehouse、RocketMQ、Hudi。我们先从整体Lakehouse架构入手，随后逐步分析架构产生的原因、架构组件特点以及构建Lakehouse架构的实操部分。 背景知识 1、Lakehouse架构 Lakehouse最初由Databrick提出，并对Lakehouse架构特征有如下要求： （1）事务支持 企业内部许多数据管道通常会并发读写数据。对ACID事务的支持确保了多方并发读写数据时的一致性问题； （2）Schema enforcement and governance Lakehouse应该有一种方式可以支持模式执行和演进、支持DW schema的范式（如星星或雪花模型），能够对数据完整性进行推理，并且具有健壮的治理和审计机制； （3）开放性 使用的存储格式是开放式和标准化的（如parquet），并且为各类工具和引擎，包括机器学习和Python/R库，提供API，以便它们可以直接有效地访问数据； （4）BI支持 Lakehouse可以直接在源数据上使用BI工具。这样可以提高数据新鲜度、减少延迟，并且降低了在数据池和数据仓库中操作两个数据副本的成本； （5）存储与计算分离 在实践中，这意味着存储和计算使用单独的集群，因此这些系统能够扩展到支持更大的用户并发和数据量。一些现代数仓也具有此属性； （6）支持从非结构化数据到结构化数据的多种数据类型 Lakehouse可用于存储、优化、分析和访问许多数据应用所需的包括image、video、audio、text以及半结构化数据； （7）支持各种工作负载 包括数据科学、机器学习以及SQL和分析。可能需要多种工具来支持这些工作负载，但它们底层都依赖同一数据存储库； （8）端到端流 实时报表是许多企业中的标准应用。对流的支持消除了需要构建单独系统来专门用于服务实时数据应用的需求。 从上述对Lakehouse架构的特点描述我们可以看出，针对单一功能，我们可以利用某些开源产品组合构建出一套解决方案。但对于全部功能的支持，目前好像没有一个通用的解决方案。接下来，我们先了解大数据时代主流的数据处理架构是怎样的。 大数据时代的架构演进 1、大数据时代的开源产品 大数据时代的开源产品种类繁多，消息领域的RocketMQ、Kafka；计算领域的flink、spark、storm；存储领域的HDFS、Hbase、Redis、ElasticSearch、Hudi、DeltaLake等等。 为什么会产生这么多开源产品呢？首先在大数据时代数据量越来越大，而且每个业务的需求也各不相同，因此就产生出各种类型的产品供架构师选择，用于支持各类场景。然而众多的品类产品也给架构师们带来一些困扰，比如选型困难、试错成本高、学习成本高、架构复杂等等。 2、当前主流的多层架构 大数据领域的处理处理场景包含数据分析、BI、科学计算、机器学习、指标监控等场景，针对不同场景，业务方会根据业务特点选择不同的计算引擎和存储引擎；例如交易指标可以采用binlog + CDC+ RocketMQ + Flink + Hbase + ELK组合，用于BI和Metric可视化。 （1）多层架构的优点：支持广泛的业务场景； （2）多层架构的缺点： 处理链路长，延迟高； 数据副本多，成本翻倍； 学习成本高； 造成多层架构缺点主要原因是存储链路和计算链路太长。 我们真的需要如此多的解决方案来支持广泛的业务场景吗？Lakehouse架构是否可以统一解决方案？ 多层架构的存储层是否可以合并？Hudi产品是否能够支持多种存储需求？ 多层架构的计算层是否可以合并？RocketMQ stream是否能够融合消息层和计算层？ 当前主流的多层架构 3、Lakehouse架构产生 Lakehouse架构是多层架构的升级版本，将存储层复杂度继续降低到一层。再进一步压缩计算层，将消息层和计算层融合，RocketMQ stream充当计算的角色。我们得到如下图所示的新架构。新架构中，消息出入口通过RocketMQ connector实现，消息计算层由RocketMQ stream实现，在RocketMQ内部完成消息计算中间态的流转；计算结果通过RocketMQHudiconnector收口落库Hudi，Hudi支持多种索引，并提供统一的API输出给不同产品。 Lakehouse架构 下面我们分析下该架构的特点。 （1）Lakehouse架构的优点： 链路更短，更适合实时场景，数据新鲜感高； 成本可控，降低了存储成本； 学习成本低，对程序员友好； 运维复杂度大幅降低； （2）Lakehouse架构的缺点 对消息产品和数据湖产品的稳定性、易用性等要求高，同时消息产品需要支持计算场景，数据湖产品需要提供强大的索引功能。 （3）选择 在Lakehouse架构中我们选择消息产品RocketMQ和数据湖产品Hudi。 同时，可以利用RocketMQ stream在RocketMQ集群上将计算层放在其中集成，这样就将计算层降低到一层，能够满足绝大部分中小型大数据处理场景。 接下来我们逐步分析RocketMQ和Hudi两款产品的特点。 RocketMQ Connector & Stream RocketMQ 发展历程图 RocketMQ从2017年开始进入Apache孵化，2018年RocketMQ 4.0发布完成云原生化，2021年RocketMQ 5.0发布全面融合消息、事件、流。 1、业务消息领域首选 RocketMQ作为一款“让人睡得着觉的消息产品”成为业务消息领域的首选，这主要源于产品的以下特点： （1）金融级高可靠 经历了阿里巴巴双十一的洪峰检验； （2）极简架构 如下图所示， RocketMQ的架构主要包含两部分包括：源数据集群NameServer Cluster和计算存储集群Broker Cluster。 RocketMQ 构架图 NameServer节点无状态，可以非常简单的进行横向扩容。Broker节点采用主备方式保证数据高可靠性，支持一主多备的场景，配置灵活。 搭建方式：只需要简单的代码就可以搭建RocketMQ集群： Jar： nohup sh bin/mqnamesrv & nohup sh bin/mqbroker n localhost:9876 & On K8S： kubectl apply f example/rocketmq_cluster.yaml （3）极低运维成本 RocketMQ的运维成本很低，提供了很好的CLI工具MQAdmin，MQAdmin提供了丰富的命令支持，覆盖集群健康状态检查、集群进出流量管控等多个方面。例如，mqadmin clusterList一条命令可以获取到当前集群全部节点状态（生产消费流量、延迟、排队长度、磁盘水位等）；mqadmin updateBrokerConfig命令可以实时设置broker节点或topic的可读可写状态，从而可以动态摘除临时不可用节点，达到生产消费的流量迁移效果。 （4）丰富的消息类型 RocketMQ支持的消息类型包括：普通消息、事务消息、延迟消息、定时消息、顺序消息等。能够轻松支持大数据场景和业务场景。 （5）高吞吐、低延迟 压测场景主备同步复制模式，每台Broker节点都可以将磁盘利用率打满，同时可以将p99延迟控制在毫秒级别。 2、RocketMQ 5.0概况 RocketMQ 5.0是生于云、长于云的云原生消息、事件、流超融合平台，它具有以下特点： （1）轻量级SDK 全面支持云原生通信标准 gRPC 协议； 无状态 Pop 消费模式，多语言友好，易集成； （2）极简架构 无外部依赖，降低运维负担； 节点间松散耦合，任意服务节点可随时迁移； （3）可分可合的存储计算分离 Broker 升级为真正的无状态服务节点，无 binding； Broker 和 Store节点分离部署、独立扩缩； 多协议标准支持，无厂商锁定； 可分可合，适应多种业务场景，降低运维负担； 如下图所示，计算集群（Broker）主要包括抽象模型和相对应的协议适配，以及消费能力和治理能力。存储集群（Store）主要分为消息存储CommitLog（多类型消息存储、多模态存储）和索引存储Index（多元索引）两部分，如果可以充分发挥云上存储的能力，将CommitLog和Index配置在云端的文件系统就可以天然的实现存储和计算分离。 （4）多模存储支持 满足不同基础场景下的高可用诉求； 充分利用云上基础设施，降低成本； （5）云原生基础设施： 可观测性能力云原生化，OpenTelemetry 标准化； Kubernetes 一键式部署扩容交付。 RocketMQ 5.02021年度大事件及未来规划 3、RocketMQConnector a、传统数据流 （1）传统数据流的弊端 生产者消费者代码需要自己实现，成本高； 数据同步的任务没有统一管理； 重复开发，代码质量参差不齐； （2）解决方案：RocketMQ Connector 合作共建，复用数据同步任务代码； 统一的管理调度，提高资源利用率； b、RocketMQ Connector数据同步流程 相比传统数据流，RocketMQ connector数据流的不同在于将 source 和 sink 进行统一管理，同时它开放源码，社区也很活跃。 4、RocketMQ Connector架构 如上图所示，RocketMQ Connector架构主要包含Runtime和Worker两部分，另外还有生态Source&Sink。 （1）标准：OpenMessaging （2）生态：支持ActiveMQ、Cassandra、ES、JDBC、JMS、MongoDB、Kafka、RabbitMQ、Mysql、Flume、Hbase、Redis等大数据领域的大部分产品； （3）组件：Manager统一管理调度，如果有多个任务可以将所有任务统一进行负载均衡，均匀的分配到不同Worker上，同时Worker可以进行横向扩容。 5、RocketMQ Stream RocketMQ Stream是一款将计算层压缩到一层的产品。它支持一些常见的算子如window、join、维表，兼容Flink SQL、UDF/UDAF/UDTF。 Apache Hudi Hudi 是一个流式数据湖平台，支持对海量数据快速更新。内置表格式，支持事务的存储层、一系列表服务、数据服务（开箱即用的摄取工具）以及完善的运维监控工具。Hudi 可以将存储卸载到阿里云上的 OSS、AWS 的S3这些存储上。 Hudi的特性包括： 事务性写入，MVCC/OCC并发控制； 对记录级别的更新、删除的原生支持； 面向查询优化：小文件自动管理，针对增量拉取优化的设计，自动压缩、聚类以优化文件布局； Apache Hudi是一套完整的数据湖平台。它的特点有： 各模块紧密集成，自我管理； 使用 Spark、Flink、Java 写入； 使用 Spark、Flink、Hive、Presto、Trino、Impala、 AWS Athena/Redshift等进行查询； 进行数据操作的开箱即用工具/服务。 Apache Hudi主要针对以下三类场景进行优化： 1、流式处理栈 (1) 增量处理； (2) 快速、高效； (3) 面向行； (4) 未优化扫描； 2、批处理栈 (1) 批量处理； (2) 低效； (3) 扫描、列存格式； 3、增量处理栈 (1) 增量处理； (2) 快速、高效； (3) 扫描、列存格式。 构建 Lakehouse 实操 该部分只介绍主流程和实操配置项，本机搭建的实操细节可以参考附录部分。 1、准备工作 RocketMQ version：4.9.0 rocketmqconnecthudi version：0.0.1SNAPSHOT Hudi version：0.8.0 2、构建RocketMQHudiconnector (1) 下载： _  git clone _ (2) 配置： /data/lakehouse/rocketmqexternals/rocketmqconnect/rocketmqconnectruntime/target/distribution/conf/connect.conf 中connectorplugin 路径 (3) 编译： cd rocketmqexternals/rocketmqconnecthudi mvn clean install DskipTest U rocketmqconnecthudi0.0.1SNAPSHOTjarwithdependencies.jar就是我们需要使用的rocketmqhudiconnector 3、运行 (1) 启动或使用现有的RocketMQ集群，并初始化元数据Topic： connectorclustertopic （集群信息） connectorconfigtopic （配置信息） connectoroffsettopic （sink消费进度） connectorpositiontopic （source数据处理进度 并且为了保证消息有序，每个topic可以只建一个queue） (2) 启动RocketMQ connector运行时 cd /data/lakehouse/rocketmqexternals/rocketmqconnect/rocketmqconnectruntime sh ./run_worker.sh Worker可以启动多个 (3) 配置并启动RocketMQhudiconnector任务 请求RocketMQ connector runtime创建任务 curl http://{runtimeip}:{runtimeport}/connectors/{rocketmqhudisinkconnectorname} ?config='{"connectorclass":"org.apache.rocketmq.connect.hudi.connector.HudiSinkConnector","topicNames":"topicc","tablePath":"file:///tmp/hudi_connector_test","tableName":"hudi_connector_test_table","insertShuffleParallelism":"2","upsertShuffleParallelism":"2","deleteParallelism":"2","sourcerecordconverter":"org.apache.rocketmq.connect.runtime.converter.RocketMQConverter","sourcerocketmq":"127.0.0.1:9876","srccluster":"DefaultCluster","refreshinterval":"10000","schemaPath":"/data/lakehouse/config/user.avsc"\}’ 启动成功会打印如下日志： 20210906 16:23:14 INFO pool2thread1 Open HoodieJavaWriteClient successfully (4) 此时向source topic生产的数据会自动写入到1Hudi对应的table中，可以通过Hudi的api进行查询。 4、配置解析 (1) RocketMQ connector需要配置RocketMQ集群信息和connector插件位置，包含：connect工作节点id标识workerid、connect服务命令接收端口httpPort、rocketmq集群namesrvAddr、connect本地配置储存目录storePathRootDir、connector插件目录pluginPaths 。 RocketMQ connector配置表 (2) Hudi任务需要配置Hudi表路径tablePath和表名称tableName，以及Hudi使用的Schema文件。 Hudi任务配置表 _点击__即可查看Lakehouse构建实操视频_ 附录：在本地Mac系统构建Lakehouse demo 涉及到的组件：rocketmq、rocketmqconnectorruntime、rocketmqconnecthudi、hudi、hdfs、avro、sparkshell0、启动hdfs 下载hadoop包 cd /Users/osgoo/Documents/hadoop2.10.1 vi coresite.xml fs.defaultFS hdfs://localhost:9000 vi hdfssite.xml dfs.replication 1 ./bin/hdfs namenode format ./sbin/startdfs.sh jps 看下namenode,datanode lsof i:9000 ./bin/hdfs dfs mkdir p /Users/osgoo/Downloads 1、启动rocketmq集群，创建rocketmqconnector内置topic QickStart：https://rocketmq.apache.org/docs/quickstart/ sh mqadmin updatetopic t connectorclustertopic n localhost:9876 c DefaultCluster sh mqadmin updatetopic t connectorconfigtopic n localhost:9876 c DefaultCluster sh mqadmin updatetopic t connectoroffsettopic n localhost:9876 c DefaultCluster sh mqadmin updatetopic t connectorpositiontopic n localhost:9876 c DefaultCluster 2、创建数据入湖的源端topic，testhudi1 sh mqadmin updatetopic t testhudi1 n localhost:9876 c DefaultCluster 3、编译rocketmqconnecthudi0.0.1SNAPSHOTjarwithdependencies.jar cd rocketmqconnecthudi mvn clean install DskipTest U 4、启动rocketmqconnector runtime 配置connect.conf workerId=DEFAULT_WORKER_1 storePathRootDir=/Users/osgoo/Downloads/storeRoot Http port for user to access REST API httpPort=8082 Rocketmq namesrvAddr namesrvAddr=localhost:9876 Source or sink connector jar file dir,The default value is rocketmqconnectsample pluginPaths=/Users/osgoo/Downloads/connectorplugins 拷贝 rocketmqhudiconnector.jar 到 pluginPaths=/Users/osgoo/Downloads/connectorplugins sh run_worker.sh 5、配置入湖config curl http://localhost:8082/connectors/rocketmqconnecthudi?config='\{"connectorclass":"org.apache.rocketmq.connect.hudi.connector.HudiSinkConnector","topicNames":"testhudi1","tablePath":"hdfs://localhost:9000/Users/osgoo/Documents/basepath7","tableName":"t7","insertShuffleParallelism":"2","upsertShuffleParallelism":"2","deleteParallelism":"2","sourcerecordconverter":"org.apache.rocketmq.connect.runtime.converter.RocketMQConverter","sourcerocketmq":"127.0.0.1:9876","sourcecluster":"DefaultCluster","refreshinterval":"10000","schemaPath":"/Users/osgoo/Downloads/user.avsc"\}' 6、发送消息到testhudi1 7、 利用spark读取 cd /Users/osgoo/Downloads/spark3.1.2binhadoop3.2/bin ./sparkshell \ packages org.apache.hudi:hudispark3bundle_2.12:0.9.0,org.apache.spark:sparkavro_2.12:3.0.1 \ conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' import org.apache.hudi.QuickstartUtils._ import scala.collection.JavaConversions._ import org.apache.spark.sql.SaveMode._ import org.apache.hudi.DataSourceReadOptions._ import org.apache.hudi.DataSourceWriteOptions._ import org.apache.hudi.config.HoodieWriteConfig._ val tableName = "t7" val basePath = "hdfs://localhost:9000/Users/osgoo/Documents/basepath7" val tripsSnapshotDF = spark. read. format("hudi"). load(basePath + "/") tripsSnapshotDF.createOrReplaceTempView("hudi_trips_snapshot") spark.sql("select from hudi_trips_snapshot").show()

在 Spring 生态中玩转 RocketMQ 系列教程现已登陆知行动手实验室，立即体验！ 移动端同学，需要在PC端登录 start.aliyun.com 进行体验。 RocketMQ 作为业务消息的首选，在消息和流处理领域被广泛应用。而微服务生态 Spring 框架也是业务开发中最受欢迎的框架，两者的完美契合使得 RocketMQ 成为 Spring Messaging 实现中最受欢迎的消息实现。本文展示了 5 种在 Spring 生态中文玩转 RocketMQ 的方式，并描述了每个项目的特点和使用场景。 一、前言 上世纪 90 年代末，随着 Java EE（Enterprise Edition）的出现，特别是 Enterprise Java Beans 的使用需要复杂的描述符配置和死板复杂的代码实现，增加了广大开发者的学习曲线和开发成本，由此基于简单的 XML 配置和普通 Java 对象（Plain Old Java Objects）的 Spring 技术应运而生，依赖注入（Dependency Injection），控制反转（Inversion of Control）和面向切面编程（AOP）的技术更加敏捷地解决了传统 Java 企业及版本的不足。随着 Spring 的持续演进，基于注解（Annotation）的配置逐渐取代了 XML 文件配置。除了依赖注入、控制翻转、AOP 这些技术，Spring 后续衍生出 AMQP、Transactional、Security、Batch、Data Access 等模块，涉及开发的各个领域。 2014 年 4 月 1 日，Spring Boot 1.0.0 正式发布。它基于“约定大于配置”（Convention over configuration)这一理念来快速地开发，测试，运行和部署 Spring 应用，并能通过简单地与各种启动器(如springbootwebstarter)结合，让应用直接以命令行的方式运行，不需再部署到独立容器中。Spring Boot 的出现可以说是 Spring 框架的第二春，它不但简化了开发的流程，目前更是事实标准。下面这幅图可以看出相同功能的 Spring 和 Spring Boot 的代码实现对比。 Apache RocketMQ 是一款是业界知名的分布式消息和流处理中间件，它主要功能是消息分发、异步解耦、削峰填谷等。RocketMQ 是一款金融级消息及流数据平台，RocketMQ 在交易、支付链路上用的很多，主要是对消息链路质量要求非常高的场景，能够支持万亿级消息洪峰。RocketMQ 在业务消息中被广泛应用，并衍生出顺序消息、事务消息、延迟消息等匹配各类业务场景的特殊消息。 本文的主角就是 Spring 和 RocketMQ，那几乎每个 Java 程序员都会使用 Spring 框架与支持丰富业务场景的 RocketMQ 会碰撞出怎么样的火花？ 二、RocketMQ 与 Spring 的碰撞 在介绍 RocketMQ 与 Spring 故事之前，不得不提到 Spring 中的两个关于消息的框架，Spring Messaging 和 Spring Cloud Stream。它们都能够与 Spring Boot 整合并提供了一些参考的实现。和所有的实现框架一样，消息框架的目的是实现轻量级的消息驱动的微服务，可以有效地简化开发人员对消息中间件的使用复杂度，让系统开发人员可以有更多的精力关注于核心业务逻辑的处理。 1. Spring Messaging Spring Messaging 是 Spring Framework 4 中添加的模块，是 Spring 与消息系统集成的一个扩展性的支持。它实现了从基于 JmsTemplate 的简单的使用 JMS 接口到异步接收消息的一整套完整的基础架构，Spring AMQP 提供了该协议所要求的类似的功能集。在与 Spring Boot 的集成后，它拥有了自动配置能力，能够在测试和运行时与相应的消息传递系统进行集成。单纯对于客户端而言，Spring Messaging 提供了一套抽象的 API 或者说是约定的标准，对消息发送端和消息接收端的模式进行规定，比如消息 Messaging 对应的模型就包括一个消息体 Payload 和消息头 Header。不同的消息中间件提供商可以在这个模式下提供自己的 Spring 实现：在消息发送端需要实现的是一个 XXXTemplate 形式的 Java Bean，结合 Spring Boot 的自动化配置选项提供多个不同的发送消息方法；在消息的消费端是一个 XXXMessageListener 接口（实现方式通常会使用一个注解来声明一个消息驱动的 POJO），提供回调方法来监听和消费消息，这个接口同样可以使用 Spring Boot 的自动化选项和一些定制化的属性。 在 Apache RocketMQ 生态中，RocketMQSpringBootStarter（下文简称 RocketMQSpring）就是一个支持 Spring Messaging API 标准的项目。该项目把 RocketMQ 的客户端使用 Spring Boot 的方式进行了封装，可以让用户通过简单的 annotation 和标准的 Spring Messaging API 编写代码来进行消息的发送和消费，也支持扩展出 RocketMQ 原生 API 来支持更加丰富的消息类型。在 RocketMQSpring 毕业初期，RocketMQ 社区同学请 Spring 社区的同学对 RocketMQSpring 代码进行 review，引出一段罗美琪（RocketMQ）和春波特（Spring Boot）故事的佳话[1]，著名 Spring 布道师 Josh Long 向国外同学介绍如何使用 RocketMQSpring 收发消息[2]。RocketMQSpring 也在短短两年时间超越 SpringKafka 和 SpringAMQP（注:两者均由 Spring 社区维护），成为 Spring Messaging 生态中最活跃的消息项目。 2. Spring Cloud Stream Spring Cloud Stream 结合了 Spring Integration 的注解和功能，它的应用模型如下： Spring Cloud Stream 框架中提供一个独立的应用内核，它通过输入(@Input)和输出(@Output)通道与外部世界进行通信，消息源端(Source)通过输入通道发送消息，消费目标端(Sink)通过监听输出通道来获取消费的消息。这些通道通过专用的 Binder 实现与外部代理连接。开发人员的代码只需要针对应用内核提供的固定的接口和注解方式进行编程，而不需要关心运行时具体的 Binder 绑定的消息中间件。 在运行时，Spring Cloud Stream 能够自动探测并使用在 classpath 下找到的 Binder。这样开发人员可以轻松地在相同的代码中使用不同类型的中间件：仅仅需要在构建时包含进不同的 Binder。在更加复杂的使用场景中，也可以在应用中打包多个 Binder 并让它自己选择 Binder，甚至在运行时为不同的通道使用不同的 Binder。 Binder 抽象使得 Spring Cloud Stream 应用可以灵活的连接到中间件，加之 Spring Cloud Stream 使用利用了 Spring Boot 的灵活配置配置能力，这样的配置可以通过外部配置的属性和 Spring Boot 支持的任何形式来提供（包括应用启动参数、环境变量和 application.yml 或者 application.properties 文件），部署人员可以在运行时动态选择通道连接 destination（例如，RocketMQ 的 topic 或者 RabbitMQ 的 exchange）。 Spring Cloud Stream 屏蔽了底层消息中间件的实现细节，希望以统一的一套 API 来进行消息的发送/消费，底层消息中间件的实现细节由各消息中间件的 Binder 完成。Spring 官方实现了 Rabbit binder 和 Kafka Binder。Spring Cloud Alibaba 实现了 RocketMQ Binder[3]，其主要实现原理是把发送消息最终代理给了 RocketMQSpring 的 RocketMQTemplate，在消费端则内部会启动 RocketMQSpring Consumer Container 来接收消息。以此为基础，Spring Cloud Alibaba 还实现了 Spring Cloud Bus RocketMQ， 用户可以使用 RocketMQ 作为 Spring Cloud 体系内的消息总线，来连接分布式系统的所有节点。通过 Spring Cloud Stream RocketMQ Binder，RocketMQ 可以与 Spring Cloud 生态更好的结合。比如与 Spring Cloud Data Flow、Spring Cloud Funtion 结合，让 RocketMQ 可以在 Spring 流计算生态、Serverless(FaaS) 项目中被使用。 如今 Spring Cloud Stream RocketMQ Binder 和 Spring Cloud Bus RocketMQ 做为 Spring Cloud Alibaba 的实现已登陆 Spring 的官网[4]，Spring Cloud Alibaba 也成为 Spring Cloud 最活跃的实现。 三、如何在 Spring 生态中选择 RocketMQ 实现？ 通过介绍 Spring 中的消息框架，介绍了以 RocketMQ 为基础与 Spring 消息框架结合的几个项目，主要是 RocketMQSpring、Spring Cloud Stream RocketMQ Binder、Spring Cloud Bus RocketMQ、Spring Data Flow 和 Spring Cloud Function。它们之间的关系可以如下图表示。 如何在实际业务开发中选择相应项目进行使用？下面分别列出每个项目的特点和使用场景。 1. RocketMQSpring 特点： 作为起步依赖，简单引入一个包就能在 Spring 生态用到 RocketMQ 客户端的所有功能。 利用了大量自动配置和注解简化了编程模型，并且支持 Spring Messaging API。 与 RocketMQ 原生 Java SDK 的功能完全对齐。 使用场景： 适合在 Spring Boot 中使用 RocketMQ 的用户，希望能用到 RocketMQ 原生 java 客户端的所有功能，并通过 Spring 注解和自动配置简化编程模型。 2. Spring Cloud Stream RocketMQ Binder 特点： 屏蔽底层 MQ 实现细节，上层 Spring Cloud Stream 的 API 是统一的。如果想从 Kafka 切到 RocketMQ，直接改个配置即可。 与 Spring Cloud 生态整合更加方便。比如 Spring Cloud Data Flow，这上面的流计算都是基于 Spring Cloud Stream；Spring Cloud Bus 消息总线内部也是用的 Spring Cloud Stream。 Spring Cloud Stream 提供的注解，编程体验都是非常棒。 使用场景： 在代码层面能完全屏蔽底层消息中间件的用户，并且希望能项目能更好的接入 Spring Cloud 生态（Spring Cloud Data Flow、Spring Cloud Funtcion 等）。 3. Spring Cloud Bus RocketMQ 特点： 将 RocketMQ 作为事件的“传输器”，通过发送事件（消息）到消息队列上，从而广播到订阅该事件（消息）的所有节点上，完成事件的分发和通知。 使用场景： 在 Spring 生态中希望用 RocketMQ 做消息总线的用户，可以用在应用间事件的通信，配置中心客户端刷新等场景。 4. Spring Cloud Data Flow 特点： 以 Source/Processor/Sink 组件进行流式任务处理。RocketMQ 作为流处理过程中的中间存储组件。 使用场景： 流处理，大数据处理场景。 5. Spring Cloud Function 特点： 消息的消费/生产/处理都是一次函数调用，融合 Java 生态的 Function 模型。 使用场景： Serverless 场景。 本文整体介绍了在 Spring 生态中接入 RockeMQ 的 5 种方法，让各位开发者对几种经典场景有宏观的了解。后续会有专栏详细介绍上述各个项目的具体使用方法和应用场景，真正地在 Spring 生态中玩转 RocketMQ！ 在 Spring 生态中玩转 RocketMQ 系列教程现已登陆知行动手实验室，立即体验！ 移动端同学，需要在PC端登录 start.aliyun.com 进行体验。 相关链接：