引言：AI 时代的数据处理变革 人工智能技术的发展经历了从感知智能到生成智能，再到智能体和具身智能的跨越式演进。这一过程不仅体现在算法模型的不断突破，更深刻地反映在对数据处理能力要求的根本性变化。根据麦肯锡的调研数据显示，2022 年，全球有 50% 的公司部署了 AI 技术，投资超过总预算的 4%。生成式 AI（GenAI）的崛起进一步推动了企业转型，其在流程优化、个性化服务等方面的应用已经超越了传统 AI 的范畴。 在这一技术变革的浪潮中，数据处理能力的重要性愈发凸显。传统的数据处理架构主要围绕结构化数据的批量处理而设计，采用的是相对静态的 ETL 模式。然而，AI 时代的数据处理需求呈现出截然不同的特征：数据源更加多样化，包括文本、图像、音频、视频等多模态数据；处理要求更加实时化，需要支持流式数据的即时处理和响应；应用场景更加智能化，需要结合大语言模型的推理能力进行数据的理解、转换和增强。 本文将从 AI 时代数据处理的挑战与机遇出发，深入分析事件驱动架构在 AI 数据处理中的技术优势，详细阐述 EventBridge for AI ETL 的实践案例，展示其在不同应用场景中的价值。我们希望能够为企业在 AI 转型过程中的数据基础设施建设，提供有价值的技术指导和实践参考，推动 AI 技术在更广泛领域的落地应用。 一、AI 时代数据处理的挑战与机遇 1. GenAI 的演进路径分析 生成式 AI 的发展经历了从简单到复杂、从单一到多元的演进过程，每个阶段都对数据处理能力提出了不同的要求。深入理解这一演进路径，对于把握 AI 数据处理的发展趋势具有重要意义。 最初起点：简单模型 API 调用阶段 在生成式 AI 发展的初期阶段，应用架构相对简单直接。用户通过 Query 向大语言模型发送请求，模型基于预训练的知识生成 Response 并返回给用户。这种架构虽然现在看来可能过于"简陋"，但却是许多初期现象级 AI 应用产品的起点，如文本总结、AI 算命、AI 情感分析等应用都采用了这种直白的架构模式。 在这个阶段，数据处理的需求相对简单，主要集中在 Prompt 的优化上。开发者需要通过精心设计的提示词来引导模型生成期望的输出，数据处理更多体现在输入文本的预处理和输出结果的后处理上。然而，这种简单的架构很快就暴露出明显的局限性：模型的知识截止时间限制了其对最新信息的获取能力，缺乏领域专业知识导致在特定场景下的表现不佳，无法处理个性化和上下文相关的复杂查询。 增强上下文：RAG 技术的兴起 为了解决简单模型 API 调用的局限性，RAG（RetrievalAugmented Generation）技术应运而生。RAG 的核心思想是在模型生成回答之前，先从外部知识库通过之前用于搜广推的向量检索技术方案，检索相关信息，然后将检索到的信息作为上下文提供给模型，从而增强模型的生成能力。 RAG 技术的引入标志着 AI 数据处理进入了一个新的阶段，数据处理需求显著增加，主要体现在两个方面：首先是问题域特有信息的处理，例如在分析用户在某个平台的购买喜好时，需要实时获取和处理用户在该平台的购买数据；其次是时效信息的处理，如股票信息、实时新闻等需要不断更新的动态数据。 RAG 技术的实现需要构建一个完整的数据处理管道，包括数据收集、预处理、向量化、存储、检索和后处理等多个环节。这对数据处理系统的实时性、准确性和可扩展性提出了更高的要求。根据 Menlo Ventures 发布的市场调研报告，RAG 以 51% 的市场份额在企业市场中占据绝对优势，充分说明其在实际应用中的重要地位。 Agent 模式：智能体的规划与工具能力 随着 AI 技术的进一步发展，单纯的检索增强已经无法满足复杂应用场景的需求。Agent 模式的出现代表了 AI 应用架构的又一次重大演进。Agent 是在特定环境下具备 plan+tools 能力的智能体，其中"特定环境"限制了 Agent 的创建面向特定的场景和问题域，"plan" 说明 Agent 具有思考和规划能力，且能够根据反馈进行循环迭代，"tools" 则是指 Agent 具备与外部交互的能力。 Agent 模式对事件驱动和数据处理提出了更加复杂和多样化的需求。Agent 需要能够动态地选择和调用不同的工具来完成任务，这要求事件驱动系统具备高度的灵活性和可扩展性。同时，Agent 的规划和决策过程需要基于实时的环境信息和历史数据，这对数据的实时性和一致性提出了严格要求。 2. 数据种类的多样化挑战 AI 时代的数据处理面临着前所未有的数据种类多样化挑战。与传统的以结构化数据为主的处理模式不同，AI 应用需要处理包括文本、图像、音频、视频在内的多模态数据，每种数据类型都有其独特的处理要求和技术挑战。 结构化数据 结构化数据具有固定的格式和明确的字段定义，是传统数据处理系统最擅长处理的数据类型。在 AI 应用中，结构化数据主要来源于数据库、数据仓库、业务系统等，包括用户信息、交易记录、日志数据等。这类数据的处理相对成熟，主要挑战在于如何高效地进行数据清洗、转换和集成。 然而，即使是结构化数据的处理，在 AI 时代也面临新的挑战。首先是数据量的急剧增长，根据 IDC 的预测，全球数据量将从 2020 年的 64.2ZB 增长到 2025 年的 175ZB 。其次是数据源的多样化，企业需要整合来自不同系统、不同格式的结构化数据。最后是实时性要求的提高，AI 应用往往需要基于最新的数据进行推理和决策。 非结构化数据 非结构化数据在 AI 应用中占据越来越重要的地位。文本数据是较为通用的非结构化数据类型，包括文档、邮件、社交媒体内容、客服对话等。这类数据的处理需要运用自然语言处理技术，包括分词、实体识别、情感分析、语义理解等。 图像和视频数据的处理更加复杂，需要运用计算机视觉技术进行特征提取、目标检测、图像分类等。音频数据的处理则涉及语音识别、音频分类、声纹识别等技术。每种非结构化数据都需要专门的预处理、特征提取和向量化技术，这大大增加了数据处理系统的复杂性。 半结构化数据 半结构化数据是介于完全结构化的数据和完全无结构的数据之间的一种数据形式。它不符合关系数据库或其他数据表形式的严格结构，但包含标签或其他标记，用于分隔语义元素和执行记录和字段的层次结构。这使得它比非结构化数据更容易分析，也更具灵活性。 AI 系统，特别是机器学习和深度学习模型，需要大量的、多样化的数据进行训练和推理。半结构化数据凭借其灵活性和丰富的上下文信息，在 AI 的多个关键环节中扮演着核心角色。譬如标注信息存储，特征工程，A2A Message Events 等等。 多模态数据的融合处理 随着 AI 技术的发展，越来越多的应用需要同时处理多种模态的数据。例如，智能客服系统需要同时处理文本、语音和图像信息；智能推荐系统需要结合用户的行为数据、内容特征和社交关系等多维信息。多模态数据的融合处理不仅需要处理每种模态的数据，还需要建立不同模态之间的关联和映射关系。 这种融合处理的挑战在于如何保证不同模态数据的时间同步、语义一致和质量统一。同时，多模态数据的存储和检索也需要专门的技术支持，传统的关系型数据库往往无法满足这种需求，需要采用向量数据库、图数据库等新型存储技术。 3. 主流数据采集方式的演变 AI 时代的数据采集方式相比传统模式发生了显著变化。传统的数据采集主要依赖定期的批量抽取，而 AI 应用往往需要实时或近实时的数据流。这种变化对数据采集系统的架构和性能提出了新的要求。 + 实时数据流采集成为主流趋势。通过消息队列、流处理框架等技术，系统能够实时捕获和处理数据变化。Apache Kafka、Apache RocketMQ 等流处理平台在 AI 数据采集中发挥着越来越重要的作用。根据市场研究数据，流处理技术已经成为 2024 年数据集成的关键趋势。 + API 驱动的数据采集也变得越来越普遍。通过 RESTful API、GraphQL 等接口，系统能够按需获取外部数据源的信息。这种方式特别适合处理第三方服务的数据，如社交媒体数据、天气信息、金融数据等。 + 事件驱动的数据采集是另一个重要趋势。当特定事件发生时，系统自动触发数据采集和处理流程。这种方式能够大大提高数据处理的效率和实时性，特别适合处理用户行为数据、系统日志等事件型数据。 4. AI 时代的数据集成挑战 （图源：_https://x.com/RLanceMartin/status/1673380038274695169_） 上图是 Langchain 在 RAG 领域定义的数据集成。诚然它具有模块化、声明式设计，并为我们提供了大量实用程序和辅助功能，但是在工程化的复杂度依旧存在，我们依然会陷入针对 Data 领域的抽象和工程化实现。 所以，在 AI 与数据集成的实践过程中，我们总结出企业普遍面临三大核心痛点，这些痛点不仅影响了 AI 应用的开发效率，也制约了 AI 技术的规模化应用。 扩展难：数据源异构性挑战 随着企业数字化程度的提高，数据源的种类和数量呈爆炸式增长。企业需要整合来自 ERP、CRM、OA、电商平台、社交媒体、IoT 设备等各种系统的数据。这些数据源在数据格式、接口协议、更新频率、访问权限等方面存在巨大差异。 一个简单的数据集成项目在初期往往进展顺利，但随着需要接入的数据源增加，系统的复杂度呈指数级增长。每增加一个新的数据源，开发团队都需要了解其特定的数据格式和接口规范，开发相应的连接器和转换逻辑，并进行充分的测试和验证。这种线性增长的开发模式严重制约了 AI 项目的扩展能力。 更为严重的是，不同数据源之间往往存在数据格式不一致、字段命名不规范、数据质量参差不齐等问题。例如，同样是用户信息，不同系统可能使用不同的用户 ID 格式，时间字段可能采用不同的时区和格式，地址信息可能有不同的结构化程度。这些差异需要在数据集成过程中进行统一处理，进一步增加了系统的复杂性。 运维难：业务复杂性增长 AI 数据处理系统的运维复杂性远超传统的数据处理系统。首先，AI 应用对数据的实时性要求更高，任何数据延迟都可能影响模型的推理效果。其次，AI 数据处理涉及多个环节，包括数据采集、清洗、转换、向量化、存储、检索等，每个环节都可能出现问题。最后，AI 模型的迭代更新频繁，数据处理逻辑也需要相应调整。 在实际运维过程中，运维团队经常面临各种突发问题：数据源突然变更接口格式导致数据采集中断，数据质量问题导致模型推理结果异常，系统负载突增导致处理延迟，存储空间不足导致数据丢失等。这些问题往往需要跨团队协作解决，涉及数据工程师、算法工程师、运维工程师等多个角色。 传统的运维方式主要依赖人工监控和处理，这种方式在面对 AI 数据处理系统的复杂性时显得力不从心。企业迫切需要智能化的运维工具和自动化的故障处理机制，以降低运维成本和提高系统可靠性。 稳定性差：数据链路可靠性问题 数据链路的稳定性是 AI 应用能否成功上线生产环境的关键因素。在 AI 应用中，数据质量和处理链路的任何问题都可能导致模型推理结果的偏差甚至错误，进而影响业务决策和用户体验。 数据链路的稳定性问题主要体现在几个方面：数据丢失或重复，由于网络故障、系统异常等原因导致数据在传输过程中丢失或重复处理；数据延迟，由于处理能力不足、网络拥塞等原因导致数据处理延迟，影响 AI 应用的实时性；数据质量下降，由于数据源变更、处理逻辑错误等原因导致数据质量下降，影响模型的推理效果；系统故障，由于硬件故障、软件 bug 等原因导致整个数据处理链路中断。 这些稳定性问题的根本原因在于传统数据处理架构的紧耦合设计。在紧耦合架构中，任何一个组件的故障都可能影响整个系统的运行。同时，缺乏有效的监控和告警机制，问题往往在造成严重影响后才被发现。 为了解决这些问题，业界开始探索基于事件驱动架构的松耦合设计。通过将数据处理流程分解为独立的事件和处理单元，系统能够实现更好的容错能力和可扩展性。同时，通过引入智能监控和自动恢复机制，系统能够及时发现和处理各种异常情况，保证数据链路的稳定运行。 二、AI 数据处理的技术基石 事件驱动架构__ 1. 事件驱动架构的核心概念 事件驱动架构（EventDriven Architecture，EDA）作为一种现代软件架构模式，为解决 AI 时代数据处理的复杂性挑战提供了强有力的技术基础。 Event的本质：状态变化的数字化表达 在事件驱动架构中，Event（事件）是系统的核心概念。简单来说，事件就是状态的显著变化，是一切能够输入计算机中且能被处理的符号的数字化表达。这种定义看似简单，但却蕴含着深刻的技术内涵。 以一个典型的 4S 店售卖汽车的业务场景为例，我们可以清晰地看到事件的本质特征。当客户购买汽车并且其状态从"For Sale"变为"Sold"时，这构成了一个销售事件。成功交易后，从账户中扣除金额形成了一个支付事件。用户点击预订试驾后，将预约信息添加到指定用户的操作产生了一个预约事件。甚至用户资料和预约单本身也可以被视为事件的载体。 这种事件化的思维方式具有重要的技术优势。首先，事件提供了系统状态变化的完整记录，使得系统具备了天然的审计和回溯能力。其次，事件的异步特性使得系统组件之间能够实现松散耦合，提高了系统的可扩展性和容错能力。最后，事件的标准化格式使得不同系统之间的集成变得更加简单和可靠。 在 AI 数据处理场景中，事件的概念得到了进一步的扩展和深化。数据的产生、变更、处理、存储等各个环节都可以被抽象为事件。例如，当新的训练数据上传到系统时，产生数据接收事件；当数据经过清洗和转换后，产生数据处理完成事件；当向量化处理完成后，产生向量生成事件；当数据成功存储到向量数据库后，产生数据入库事件。这种事件化的处理方式使得整个 AI 数据处理流程变得清晰、可控和可监控。 2. EventBridge 架构深度解析 阿里云 EventBridge 作为事件驱动架构的具体实现，在技术架构设计上充分体现了 EDA 的核心理念，同时针对 AI 数据处理的特殊需求进行了深度优化。 源、过滤、转换、目标 EventBridge 的技术架构围绕四大核心能力构建：源（Source）、过滤（Filter）、转换（Transform）、目标（Sink）。这四大能力形成了完整的事件处理链路，为 AI 数据处理提供了全面的技术支撑。 源（Source）能力负责事件的接入和采集。EventBridge 支持多种类型的事件源，包括结构化数据源和非结构化数据源。结构化数据源涵盖了消息队列（Kafka、RocketMQ 等）、数据库（关系型数据库、数据仓库）、可观测性平台（SLS、Prometheus）、API 接口等。非结构化数据源则包括对象存储（CSV、PDF、TXT 等文件格式）以及各种自定义数据源。这种多样化的数据源支持使得 EventBridge 能够适应 AI 应用中复杂多变的数据接入需求。 过滤（Filter）能力提供了灵活的事件筛选机制。通过事件模式匹配，系统可以根据预定义的规则对事件进行筛选和路由。EventBridge 支持多种匹配模式，包括指定值匹配、前缀匹配、包含匹配、除外匹配、多模式匹配等。这种细粒度的过滤能力使得系统能够精确地控制事件的处理流程，避免不必要的计算资源消耗。 转换（Transform）能力是 EventBridge 在 AI 领域的核心创新。系统支持多种转换方式，包括自定义代码转换、自定义模型转换、自定义 API 转换等。特别值得注意的是，EventBridge 集成了百炼模型服务，能够调用大语言模型进行智能化的数据转换。这种 AI 驱动的转换能力使得系统能够处理传统 ETL 工具难以处理的复杂数据转换任务。 目标（Sink）能力负责处理后事件的输出和存储。EventBridge 支持多种目标类型，包括消息队列、数据库、数据仓库、可观测性平台、函数计算、API 接口、通知服务等。这种多样化的目标支持使得处理后的数据能够灵活地流向不同的下游系统，满足 AI 应用的多样化需求。 事件总线模型：N:M 的灵活路由 EventBridge 的事件总线模型采用了经典的 EDA（事件驱动）架构中的 N:M 模型，提供了多事件路由、事件匹配、事件转换等核心能力，帮助用户快速搭建事件驱动架构。 在事件总线模型中，多个事件源可以同时向事件总线发送事件，事件总线根据预定义的规则将事件路由到相应的目标服务。这种 N:M 的路由模式具有重要的技术优势。 + 首先，它实现了事件源和目标服务之间的完全解耦，事件源不需要知道有哪些目标服务在消费事件，目标服务也不需要知道事件来自哪个源。 + 其次，它支持动态的路由配置，可以在运行时添加或删除事件源和目标服务，而不影响系统的正常运行。 + 最后，它提供了强大的事件复制和广播能力，一个事件可以同时被多个目标服务处理，实现了数据的多路分发。 在 AI 数据处理场景中，事件总线模型的这些特性具有重要价值。例如，当新的训练数据到达时，可以同时触发数据预处理、质量检查、备份存储等多个处理流程。当模型推理完成时，可以同时更新缓存、记录日志、发送通知等。这种并行处理能力大大提高了 AI 数据处理的效率。 事件流模型：1:1 的高效传输 除了事件总线模型，EventBridge 还提供了事件流模型，采用标准的 Streaming（1:1）流式处理场景。事件流模型没有总线概念，适用于端到端的数据转储、数据同步及数据处理等场景，帮助用户轻松构建云上数据管道服务。 事件流模型的核心优势在于其高效的点对点传输能力。在这种模型中，事件从源直接流向目标，中间经过匹配和转换处理，但不需要经过复杂的路由逻辑。这种简化的处理流程使得事件流模型在处理大量数据时具有更高的性能和更低的延迟。 在 AI 数据处理中，事件流模型特别适合处理实时数据流。例如，将实时产生的用户行为数据直接流式处理并存储到向量数据库中，或者将传感器数据实时转换为模型输入格式。这种高效的流式处理能力为实时 AI 应用提供了重要的技术支撑。 3. 事件驱动架构在 AI 领域的应用价值 事件驱动架构在 AI 领域的应用价值不仅体现在技术层面的优势，更重要的是它为 AI 应用的规模化部署和运营提供了坚实的基础。 松散耦合设计 松散耦合是事件驱动架构的核心特征，也是其在 AI 领域应用的重要价值所在。在 AI 系统中，不同的组件往往由不同的团队开发和维护，包括数据工程团队、算法团队、平台团队等。松散耦合的设计使得这些团队能够独立地开发和部署各自的组件，而不需要过多地考虑其他组件的实现细节。 这种设计理念特别适合AI项目的迭代开发模式。AI 算法和模型往往需要频繁地更新和优化，如果系统采用紧耦合的设计，每次算法更新都可能需要修改多个组件。而在松散耦合的架构中，算法的更新只需要修改相应的事件处理逻辑，不会影响其他组件的正常运行。 可扩展性/稳定性保障 AI 应用的负载往往具有很强的不确定性和波动性。在某些时段，系统可能需要处理大量的数据和请求；而在其他时段，系统的负载可能相对较低。事件驱动架构的可扩展性特征使得系统能够根据实际负载动态调整资源配置。 在事件驱动架构中，每个组件都可以独立地进行扩展。当某个组件的处理能力不足时，可以增加该组件的实例数量，而不需要扩展整个系统。这种细粒度的扩展能力使得资源配置更加精确和高效。 端到端的实时传输 AI 应用往往对实时性有很高的要求，特别是在实时推理、智能客服等场景中。事件驱动架构的实时传输特性使得系统能够快速响应各种事件，大大提高了 AI 应用的实时性。 采用事件驱动架构的系统在可扩展性、可维护性和可靠性方面都有显著提升。在 AI 领域，这些优势更加明显，因为 AI 应用往往需要处理更加复杂和多变的数据处理需求。 三、解决方案详解EventBridge 多源 RAG 能力 EventBridge 多源 RAG 能力代表了事件驱动架构在 AI 数据处理领域的重要突破。这一能力的核心价值在于将传统的数据处理流程与现代 AI 技术深度融合，为企业构建智能化的数据管道提供了全新的技术范式。 1. 多源数据接入 在 AI 时代，数据源的多样性和复杂性达到了前所未有的程度。EventBridge 多源 RAG 能力通过统一的接入框架，实现了对各种异构数据源的无缝集成，为 AI 应用提供了丰富的数据基础。 非结构化数据 Loader 技术 非结构化数据在 AI 应用中占据越来越重要的地位，特别是在 RAG（检索增强生成）场景中。EventBridge 针对非结构化数据的处理需求，开发了一套完整的 Loader 技术体系，支持多种数据格式的智能解析和处理。 分块处理策略是非结构化数据处理的关键技术。在 RAG 应用中，长文档需要被分割成适当大小的文本块，以便进行向量化和检索。EventBridge 提供了多种分块策略，包括基于字符数的固定分块、基于段落结构的逻辑分块等。 单文档与批量加载是系统设计中的重要考虑因素。对于实时性要求较高的场景，系统支持单文档的即时加载和处理，确保新文档能够快速进入 RAG 系统。对于批量数据处理场景，系统提供了高效的批量加载机制，支持并行处理和断点续传，确保大规模数据的可靠处理。 结构化数据源集成 结构化数据源的集成是 EventBridge 多源 RAG 能力的另一个重要组成部分。与非结构化数据不同，结构化数据具有明确的格式定义和字段结构，但其集成挑战主要体现在数据源的多样性和实时性要求上。 数据系统集成是结构化数据接入的重要方式。EventBridge 支持主流的消息队列（如 Kafka、RocketMQ、MQTT），日志服务（如 SLS，Simple Log Service），数据库服务（如 MySQL）等。 实时流数据处理是 EventBridge 在结构化数据处理方面的重要创新。系统采用了流式处理架构，能够实时处理高吞吐量的数据流。可实现复杂的流式数据转换和聚合操作，为实时 RAG 应用提供丰富的事件源。 2. 向量数据库入库优化 向量数据库是 RAG 应用的核心基础设施，EventBridge 在向量数据库入库方面的优化为 RAG 应用提供了高效、可靠的数据存储支撑。 向量数据库支持 随着 AI 技术的发展，向量数据库市场呈现出百花齐放的态势。不同的向量数据库在性能特征、功能特性、成本结构等方面各有优势。EventBridge 通过提供统一的向量数据库接入接口，支持 Dashvector、Milvus 等主流向量数据库产品，为用户提供了灵活的选择空间。 传统数据库向量插件兼容 除了向量数据库，许多传统数据库也推出了向量扩展插件，如 PostgreSQL 的 PGvector、MySQL 的向量索引等。这些向量插件使得用户能够在现有的数据库基础设施上实现向量存储和检索功能，降低了系统的复杂度和成本。EventBridge 也即将支持向量插件方式入库，为开发者带来更多选择。 一键白屏化入库体验 为了降低向量数据库使用的技术门槛，EventBridge 提供了一键白屏化的入库体验。用户只需要通过简单的图形界面配置，就能够实现复杂的向量数据处理和入库流程。 + 直观的拖拽式配置界面：用户可以通过拖拽的方式配置数据源、处理逻辑、目标数据库等，系统会自动生成相应的处理流程。这种可视化的配置方式大大降低了系统使用的复杂度。 + 丰富的预置模板：涵盖常见的 RAG 应用场景，用户可以基于这些模板快速搭建自己的数据处理流程，然后根据具体需求进行定制化调整。 + 完善的监控仪表板和告警机制：用户可以实时查看数据处理的状态、性能指标、错误信息等，及时发现和解决问题。 四、解决方案详解实时推理与异步推理能力__ EventBridge 在推理接入方面提供了灵活的选择，支持实时推理和异步推理两种模式，以满足不同应用推理场景的需求。 1. 智能数据转换能力 数据转换是 ETL 流程中最复杂也是最关键的环节。EventBridge 多源 RAG 能力在数据转换方面的创新，主要通过深度集成大语言模型（LLM）的推理能力，将其自然语言理解和生成能力引入数据处理流程，实现传统 ETL 工具难以处理的复杂数据转换任务。 + 在数据清洗方面，LLM 能够智能识别和处理各种数据质量问题。例如，基于上下文自动纠错包含拼写错误的文本数据；将格式不规范的地址信息标准化为统一格式；将包含缩写和俚语的文本转换为标准的表达方式。 + 在数据增强方面，LLM 能够为原始数据添加语义信息，为 AI 应用提供更加丰富的数据基础。例如，从产品描述文本中自动提取产品的关键特征和属性；分析用户评论的情感倾向和关键观点；提取新闻文章的关键事件和实体信息。 + 在数据转换方面，LLM 能够实现复杂的格式转换和结构重组，大大降低了数据集成的复杂度。例如，将非结构化的文本转换为结构化的 JSON 格式；将表格数据转换为自然语言描述；将多种数据源的信息融合为统一的数据模型等。 2. 结构化输出技术 结构化输出是 EventBridge 在 AI 数据处理方面提供的重要能力。传统的大语言模型输出通常是非结构化的自然语言文本，往往需要进行二次解析和处理（例如，使用正则表达式、自然语言处理工具等方法从模型输出中提取结构化信息），这个过程不仅复杂而且容易出错。EventBridge 支持结构化输出，使得大语言模型能够直接生成 JSON、XML 等结构化的数据格式，避免了二次解析的复杂性，大大简化了数据处理流程。 在实际应用中，结构化输出技术能够支持复杂的数据结构定义。用户可以定义包含嵌套对象、数组、枚举值等复杂结构的输出格式，模型会严格按照定义的格式生成输出。这种精确的格式控制使得 AI 数据处理能够与下游系统无缝集成。 + JsonSchema 原生支持：JsonSchema 是 JSON 数据格式的标准化描述语言，广泛应用于 API 设计和数据验证。EventBridge 提供了对 JsonSchema 的原生支持，允许用户定义期望的输出格式，并确保模型输出严格符合定义的格式。JsonSchema 支持复杂的数据结构定义，包括字段类型、约束条件、默认值等详细信息，系统会在模型推理过程中进行实时验证和纠正，确保数据处理结果的一致性和可靠性。对于支持 JsonSchema 的模型，系统会优先使用其原生能力进行结构化输出，通常具有更好的性能和更高的准确性。 + 提示词注入优化：对于不支持 JsonSchema 原生能力的模型，EventBridge 采用智能提示词注入技术。系统会分析用户输出格式，并结合模型特性，生成相应的提示词模板并注入到模型的输入中，引导模型生成符合要求的结构化输出。系统将根据模型的输出质量动态调整提示词，优化输出效果。在多轮对话和上下文处理方面，系统能够智能管理提示词上下文，维护完整的对话历史，确保每轮输出的格式一致。因此，EventBridge 能够支持各类大语言模型，无论其是否原生支持结构化输出，都能生成高质量的结构化数据，为用户提供更广泛的模型兼容性和更大的选择空间。 五、EventBridge for AI ETL 的最佳实践 _ _ EventBridge 通过丰富的应用场景实践，为不同行业和业务需求提供了完整的解决方案。这些最佳实践不仅验证了技术方案的可行性，更为企业在 AI 数据转型过程中提供了宝贵的经验参考。 1. 数据预处理（Data Preprocessing） 数据预处理是机器学习和 AI 应用中的关键环节，数据质量直接影响模型的训练效果和推理准确性。EventBridge 在数据预处理方面提供了全面的解决方案，特别是在 SFT（Supervised FineTuning）训练数据准备方面展现出独特优势。 训练数据准备 监督微调（SFT）是大语言模型训练中的重要环节，需要高质量的标注数据来指导模型学习特定任务的能力。EventBridge 通过事件驱动的数据处理流程，能够高效地处理和准备 SFT 训练所需的数据集。 在数据收集阶段，系统能够从多个数据源实时收集原始数据，包括用户对话记录、文档库、知识库、API 响应等。通过事件驱动的方式，当新的数据产生时，系统能够立即触发数据处理流程，确保训练数据的时效性。例如，当客服系统产生新的对话记录时，系统会自动提取对话内容，进行格式标准化，并添加到训练数据集中。 在数据标注方面，EventBridge 集成了大语言模型的能力，能够实现半自动化的数据标注。系统可以使用预训练的模型对原始数据进行初步标注，然后通过人工审核和修正，形成高质量的训练样本。这种人机结合的标注方式大大提高了数据标注的效率和质量。 在数据格式转换方面，不同的模型训练框架往往需要特定的数据格式。EventBridge 能够借助中间函数计算节点，根据目标模型的要求，自动将数据转换为相应的格式。例如，将对话数据转换为 ChatML 格式，将问答数据转换为 Alpaca 格式等。这种自动化的格式转换能力大大简化了数据准备的工作量。 数据错误处理 数据错误是影响 AI 模型性能的重要因素，包括数据缺失、格式错误、逻辑错误等多种类型。EventBridge 通过 LLM 节点和函数节点，能够有效提高数据质量。 在数据缺失处理方面，系统采用了多种策略。对于数值型数据，可以使用均值、中位数、众数等统计方法进行填充；对于分类型数据，可以使用最频繁的类别进行填充；对于文本数据，可以使用语言模型生成合理的填充内容。系统会根据数据的特征和业务需求自动选择最合适的填充策略。 在格式错误处理方面，系统能够智能识别和修复各种格式问题。例如，对于日期时间数据，系统能够识别多种日期格式并统一转换为标准格式；对于数值数据，系统能够处理千分位分隔符、货币符号等格式问题；对于文本数据，系统能够处理编码问题、特殊字符等。 在逻辑错误处理方面，系统通过规则引擎和机器学习模型相结合的方式进行检测和修复。例如，检测年龄数据是否在合理范围内，检测地址信息是否符合地理逻辑，检测业务数据是否符合业务规则等。当发现逻辑错误时，系统会根据预定义的修复策略进行自动修复或标记为需要人工处理。 数据重复去除 数据重复是大规模数据处理中的常见问题，特别是在多源数据集成的场景中。EventBridge 提供了多过滤和中间过程转换，确保数据的唯一性和一致性。 在精确重复检测方面，系统通过哈希算法快速识别完全相同的记录。对于大规模数据集，系统采用了分布式哈希计算，能够在保证准确性的同时提高处理效率。 在近似重复检测方面，系统使用了多种相似度计算方法。对于文本数据，采用编辑距离、余弦相似度、Jaccard 相似度等方法；对于数值数据，采用欧氏距离、曼哈顿距离等方法；对于结构化数据，采用字段级别的相似度计算。系统会根据数据类型和业务需求自动选择最合适的相似度计算方法。 在重复处理策略方面，系统提供了多种选择。可以保留最新的记录，保留质量最高的记录，或者将重复记录合并为一条记录。系统还支持自定义的重复处理逻辑，用户可以根据具体的业务需求定义重复处理规则。 数据缺失补全 数据缺失是实际数据中的普遍现象，如何合理地处理缺失数据对模型性能有重要影响。EventBridge 可通过中间函数计算节点对数据进行处理和补全，提高数据的完整性和可用性。 在统计方法补全方面，系统支持多种经典的统计填充方法。对于数值型数据，可以使用均值、中位数、众数、线性插值、多项式插值等方法；对于分类型数据，可以使用众数、随机填充、基于分布的填充等方法。系统会根据数据的分布特征自动选择最合适的填充方法。 在机器学习方法补全方面，系统使用训练好的模型来预测缺失值。例如，使用回归模型预测数值型缺失值，使用分类模型预测分类型缺失值。这种方法能够考虑数据之间的复杂关系，通常能够获得更好的填充效果。 在深度学习方法补全方面，系统采用了自编码器、生成对抗网络等深度学习技术。这些方法能够学习数据的深层特征和复杂模式，对于高维数据和复杂缺失模式具有更好的处理效果。 2. RAG 检索增强生成（Retrievalaugmented Generation）数据入库 RAG 技术作为当前 AI 应用的重要范式，其数据入库过程的质量直接影响检索和生成的效果。如上文介绍 EventBridge 在 RAG 数据入库方面提供了从简单到复杂的完整解决方案。 简易数据入库流程 对于单一数据源的 RAG 应用场景，EventBridge 提供了简化的数据入库流程。这种流程特别适合快速原型开发和小规模应用部署。 在数据源配置方面，用户只需要指定数据源的类型和连接信息，系统会自动处理数据的读取和解析。支持的数据源包括对象存储（OSS）中的文档文件、数据库中的文本字段、API 接口返回的文本数据等。 在文档处理方面，系统提供了自动化的文档解析和分块处理。对于 PDF、Word、TXT 等常见文档格式，系统能够自动提取文本内容，并根据文档结构进行智能分块。分块策略可以基于段落、章节、固定长度等多种方式，用户可以根据具体需求进行配置。 在数据入库方面，系统支持多种向量数据库，用户可以选择最适合的存储方案。系统会自动处理向量数据的索引构建和存储优化，确保检索性能和存储效率。 增强多源数据入库 对于复杂的企业级 RAG 应用，往往需要整合来自多个数据源的信息。EventBridge 的增强多源数据入库能力能够处理这种复杂场景的需求。 在多源数据协调方面，系统提供了统一的数据处理框架。不同数据源的数据会被转换为统一的内部格式，然后进行统一的处理和存储。这种设计使得系统能够无缝地处理来自不同源的异构数据。 在数据融合方面，系统能够智能地合并来自不同源的相关信息。例如，将产品数据库中的产品信息与用户评论系统中的评论信息进行关联，形成更加完整的产品知识。系统使用实体识别和关系抽取技术来发现数据之间的关联关系。 在数据一致性方面，系统提供了多种一致性保证机制。包括数据版本管理、冲突检测和解决、数据同步等。当多个数据源包含相同实体的不同信息时，系统能够智能地进行信息合并和冲突解决。 在增量更新方面，系统支持实时的数据更新和同步。当源数据发生变化时，系统能够自动检测变化并更新相应的向量表示。这种增量更新能力确保了 RAG 系统能够获取最新的信息。 3. 数据清洗与标准化 (Data Cleansing & Standardization) 数据清洗与标准化是数据处理中的基础环节，EventBridge 通过集成大语言模型的能力，实现了智能化的数据清洗和标准化处理。 地址信息的标准化是一个典型的数据清洗场景。在实际应用中，来自不同渠道的地址信息往往格式不一、存在错别字、缺少关键信息等问题。EventBridge 通过 LLM 驱动的地址标准化能力，能够将各种格式的地址信息转换为统一的标准格式。 在地址解析方面，系统使用大语言模型来理解地址的语义结构。即使地址信息存在格式不规范、字段顺序混乱等问题，模型也能够准确识别出省、市、区、街道等各个组成部分。例如，对于"北京市海淀区中关村大街1号"这样的地址，系统能够准确解析出省份为"北京"，城市为"北京"，区县为"海淀区"，街道为"中关村大街1号"。 在错误纠正方面，系统能够自动识别和纠正地址中的错别字和格式错误。例如，将"北京市海定区"纠正为"北京市海淀区"，将"中关村大街一号"标准化为"中关村大街1号"。系统使用了地理知识库和语言模型相结合的方法，确保纠正的准确性。 在信息补全方面，系统能够根据已有的地址信息补全缺失的部分。例如，根据详细地址自动推断邮政编码，根据区县信息补全城市和省份信息。系统集成了完整的地理信息数据库，能够提供准确的地理信息补全。 在格式标准化方面，系统将所有地址信息转换为统一的标准格式。输出格式包括结构化的字段信息（省、市、区、街道、邮政编码等）和标准化的地址字符串。这种标准化的输出格式便于后续的数据处理和分析。 4. 合规与隐私保护 (Compliance & Privacy Protection) 在数据处理过程中，合规与隐私保护是不可忽视的重要方面。EventBridge 提供了全面的合规和隐私保护能力，确保数据处理符合相关法规要求。 数据脱敏是保护隐私的重要技术手段，EventBridge 可借助大模型，实现隐私信息识别，隐私信息处理等多种数据脱敏处理场景。为用户提供端到端的数据处理能力。 通过这些全面的合规和隐私保护措施，EventBridge 可确保了企业应用中的数据处理过程的合法性和安全性，为企业数据提供可靠的合规安全保证。 六、未来展望与发展路径__ AI 数据处理领域技术正处于快速发展的阶段，未来几年将会出现更多的技术创新和应用突破。EventBridge 作为这一领域的重要参与者，期待与更多开发者共同推动 AI 数据处理技术的发展和应用。 AI 数据处理技术的发展方向 在技术发展方向上，AI 数据处理将朝着更加智能化、自动化、实时化的方向发展。 智能化方面，未来的数据处理系统将更加依赖 AI 技术来实现自动化的数据理解、清洗、转换和增强。大语言模型的能力将进一步提升，能够处理更加复杂的数据处理任务。同时，多模态 AI 技术的发展将使得系统能够统一处理文本、图像、音频、视频等多种类型的数据。 自动化方面，数据处理流程将更加自动化，减少人工干预的需求。自动化的数据发现、数据治理、数据安全等功能将成为标准配置。机器学习技术将被广泛应用于数据处理流程的优化，系统能够自动学习和改进处理策略。 实时化方面，实时数据处理将成为主流需求。边缘计算技术的发展将使得数据处理能够在更接近数据源的地方进行，减少数据传输的延迟。流式处理技术将进一步成熟，能够支持更加复杂的实时数据处理场景。 行业标准化趋势 随着 AI 数据处理技术的成熟，行业标准化将成为重要趋势。标准化有助于降低技术门槛、促进技术交流、推动产业发展。在数据格式标准化方面，将出现更多的行业标准来规范数据的格式和结构，例如 CloudEvents 等。这些标准将有助于不同系统之间的数据交换和集成。 在 API 接口标准化方面，将出现统一的 API 规范来规范数据处理服务的接口。这些规范将有助于提高系统的互操作性和可移植性。 技术挑战与解决方案 未来的发展过程中，AI 数据处理技术仍将面临一些挑战，需要持续的技术创新来解决。 在数据质量挑战方面，随着数据源的增加和数据量的增长，数据质量问题将更加突出。需要开发更加智能的数据质量检测和修复技术，利用 AI 技术来自动识别和处理数据质量问题。 在性能挑战方面，随着数据处理需求的增长，系统的性能要求将更加严格。需要在算法优化、架构设计、硬件加速等方面持续创新，提升系统的处理能力。 在成本挑战方面，AI 数据处理的成本仍然较高，特别是大语言模型的调用成本。需要通过技术优化、资源调度、成本控制等手段来降低使用成本。 在安全挑战方面，数据安全和隐私保护的要求将更加严格。需要在数据加密、访问控制、隐私计算等方面持续投入，确保数据的安全性。 结语：构筑 AI 时代的数据传输基础设施 通过将事件驱动架构的技术优势与 AI 时代的数据处理需求深度融合，EventBridge 为企业构建智能化数据管道提供了全新的技术范式。 欢迎更多的数据侧伙伴加入 EventBridge 的生态体系，共同构建更多的数据源连接器、处理算法、应用模板等。通过开放的技术合作，更好地满足用户的多样化需求。AI 时代的数据基础设施建设是一个长期的过程，需要持续的技术创新和生态建设。我们相信，通过持续的努力和合作，我们能够构建更加智能、高效、可靠的 AI 数据集成基础设施，为人工智能技术的发展和应用提供强有力的支撑。 欢迎加入 EventBridge 用户交流群（钉钉群号：31481771）进行交流～

沈林，Apache RocketMQ PMC 成员，阿里云 EventBridge 负责人，专注于 EDA 研究。本文整理自作者在 Community Over Code Asia 2025 会议发表的主题演讲《Apache RocketMQ EventBridge: Why Your GenAI Needs EDA？》。 EDA 的核心特点是：以事件为中心，实时响应变化。它不像传统“请求响应”模式那样被动等待，而是“感知→触发→行动”全自动流转。在 AI 系统中，数据流、模型训练和推理、外部反馈等都可以作为“事件”，触发 AI 自动决策和联动执行。EDA 就像是 AI 时代的“神经系统”，让 AI 不仅能“思考”，还能“感知”和“行动”。它提升了系统的实时性、灵活性和自动化水平，是构建智能系统的关键支撑。AI 赋予系统“大脑”，EDA 构建系统的“神经”。 本文主要探讨在 AI 时代，EDA 的重要价值及它可以帮助我们解决的问题。 EDA 的第一重价值：通过 RAG 缓解 AI 幻觉 大家可能还有印象，2023 年上半年，Google 的早期 AI 模型发布时，回答一个关于詹姆斯·韦伯空间望远镜的问题时，犯了一个低级“错误”，这个答案本来在 Google 上很容易搜索到，但是 AI“一本正经”的给了一个错误答案，直接导致谷歌当天的股价跌了 8% 左右。但 AI 完全没有意识到自己的错误，这是为什么？ 1. 为什么会有 AI 幻觉？ AI 幻觉的产生机制比较复杂，可简单从训练和推理两个阶段进行分析： + 训练阶段： 数据覆盖不足：若训练数据不包含特定信息，模型无法“无师自通”； 过拟合：模型过度学习训练数据中的细节与噪声，导致在面对新数据时泛化能力差； 通用性与精度取舍：通用大模型为覆盖广泛领域，在特定垂直领域的准确性可能有所牺牲。 + 推理阶段： 自回归生成：LLM（大语言模型）推理本质上是一个自回归过程，基于现有 Token 预测下一个最可能的 Token，这种概率性生成机制使得幻觉成为其固有潜在分布的一部分。 连贯性优先于准确性：GenAI 输出的时候倾向于生成流畅连贯的答案，而非绝对准确的答案。 2. 如何减少 AI 幻觉？ 为了解决 AI 幻觉，现在一般有三种主流的方式： + 模型微调（Finetuning）： 模型不好？最能直接想到的方法就是优化模型：丰富模型训练的数据、优化模型的参数，让其在垂直场景领域，回答更加精准。这种方式在很多场景是非常有效的，而且依旧被广泛采用。但是这种方式，要求也是比较高的，如果没有一定的人力和算力成本投入，将很难实现。尤其是在知识更新频繁的领域，模型需要不断调整，长期维护，投入代价相对较高。 + 提示词工程（Prompt Engineering）： 那可不可以不调整模型，而是在向 LLM 提问的时候，把相关的数据和限定条件一起给到 LLM？答案是可以的，这就是提示词工程。 但是如何构造一个好的提示词，把 LLM 需要的上下文信息给到它，这个要求也是非常高的。不同人使用，提示词的构造水平也不同：这种方式就像是把“问问题”变成了一件“手工艺术活”。而且提示词优化虽然可以“压平”部分幻觉，但只要模型权重未变，提示词没有带上相关数据，提示词只能暂时把幻觉“藏”起来，而无法真正去除。 + 检索增强生成（RAG）： 那可不可以自动帮我们生成一个高质量的提示词呢？在这个提示词中，包含了 LLM 回答需要的关键信息。这个就和我们最后一个要讲的 RAG 非常像了，让我们看下 RAG 到底是什么。 3. 什么是 RAG？ RAG 可以简单理解为：向 LLM 提问的时候，同时给这个问题，检索一个上下文，一起给到 LLM。比如：如果我们问 DeepSeek：本次 Apache 峰会有哪些讲师聊到了 RAG 这个话题？DeepSeek 肯定不知道，因为它没有这个数据，网上暂时也还没有相关数据。但如果给到它一个关于本次大会讲师的讲稿资料包。这样 DeepSeek 就有非常强相关的上下文，回答问题的时候，就不会跑题答偏。 那 AI 要如何做到这一点呢？主要分两步： + 建立索引：首先，需要提前把讲师资料包存起来。但资料包可能非常大，我们需要快速找到跟提问的问题相关联的数据，这里就需要用到向量化。向量化本质上是对一个事物，从多个特征维度，进行数值标记。比如，标记我这个人，可以从年龄、身高、性别等多个特征标记，标记越多越清晰。如果两个向量在多维空间中的“位置”越接近，说明它们越相似。所以，我们需要提前把数据进行向量化，存到向量数据库里。 + 检索生成：然后，当我们向 LLM 提问时，可以先把问题向量化，根据向量化后的结果，去向量数据库查询关联性最大的原始知识数据。最后，将查到的知识数据，作为上下文和问题一起传给 LLM，LLM 就可以给一个更加准确的回答了。 从这个过程中，我们会发现 RAG 有两个非常明显的优点： + 不需要用知识库数据给大模型训练，既节省了成本，又保证了数据隐私； + 不需要用提示词工程这样的“手工艺术活”，就可以让 AI 出现幻觉的概率变得足够低。 4. 为什么 EventBridge 适合做 RAG？ 为什么是 EventBridge 适合来做 RAG 呢？ 我们先来看下什么是 EventBridge： + EventBridge 的整个模型其实非常简洁：我们从下图左侧开始往右看，EventBridge 可以方便的把外部的数据，以标准化的事件格式，配合事件 Schema 集成到内部，中间可以存入事件总线（BUS），也可以选择不存储，然后通过过滤/转换，推送到下游服务中。 + 这个链路，正好可以满足 RAG 过程中需要的三要素：获取上游丰富的数据、自定义切分和向量化、持久化到多种向量化数据库中。 5. EventBridge 如何实现 RAG？ 用一个场景举例，比如我们想建一个关于 EventBridge 知识的智能问答机器人，可以回答关于 EventBridge 的常见问题： + 我们需要把存在上游 OSS 的 EventBridge 文档，通过 EventBridge 的事件流，进行 Chunk 切分、Embedding 向量化，然后存储到向量数据库； + 完成之后，当我们向 EventBridge 智能机器人提问“EventBridge 是什么？”，智能机器人会先把这个问题向量化，然后去向量数据库查找匹配度最高的相关内容，并一起传递给 LLM，LLM 就能结合查到的资料，给出非常精准的回答，减少幻觉产生。 目前，阿里云大模型服务平台百炼的知识库 RAG 场景，已采用 EventBridge 的事件流能力，帮助众多客户减少了 LLM 问答中的幻觉问题，尤其在细分垂直领域效果显著。如果您感兴趣可以进行体验： _https://bailian.console.aliyun.com/?&tab=app/knowledgebase_ EDA 的第二重价值：推理触发器（Inference Trigger） 我们第二个想讨论的场景是推理触发器。 1. 程序使用 LLM 的规模将远超人类 目前，我们日常接触最多的 LLM 场景是人与 LLM 服务直接对话，如问 DeepSeek 一个问题或智能客服等。 但更常见且增长迅速的方式是程序触发 LLM。例如供应链优化和金融订单风控。 观察微服务就会发现，人调用 API 的量级远不如程序调用 API 的量级。相应地，我们可以想象，未来程序触发 LLM 的规模，也将远远超过人工使用 LLM。 这其中的机会，我们应该怎么把握？ 2. 推理诉求无处不在 事实上，我们现有的商业系统中，已经存储了大量现成需要推理的场景。比如： + 消息服务里，存储了客户的评论，需要对其打标分析，这条评论是积极的还是消极的，并给个分数； + DB 里存储了产品的描述介绍信息，想让 AI 给一些产品描述优化建议； + OSS 或 S3 存储了大量的文档，想让 AI 对每个文档生成一个 100 字的文档摘要。 这些诉求在以前可能需要人工处理，但现在都可以交给 LLM，从而极大提升工作效率。那怎么让现有的商业系统，方便、快捷、低成本的使用 AI，甚至不需要写一行代码，这个就是 EventBridge 擅长的地方了。 3. 推理触发器：让模型被更好地使用 为此，EventBridge 提供了三把武器： + 第一把武器——实时推理并将结果存到目标端：通过 EventBridge 可以实时监听并获取存在 DB、消息、或者存储服务中的数据，然后实时调用 LLM 推理服务，并将推理结果输出存到目标端。此过程也可以结合上一部分讲到的 RAG，但中间不一定是一个 LLM，也可以是一个 Agent，甚至是一个 AI Workflow。 这个过程看似简单，但有很多需要注意的地方： 数据合并：我们刚才聊到，LLM 的推理本质上是一个自回归过程，这次的输出会作为下一次的输入，无法一次性拿到结果，很多 LLM 只能支持以流式的方式返回数据，但下游往往需要的是一个确定性的结果，所以我们需要对流式数据进行合并再输出； 数据格式：很多业务场景下有明确的格式要求。比如上面提到的例子，让 LLM 对客户评论打标和评分，需要输出一个 JSON 结构。但不是所有 LLM 在 API 层面都支持 JSON 结构输出，我们需要通过提示词进行优化，让它尽可能输出一个符合要求的 JSON 结构。 推理吞吐：LLM 的自回归生成方式，导致单次请求 RT 长、TPS 低。所以，我们需要提升高并发能力，把昂贵的 GPU 资源使用效率发挥到极致，同时需要做好 TPM 和 RPM 的限流，也就是每分钟请求次数和每分钟 Token 数的限流，以保证链路不会有大量限流异常。 可以看到，具体落地会遇到很多挑战，但 EventBridge 可以帮助客户便捷高效地解决这些问题。 + 第二把武器——基于推理结果触发任务执行： 除了让 LLM 推理输出结果存到目标端，EventBridge 还可以让 Agent 基于上游的某条消息，去调用某一个 Service，执行某一个动作，如发送邮件。 + 第三把武器——离线异步推理提高资源利用率： 对于实时性要求不高的推理场景，可以通过 EventBridge 实现离线异步推理，让稀缺的 GPU 资源被更好地调度利用，在云上的成本至少比实时推理便宜一半。 AI 的强大在于其应用，而 EDA（事件驱动架构）非常适合作为推理触发器，激活 AI 的价值。 EDA 的第三重价值：构建 Agent 通信基础设施 现在 AI Infra 非常热门，其概念非常广泛。对标 IT Infrastructure，我们这里讨论的话题是 AI 的通信。 1. 微服务的通信离不开 Messaging，Agent 间的通信应该如何？ 在微服务时代，消息系统在微服务间的通信中扮演了重要角色。到了 AI 时代，消息系统是否依然起着关键作用？具体形式和产品又会有哪些变化？ 2. Agent 和 Service 的通信：Function Calling、MCP 为了回答这个问题，我们先看下现在 AI 的通信是怎么做的。 首先，我们看下 Agent 和传统 Service 之间的通信。目前有两种主流的方式：Function Calling和MCP。 + Function Calling：是 OpenAI 公司在 2023 年提出的。因为 LLM 本身是文本生成器, 不具备访问外部系统的能力。但是我们可以对 LLM 进行训练微调，让 LLM 理解外部的一些工具函数的定义，这样在遇到提问时，就可以按需生成这些工具函数需要的参数，然后调用这些工具函数。 + MCP：是 2024 年 11 月 Anthropic 提出的，全称 Model Context Protocol‌，其本意是用来解决 LLM 无状态的问题。LLM 每次调用都是独立的，而 MCP 是用来给 LLM 提供运行上下文，相当于一个“Session 机制”。但是为什么 MCP 会被拿来和 Function Calling 放在一起呢？因为它也可以拿来调用工具函数。和 Function Calling 不同的是，它不需要 LLM 提前训练微调来理解函数的入参和返回值，而是通过上下文“提示词”告诉 LLM 参数返回值。所以 MCP 相比 Function Calling，对模型的依赖更小，更加通用，但效果相比 Function Calling 要差一点。 3. Agent 和 Agent 的通信：A2A 我们再来看下 Agent 与 Agent 之间是怎么通信的。 Google 在今年 4 月份的时候，提出了一个 A2A 的通讯协议，其核心运行机制分为四步： + 第一步：Client Agent 通过 Agent Card，看下远端 Agent 有哪些能力； + 第二步：根据 Agent Card 的能力描述，调用远端 Agent，创建一个 Task，让其帮忙完成一个任务； + 第三步：由于任务可能比较耗时，不一定能够立即响应。所以 A2A 协议允许远端 Agent 通过 SSE 协议，不间断的将任务的状态信息更新给 Client Agent； + 第四步：再将结果返回给 Client Agent，当然结果本身也是可以流式返回的。 4. MCP 和 A2A 之间的区别 那 MCP 和 A2A 协议有什么区别呢？Google 给它们的关系做了一个描述： + A2A 协议像一种沟通语言：如果把 Agent 比做人，一个人如果能力有限，想让其他 Agent 帮忙怎么办？A2A 协议就可以派上用场了，A2A 协议像一种沟通语言，可以让 Agent 和其他 Agent 用同一种语言交流，不至于说话的时候驴唇不对马嘴。 + MCP 就像工具使用说明书：Service 等价于人使用工具，可以提升人解决问题的能力。不过，使用工具也需要有些技巧，MCP 就像工具使用说明书，可以让Agent 更方便的使用这些 Service，来扩展 Agent 的能力。 我们把 Agent 比做人，把 Service 比做工具。但是，请人帮忙和请工具帮忙，真的可以分得这么清楚吗？ 5. 预测 1：A2A and MCP 可能走向融合 A2A 和 MCP的职责，设想很完美，但是实际运行的时候会遇到很多挑战。我们先来一起看看在 MCP 和 A2A 协议下，两者用来声明一个 Agent 或者 Service 能力的时候是怎么样的？ + 这里举例了一个“查询北京天气”的服务，会发现两者声明自己能力的时候，非常类似： + 其次，两者的传输层协议也都非常相似，都支持 SSE 和 JSONGRPC； + 最后，我们从工程师开发角度，推演一个场景：当一个 Agent 需要获取“查询天气”的能力时，它并不真正关心该能力是由一个 Service 还是另一个 Agent 提供的。Agent 的核心关注点在于能力的接口定义：即有哪些可用能力、如何调用，以及预期的返回结果是什么。至于该能力的后端提供者是 Service 还是 Agent，对于调用方而言是无需关注的实现细节。 这里我们的第一个预测是：A2A and MCP 未来可能会合并，但具体怎么发展，还要看生态的选择。 6. 预测 2: 点对点的通信是不够的 这里的第二个预测是：现有 MCP 和 A2A 协议中，只包含的点对点通信是不够的。按照 A2A 协议的推演，当一个系统中有很多 Agent 时，所有 Agent 都通过“长连接”集成在一起：大家第一个直观的感受是什么？ 连接太多了！ 如果两个 Agent 通过“长连接”集成在一起，感觉可能也没有什么。但是如果一个 Agent 同时需要和数百个甚至上千个 Agent 通信，系统中就会产生大量的长连接。 + 首先对每一个 Agent 来讲，资源开销就非常大； + 其次，网状的连接，一旦某一个 Agent 出现问题，hang 住了某些资源，会不会拖垮其他 Agent 的服务？甚至拖垮整个系统？这类问题在微服务中，再常见不过了。 + 最后，即使不会被出问题的 Agent 拖垮服务，但当这个出问题的 Agent 恢复时，之前的通讯是否依旧可以继续追踪？再次执行，是否已经幂等，是否有风险？ 这里面有非常多的稳定、性能、成本、扩展性的挑战。这些问题在微服务中已经被多次验证过，有些经验我们可以学习过来。 7. EventBridge Super Agent 基于上面两个预测判断，我们给出了一个 RocketMQ EventBridge 的回答: 在这个模型中，我们引入了一个 EventBridge Agent Proxy 的角色。我们姑且称它为“Super”Agent ，但它不是一个真正的 Agent，而是可以代理 Agent 的能力。 + 首先，所有 Agent 都可以写一份自己的个人简历，把自己拥有的能力，注册到“Super”Agent上； + 如果某个 Agent 需要调用其他 Agent 的能力，它可以在 “Super” Agent 中查找是否有其需要的 Agent。如果有，就可以直接通过与 “Super” Agent 的交互，来获得这个能力； + 当这个 Agent 需要多个其他 Agent 的能力时，也不需要和每一个 Agent 交互，都可以通过 “Super” Agent 代理实现，将原本的 N:N 模型简化为 1:1 模型。 + 除此之外，“Super” Agent 中的 Proxy 除了 A2A 协议，还会路由和跟踪每一个 Task 的运行状态，即使在异常/重启/集群扩容等场景下，每一个 Task 都能被按预期处理，并把状态同步回 Agent。 “Super”Agent 和微服务注册中心有点类似，不过区别在于，它不光是提供了微服务查找寻址的作用，同时还起到了服务代理的作用。如果我们脑洞再大一点，可以不仅局限于 Task 级别的任务追踪和管理，甚至还可以往上考虑一层，提供“User”级别的上下文： + 我们现在的 Agent 都是没有记忆的，我们之前跟它说过的话，过几天再问它，它就不记得你了。 + 但是每个人使用工具的习惯是不一样的。如果 Agent 能更好的理解你，记得你，就可以提供更加人性化的服务。 + 作为 Agent 注册和代理中心，如果在为 Agent 提供代理的同时，还能同时提供“User”的上下文，并且用 Agent 的越多，“User”的身份画像越完善；反过来，Agent 越依赖，进入一个正向循环。 目前，EventBridge Agent 代理还处于理论探索阶段，欢迎大家一起交流。 参考文献与延伸阅读 + [RetrievalAugmented Generation for KnowledgeIntensive NLP Tasks, Lewis et al., 2020] + [Model Context Protocol (MCP) Specification, 2024] + [A2A: AgenttoAgent Communication Protocol, Google, 2024] + Apache RocketMQ EventBridge 官方文档： _https://rocketmq.apache.org/_

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

作者简介：艾阳坤，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 进行体验。 相关链接：