最新文章

2024年8月30日

基于 RocketMQ 的云原生 MQTT 消息引擎设计
作者|沁君 概述 随着智能家居、工业互联网和车联网的迅猛发展,面向 IoT(物联网)设备类的消息通讯需求正在经历前所未有的增长。在这样的背景下,高效和可靠的消息传输标准成为了枢纽。MQTT 协议作为新一代物联网场景中得到广泛认可的协议,正逐渐成为行业标准。 本次我们将介绍搭建在 RocketMQ 基础上实现的 MQTT 核心设计,本文重点分析 RocketMQ 如何适应这些变化,通过优化存储和计算架构、推送模型及服务器架构设计,推动 IoT 场景下消息处理的高效性和可扩展性以实现 MQTT 协议。 此外,阿里云 MQTT 以 RocketMQMQTT 为基础,不断进行迭代创新。阿里云是开源 RocketMQMQTT 的主要贡献者和使用者之一。面对设备通信峰谷时段差异性的挑战,本文将介绍阿里云如何将 Serverless 架构应用于消息队列,有效降低运营成本,同时利用云原生环境的特性,为 IoT 设备提供快速响应和灵活伸缩的通讯能力。 进一步地,我们将探讨介绍在云端生态体系中整合 MQTT 的实践,介绍基于统一存储的数据生态集成方案,展示其强大的技术能力和灵活的数据流转能力。 loT 消息场景 (消息场景对比) 物联网技术,作为当代科技领域的璀璨明星,其迅猛发展态势已成共识。据权威预测,至 2025 年,全球物联网设备安装基数有望突破 200 亿大关,这一数字无疑昭示着一个万物互联时代的到来。 更进一步,物联网数据量正以惊人的年增长率约 28% 蓬勃膨胀,预示着未来数据生态中超过 90% 的实时数据将源自物联网。这一趋势深刻改变了数据处理的格局,将实时流数据处理推向以物联网数据为核心的新阶段。 边缘计算的崛起,则是对这一变革的积极响应。预计未来,高达 75% 的数据处理任务将在远离传统数据中心或云端的边缘侧完成。鉴于物联网数据的海量特性,依赖云端进行全部数据处理不仅成本高昂,且难以满足低延迟要求。因此,有效利用边缘计算资源,就地进行数据初步处理,仅将提炼后的关键信息上传云端,成为了提升效率、优化用户体验的关键策略。 在此背景下,消息传递机制在物联网场景中的核心价值愈发凸显: + 桥梁作用:消息系统充当了物联网世界中的“神经网络”,无缝衔接设备与设备、设备与云端应用间的沟通渠道,构筑起云边端一体化的应用框架,确保了信息交流的即时与高效。 + 数据加工引擎:面对物联网持续涌动的数据洪流,基于消息队列(MQ)的事件流存储与流计算技术成为了解锁实时数据分析潜能的钥匙。这一机制不仅能够实时捕捉、存储数据流,还支持在数据产生的瞬间执行计算操作,为物联网应用提供了强大的数据处理基础架构,助力实现数据的即时洞察与决策响应。 总之,消息技术不仅是物联网架构的粘合剂,更是驱动数据流动与智能决策的核心动力,其在物联网领域的应用深度与广度,正随着技术迭代与市场需求的双重驱动而不断拓展。 同时传统消息场景和物联网消息场景有很多的不同,包含以下几个特点: 1)硬件资源差异 传统消息场景依托于高性能、高可靠性的服务集群,运算资源充沛,客户端部署环境多为容器、虚拟机乃至物理服务器,强调集中式计算能力。相比之下,物联网消息场景的客户端直接嵌入至网络边缘的微型设备中,如传感器、智能家电等,这些设备往往受限于极低的计算与存储资源,对能效比有着极高要求。 2)网络环境挑战 在经典的内部数据中心(IDC)环境中,消息处理享有稳定的网络条件和可控的带宽、延迟指标。而物联网环境则拓展至公共网络,面对的是复杂多变的网络状况,尤其在偏远地区或网络覆盖弱的区域,不稳定的连接成为常态,对消息传输的健壮性和效率提出了更高挑战。 3)客户端规模的量级跃迁 传统消息系统处理的日均消息量通常维持在百万级,适用于集中度较高的消息分发。物联网的兴起促使设备数量呈爆炸性增长,动辄涉及亿万级别的终端节点,这对系统的扩展性、消息路由的高效性提出了全新的要求。 4)生产与消费模式的演变 传统场景倾向于采用集中式同步生产模式,强调消息的一致性与有序处理。而物联网消息生成则体现出分散化的特性,每个设备根据其感知环境独立产生少量消息,这对消息收集与整合机制设计提出了新的思考。消费模式上,物联网场景经常涉及大规模广播或组播,单条消息可能触达数百万计的接收者,要求系统具备高效的广播能力和灵活的订阅管理机制。 RocketMQ 的融合架构设计 (融合架构设计图) 我们看到,物联网所需要的消息技术与经典的消息设计有很大的不同。接下来我们来看看基于 RocketMQ 的融合架构 MQTT 设计为了解决物联网的消息场景有哪些特点。 1)融入 MQTT 协议,适应物联网环境特性 RocketMQ 5.0 通过整合 RocketMQMQTT,紧密贴合了物联网领域广泛采用的MQTT协议标准。此协议针对低功耗、网络条件不稳定的情况优化,以其轻量级特性和丰富的功能集脱颖而出,支持不同的消息传递保障级别,满足了从“最多一次”到“仅一次”的多样化需求。协议的领域模型与 RocketMQ 的核心组件高度协调,促进了消息、主题、发布订阅模式的自然融合,为建立一个从设备到云端的无缝消息传递体系打下了稳固的基础。 2)灵活的存储与计算解耦架构 为了应对物联网场景下对高并发连接和大规模数据处理的需求,RocketMQ 5.0 采取了存储与计算分离的架构设计。RocketMQ Broker 作为核心存储组件,确保了数据的持久化与可靠性,而 MQTT 相关的逻辑操作则在专门的代理层实施,这不仅优化了对大量连接、复杂订阅关系的管理,也增强了实时消息推送的能力。这种架构允许根据业务负载动态调整代理层资源,通过增加代理节点来平滑应对设备连接数的增加,体现了系统良好的弹性和扩展潜力。 3)促进端云数据协同的整合架构 RocketMQMQTT 通过其整合的架构设计,促进了物联网设备与云端应用之间的高效数据共享。基于统一的消息存储策略,每条消息在系统内只需存储一次,即可供两端消费,减少了数据冗余,提高了数据流通的效率。此外,RocketMQ 作为数据流的存储中枢,自然而然地与流计算技术结合,为实时分析物联网生成的海量数据提供了便利,加速了数据价值的发掘过程。 存储设计 首先要解决的问题是物联网消息的存储模型。在发布订阅业务模型中,常用的存储模型有两种,写放大和读放大,我们将依次分析两种模型。 (写放大模型) (读放大模型) 写放大模型: 在此模型下,每位消费者拥有专属的消息队列,每条消息需要复制并分布到所有目标消费者的队列中,消费者仅关注并处理自己队列中的消息。以三级主题/Topic/subTopic/test 为例,若该主题吸引了大量客户端订阅,采取“一客一队列”的策略,即每个符合订阅规则的客户端或通配符订阅均维护一份消息副本,将导致消息的存储需求随订阅者数量线性增长。 尽管这种模式在某些传统消息场景,比如遵循 AMQP 协议的应用中表现得游刃有余,因为它确保了消息传递的隔离性和可靠性。但在物联网场景下,特别是当单个消息需被数以百万计的设备消费时,“写放大”策略将引发严重的存储资源消耗问题,迅速成为不可承受之重。 读放大的考量与挑战: 鉴于物联网场景的特殊需求,直接应用传统的“写放大”模型显然是不可行的。为解决这一难题,RocketMQMQTT 采取了更为高效与灵活的存储策略,旨在减少存储冗余,提高系统整体的可扩展性和资源利用率: 在“读放大”模式下,每条消息实际上被存储一次,而为了支持通配符订阅的高效检索,系统在消息存储阶段会创建额外的索引信息——即 consume queue(消费队列)。对于如/Topic/subTopic/这样的通配符订阅,系统会在每个匹配的通配符队列中生成相应的索引,使得订阅了不同通配符主题或具体主题的消费者,都能通过这些共享的存储实体找到并消费到消息。尽管这看似增加了“读”的复杂度,但实际上,每个 consume queue 作为索引,其体积远小于原始消息,显著降低了整体存储成本,同时提高了消息检索与分发的效率。 (原子分发示意图) 为此,我们设计了一种多维度分发的 Topic 队列模型,如上图所示,消息可以来自各个接入场景(如服务端的 MQ/AMQP、客户端的 MQTT),但只会写一份存到 commitlog 里面,然后分发出多个需求场景的队列索引(ConsumerQueue),如服务端场景(MQ/AMQP)可以按照一级 Topic 队列进行传统的服务端消费,客户端 MQTT 场景可以按照 MQTT 多级 Topic 以及通配符订阅进行消费消息。这样的一个队列模型就可以同时支持服务端和终端场景的接入和消息收发,达到一体化的目标。 实现这一模型,RocketMQ依托了两项关键技术特性: + 轻型队列(Light Message Queue) 这一特性允许一条消息被灵活地写入多个 topic queue 中,确保了消息能够高效地适应各种复杂的订阅模式,包括但不限于通配符订阅。它为读放大模型的实现提供了必要的灵活性和效率基础。 + 百万队列能力 RocketMQ 通过集成 RocksDB 这一高性能键值存储引擎,充分利用其在顺序写入方面的优势,实现了百万级别的队列管理能力。特别是通过定制化配置,去除了 RocksDB 内部的日志预写(WriteAhead Log, WAL),进一步优化了存储效率。RocksDB 不仅为 consume queue 提供了稳定高效的存储方案,还确保了即便在极端的队列数量下,系统依然能够保持高性能的索引处理能力。 (轻型队列的实现) 通过采用“读放大”模型,结合 RocketMQ 的轻型队列特性和百万队列的底层技术支持,我们不仅有效解决了物联网环境下消息存储与分发的挑战,还实现了存储成本与系统性能的双重优化。这种设计不仅减少了存储空间的占用,还通过高度优化的索引机制加快了消息检索速度,为大规模物联网设备的消息通信提供了一个既经济又高效的解决方案。 推送模型 (RocketMQMQTT 推送模型) 在介绍完底层队列存储模型之后,我们将重点探讨匹配查找和可靠送达的实现机制。在传统的消息队列 RocketMQ 中,经典的消费模式是消费者通过客户端直接发起长轮询请求,以精准地获取对应主题的队列消息。然而,在 MQTT 场景下,由于客户端数量众多且订阅关系复杂,长轮询模式显得不够有效,因此消费过程变得更加复杂。为此,我们采用了一种推拉结合的模型。 本模型的核心在于终端通过 MQTT 协议连接至代理节点,消息可以来源于多种场景(如MQ、AMQP、MQTT)。当消息存入主题队列后,通知逻辑模块将实时监测到新消息的到达,进而生成消息事件(即消息的主题名称),并推送至网关节点。网关节点根据连接终端的订阅状态进行内部匹配,识别能够接收这一消息的终端,随后触发拉取请求,以从存储层读取消息并推送至终端。 在这个流程中,一个关键问题是通知模块如何确定终端感兴趣的消息,以及哪些网关节点会对此类消息感兴趣。这实际上是一个核心的匹配搜索问题。常见的解决方案主要有两种:第一种是简单的事件广播,第二种是将线上订阅关系集中存储(例如图中的 Lookup 模块),然后进行匹配搜索,再执行精准推送。 虽然事件广播机制在扩展性上存在一定问题,但其性能表现仍然良好,因为我们推送的数据量相对较小,仅为 Topic 名称。此外,同一 Topic 的消息事件可以合并为一个事件,这是我们当前在生产环境中默认采用的方式。另一方面,将线上订阅关系集中存储在 RDS 或 Redis 中也是一种普遍的做法,但这需要保证数据的实性,匹配搜索的过程可能会对整体实时消息链的延迟产生影响。 在该模型中,还设计了一个缓存模块,以便在需要广播大量消息时,避免各个终端对存储层发起重复的读取请求,从而提高整体系统的效率。 阿里云 MQTT 在 Serverless 上的实践 随着云原生技术的不断发展,现代消息中间件逐渐以容器编排为基础,如何实现真正的无服务器架构及秒级弹性管理已成为一项重要的研究课题。 阿里云作为开源 RocketMQMQTT 的主要贡献者和使用者之一,在 MQTT 弹性设计上有很多优化方式和实践经验。我们将介绍阿里云在弹性上的设计思路,展示其如何实现高效、弹性强的 MQTT 消息中间件: 1)抽离网络连接层 阿里云 MQTT 采用类似 Sidecar 的模式,将网络连接层与核心业务逻辑进行分离,使用 Rust 语言来处理网络连接。与 Java 相比,Rust 在内存消耗和启动速度上具有显著优势,尤其在处理大规模 MQTT 连接时,能够有效降低内存占用。 2)秒级扩容 每个 Pod 的资源请求设置较低,同时预留部分 Pod 专门运行 Rust 进程。在扩容需求出现时,系统能够快速启动 MQTT Proxy 进程,省去 Pod 创建和资源挂载的时间,从而显著提升响应速度。 3)弹性预测与监控 利用连接数、TPS、内存、CPU 等白盒指标,以及 RT 等黑盒指标,阿里云 MQTT 依据指标联动规则,制定了合理的扩容策略。这使得系统能够提前预测负载变化并启动 Pod 扩容,确保长期平稳运行。 通过以上设计思路,阿里云能够构建一个高效、弹性强的基于 RocketMQ 的 MQTT 实现方案,充分利用 Rust 带来的性能优势,同时保持系统的稳定性与可扩展性。这种创新设计将在实际应用中显著提升用户体验,助力系统整体性能的优化。 阿里云 MQTT 在车联网中的实践架构 随着汽车出行领域新四化(电气化、智能化、网联化和共享化)的推进,各大汽车制造商正逐步构建以智能驾驶和智能网联为核心的车联网系统。这一新一代车联网系统对底层消息采集、传输和处理的平台架构提出了更高的要求。接下来,我们将介绍阿里云 MQTT 在车联网中的实践架构及其应用价值。 在架构图中,我们可以看到常见的车联网设备,包括车载终端、路测单元和手机端系统。这些设备确保了安全的连接与数据传输。车端的功能涵盖车机数据上报、POI 下发、文件推送、配置下发、消息推送等全新车联业务。这些操作将产生海量的消息 Topic,需要更加安全、稳定的接入与传输,以实现可靠的消息订阅与发布。路端则强调路侧 RSU 的安全接入,支持消息的采集、传输以及地图数据的实时更新。 接入端支持多种协议,包括 TCP、x509、TLS、WSS、WS、OpenAPI 和 AMQP,以满足不同应用场景的灵活需求。这种多协议支持确保了设备之间的无缝互联与高效通信。 在流转生态方面,物联网场景下,各种设备持续产生大量数据,业务方需要对这些数据进行深入分析与处理。采用 RocketMQ 作为存储层,系统能够只保存一份消息,并支持物联网设备和云端应用的共同消费。RocketMQ 的流存储特性使得流计算引擎能够无缝、实时地分析物联网数据,为关键决策提供及时支持。 借助阿里云 EventBridge,MQTT 物联网设备所生成的信息可以顺利流转至 Kafka、AMQP、FC、Flink 等其他中间件或数据处理平台,实现深度的数据分析与处理。事件总线 EventBridge 是阿里云提供的一款 Serverless 总线服务,支持阿里云服务、自定义应用、SaaS 应用以标准化、中心化的方式接入,并能够以标准化的 CloudEvents 1.0 协议在这些应用之间路由事件,帮助轻松构建松耦合、分布式的事件驱动架构。这种灵活的数据流转能力不仅提升了处理速度,还为未来智能化应用的创新和发展奠定了基础。 通过以上架构,可以清晰地看到阿里云 MQTT 在车联网领域的最佳实践,为实现未来智能出行提供了可靠的技术支撑。 结语 在物联网的蓬勃发展背景下,消息传递技术的不断演进已成为支撑智能家居、工业互联网以及车联网等领域的重要基石。通过对 RocketMQ 和 MQTT 协议的深度融合,我们不仅有效解决了物联网时代对高效、可靠消息传输的需求,也为设备通信带来了灵活的解决方案。 阿里云在这一领域的积极探索,通过引入 Serverless 架构,不断推进 MQTT 的技术迭代与创新。这样的设计能够在面对高并发连接和海量数据时,动态调整资源配置,降低成本并提升响应速度,确保了实时数据处理的高效性。 当前,社区正在推动 MQTT 5.0 协议方面已取得显著进展,新的协议特性如更丰富的错误码、更灵活的连接选项以及 will 消息、retain 消息、共享订阅功能都将进一步提升系统的灵活性和可靠性。与此同时,我们在致力于实现更快的弹性扩展能力,以便在面对突发流量时及时响应,提高系统的可用性和灵活性。 随着 IoT 生态体系的不断完善,面对日益复杂的消息场景,消息技术的价值愈发凸显。我们相信,未来通过不断优化的消息架构,能够推动更深层次的智能化应用,同时为构建万物互联的未来奠定坚实的基础。让我们期待 MQTT 在物联网技术的场景中展现其无限可能,同时也继续探索持续探索在确保安全、稳定、高效的消息中间件。

2024年8月22日

一文详解 RocketMQ 如何利用 Raft 进行高可用保障
作者|季俊涛 前言 Apache RocketMQ 自诞生以来,因其架构简单、业务功能丰富、具备极强可扩展性等特点被众多企业开发者以及云厂商广泛采用。历经十余年的大规模场景打磨,RocketMQ 已经成为业内共识的金融级可靠业务消息首选方案,被广泛应用于互联网、大数据、移动互联网、物联网等领域的业务场景。由于其业务场景愈加丰富,在工业界的使用率日益提高,开发者们也必须更完善地考虑 RocketMQ 的可靠性、可用性。 由于 RocketMQ 底层实际上是一种基于日志的存储系统,而前人为了避免这种存储系统中单个机器可能出现的数据丢失、单点故障等问题,已经有了相对成熟的解决方案——例如同时复制数据到多个机器上。在这个过程中,需要解决的问题便被简化了:如何保证多个机器上的数据是一致的,而且这种一致性强大到可以对抗宕机、脑裂等问题。而这些问题,可以通过分布式一致性算法来彻底解决。 在开源的 Apache RocketMQ 中,我们已经引入了 DLedger [2] 和 SOFAJRaft [3] 来作为 Raft [4] 算法的具体实现,以支撑系统高可用。本文将介绍 RocketMQ 如何利用Raft(一种简单有效的分布式一致性算法)进行高可用的保障。 分布式一致性算法:Raft 共识(Consensus)是分布式系统中实现容错的一个基本问题。它指的是在一个系统中,多个服务器需要就某些值达成一致的意见。一旦它们对一个值做出了决定,这个决定就是不可变更的 [1]。典型的共识算法能够在多数服务器可用的情况下继续运行;比如说,在一个有 5 台服务器的集群中,即使有 2 台服务器宕机,整个集群依然能够正常运作。如果宕机的服务器数量超过半数,集群就无法继续正常运行了(但它也绝不会返回错误的结果)。 业界比较有名的分布式一致性算法是 paxos [12],不过可惜的是它比较晦涩难懂,难懂的代价就是很少有人能掌握它然后基于它做出可靠的实现。不过幸好 Raft 及时出现,它易于理解,并且已经有非常多的业界使用先例,比如 tikv、etcd 等。 这是 Raft 的原始论文,详细描述了 Raft:《In Search of an Understandingable Consensus Algorithm》 [4]。本文的略短版本在 2014 年 USENIX 年度技术会议上获得了最佳论文奖。有意思的是,在论文的 Abstract 中,第一句话便是:Raft is a consensus algorithm for managing a replicated log. 在英文中,log 不仅有“日志”的意思,还有“木头”的意思,一组"replicated log"便组成了木筏,这和 Raft 的英文原意不谋而合。而且,Raft 的主页中,也采用了三根木头组成的筏作为 logo(如图 1)。当然,对 Raft 更正式的解释还是这些单词的首字母缩写:Re{liable | plicated | dundant} And FaultTolerant. 也就是 Raft 被提出时要解决的问题初衷。 (图 1:Raft 主页标题后附的头图) Raft 算法是一种为了管理复制日志的共识算法,它将整个共识过程分解为几个子问题:领导者选举、日志复制和安全性。整个 Raft 算法因此变得易理解、易论证、易实现,从而让分布式一致性协议可以较为简单地实现。Raft 和 Paxos 一样,只要保证 n/2+1 节点即多数派节点正常就可以对外提供服务。 在 Raft 算法中,集群中的每个节点(服务器)可以处于以下三种状态之一: a. Follower(跟随者):这是所有节点的初始状态。它们被动地响应来自领导者和候选者的请求。 b. Candidate(候选者):当跟随者在一段时间内没有收到来自领导者的消息时,它们会成为候选者,并开始选举过程以成为新的领导者。 c. Leader(领导者):集群中的管理节点,负责处理所有客户端请求,并将日志条目复制到其他节点。 Raft 算法通过任期的概念来分隔时间,每个任期开始都会进行一次领导者选举。任期是一个递增的数字,每次选举都会增加。如果跟随者在“选举超时”之内没有收到领导者的心跳,它会将自己的任期号加一,并转变为候选者状态来发起一次领导者选举。候选者首先给自己投票,并向其他节点发送请求投票的 RPC。如果接收节点在当前任期内还没有投票,它会同意投票给请求者。 如果候选者在一次选举中从集群的大多数节点获得了选票,它就会成为领导者。在此过程中生成的任期号用于节点之间的通信,以防止过时的信息导致错误。例如,如果节点收到任期号比自己小的请求,它会拒绝该请求。 极端情况下集群可能会出现脑裂或网络问题,此时集群可能会被分割成几个互不通信的子集。不过由于 Raft 算法要求一个领导者必须拥有集群大多数节点的支持,这保证了即使在脑裂的情况下,最多只有一个子集能够选出一个有效的领导者。 Raft 通过日志复制来保持节点间的一致性。对于一个无限增长的序列 a[1, 2, 3…],如果对于任意整数 i,a[i] 的值满足分布式一致性,这个系统就满足一致性状态机的要求 基本上所有的真实系统都会有源源不断的操作,这时候单独对某个特定的值达成一致显然是不够的。为了让真实系统保证所有的副本的一致性,通常会把操作转化为 writeaheadlog(WAL)。然后让系统中所有副本对 WAL 保持一致,这样每个副本按照顺序执行 WAL 里的操作,就能保证最终的状态是一致的,如图 2 所示。 (图 2:如何通过日志复制来确保节点间数据一致。阶段一为 Client 向 leader 发送写请求,阶段二为 Leader 把‘操作’转化为 WAL 并复制,阶段三为 Leader 收到多数派应答,并将操作应用到状态机) 领导者在收到客户端的请求后,会先将请求作为新的日志条目追加到它的日志中,然后并行地将该条目复制到其他节点。只有当大多数节点都写入了这个日志条目,领导者才会将该操作提交,并应用到它的状态机上,同时通知其他节点也提交这个日志条目。因此,Raft 确保了即使在领导者崩溃或网络分区的情况下,也不会有数据丢失。任何被提交的日志条目都保证在后续的任期中也存在于任意新的领导者的日志中。 简单来说,Raft 算法的特点就是 Strong Leader: a. 系统中必须存在且同一时刻只能有一个 Leader,只有 Leader 可以接受 Clients 发过来的请求; b. Leader 负责主动与所有 Followers 通信,负责将“提案”发送给所有 Followers,同时收集多数派的 Followers 应答; c. Leader 还需向所有 Followers 主动发送心跳维持领导地位(保持存在感)。 一句话总结 Strong Leader: "你们不要 BB! 按我说的做,做完了向我汇报!"。另外,身为 Leader 必须保持一直 BB(heartbeat)的状态,否则就会有别人跳出来想要 BB 。 为了更直观的感受到 Raft 算法的运行原理,笔者强烈推荐观看下面网站中的演示。它以动画的形式,非常直观地展示了 Raft 算法是如何运行的,以及如何应对脑裂等问题的:_https://thesecretlivesofdata.com/raft/_ (图 3:Raft 算法选举过程图示) 相信观看过上面网站中的图解后,读者应该了解了 Raft 的设计思想与具体算法,下面我们直接切入正题,讲解 RocketMQ 与 Raft 的前世今生。 RocketMQ 与 Raft 的前世今生 RocketMQ 尝试融合 Raft 算法已经非常之久,这期间的融合方式也经历过变革。发展至今,Raft 也只是 RocketMQ 高可用机制中的一小部分,RocketMQ 已然发展出了一套适合自身的高可用共识协议。 本章主要阐述 RocketMQ 为了在系统内实现 Raft 算法作出过哪些尝试,以及当前 Raft 在 RocketMQ 中的存在形态与具体作用。 Raft 在 RocketMQ 中的初期形态 RocketMQ 引入 Raft 协议的主要原因是为了增强系统的高可用性和故障自动恢复能力。在 4.5 版本之前,RocketMQ 只有 Master/Slave 一种部署方式,即一组 broker 中仅有一个 Master,有零到多个 Slave,这些 Slave 以同步或者异步的方式去复制 Master 中的数据。然而这种方式存在一些限制: 1. 故障转移不是完全自动的: 当 Master 节点出现故障时,需要人工介入进行手动重启或者切换到 Slave 节点。 2. 对外部依赖较高: 虽然可以通过第三方协调服务(如 ZooKeeper 或 etcd)实现自动选主,但这增加了部署和运维的复杂性,同时第三方服务本身的故障也可能影响到 RocketMQ 的集群。 为了解决上述问题,RocketMQ 引入了基于 Raft 协议的 DLedger 存储库 [5]。DLedger 是一个分布式日志复制技术,它使用 Raft 协议,可以在多个副本之间安全地复制和同步数据。RocketMQ 4.5 版本发布后,可以采用 RocketMQ on DLedger 方式进行部署。DLedger commitlog 代替了原来的 CommitLog,使得 CommitLog 拥有了选举复制能力,然后通过角色透传的方式,raft 角色透传给外部 broker 角色,leader 对应原来的 master,follower 和 candidate 对应原来的 slave: (图 4:RocketMQ on DLedger 部署形态,每个 broker 间的角色由 Raft CommitLog 向外透传) 因此 RocketMQ 的 broker 拥有了自动故障转移的能力。在一组 broker 中, Master 挂了以后,依靠 DLedger 自动选主能力,会重新选出 leader,然后通过角色透传变成新的 Master。DLedger 还可以构建高可用的嵌入式 KV 存储。我们把对一些数据的操作记录到 DLedger 中,然后根据数据量或者实际需求,恢复到hashmap 或者 rocksdb 中,从而构建一致的、高可用的 KV 存储系统,应用到元信息管理等场景。 我们测试了各种故障下 Dledger 表现情况,包括随机对称分区,随机杀死节点,随机暂停一些节点的进程模拟慢节点的状况,以及 bridge、partitionmajoritiesring 等复杂的非对称网络分区。在这些故障下,DLedger 都保证了一致性,验证了 DLedger 有很好可靠性 [6]。 总结来说,引入 Raft 协议后,RocketMQ 的多副本架构得以增强,提高了系统的可靠性和自我恢复能力,同时也简化了整个系统的架构,降低了运维的复杂性。这种架构通过 Master 故障后短时间内重新选出新的 Master 来解决单主问题,但是由于 Raft 选主和复制能力一同在数据链路(CommitLog)上,因此存在以下问题: 1. Broker 组内的副本数必须是 3 副本及以上才有切换能力,因此部署的最低成本是有上升的。 2. Raft 多数派限制导致三副本副本必须两副本响应才能返回,五副本需要三副本才能返回,因此 ACK 是不够灵活的,这也导致“发送延迟低”和“副本冗余小”两种要求很难做到。 3. 由于存储复制链路用的是 OpenMessaging DLedger 库,导致 RocketMQ 原生的一些存储能力没办法利用,包括像 TransientPool、零拷贝的能力,如果要在 Raft 模式下使用的话,就需要移植一遍到 DLedger 库,开发特性以及 bug 修复也需要做两次,这样的维护和开发成本是非常高的。 此外,将选举逻辑嵌入数据链路中可能会引发一连串的问题,这直接与我们追求的高可用和稳定性目标背道而驰。以选举发生在数据链路中的假设情景为起点,我们可以设想一个由多个节点构成的存储集群,其中节点需要定期进行选举来决定谁负责数据流的管理任务。在这种设计下,选举不仅是控制面的一部分,而且直接影响数据链路的稳定性。一旦发生选举失败或不一致的情况,整个数据链路可能会受阻,导致写入能力的丧失或数据丢失。 但是我们将目光看向 PolarStore 时就能发现,它的设计思想包含了“控制面和数据面分离”:数据面操作仅依赖于本地缓存的全量元数据,而对控制面的依赖最小化。这种设计的优势在于即使控制面完全不可用,数据面依然能够依据本地缓存维持正常的读写操作。在这种情况下,控制面的选举机制永远不会影响到数据面的可用性。这种分离架构为存储系统带来了相当强的鲁棒性,即使在遭遇故障时也能够保持业务的连续性。 总而言之,将选举逻辑与数据链路解耦是保障存储系统高可用性和稳定性的关键。通过将控制面的复杂性和潜在故障隔离,可以确保即使在面临控制面故障时,数据面依然能够保持其核心功能,从而为用户提供持续的服务。这种健壮的设计理念在现代分布式存储系统中是至关重要的——在控制面遭遇问题时,数据面能够以最小的影响继续运作。 这个例子告诉我们,数据面的可用性如果和控制面解耦,那么控制面挂掉对数据面的影响很轻微。否则,要么要不断去提高控制面的可用性,要么就要接受故障的级联发生 [7]。这也是我们后文中 RocketMQ 的演进方向。 现在的 Raft Controller:控制面/数据面分离 上文中提到,我们以 DLedger 的形式将 Raft 引入了 RocketMQ,但这种引入实际上是给了 CommitLog 选举的能力。这种设计固然直接有效,能够直接赋予一致性给最重要的组件。但是这样的设计也让选举、复制的过程被耦合到了一起。当二者耦合时,每次的选举、拷贝便都强依赖 DLedger 的 Raft 实现,这对于未来的扩展性是非常不友好的。 那么,有没有更可靠、更灵活的解决方案呢?我们不妨把目光转向学术界,看看他们的灵感。在 OSDI' 20 上,Meta 公司管控平面元数据存储统一平台 Delos [8] 相关论文获得了的 Best Paper Award。这篇论文提供了一个全新的视角与思路来解决选主过程中“控制面与数据面耦合”的问题:虚拟共识(Virtual Consensus)。它在论文中描述了在生产环境中实现在线切换共识协议的工作。这种设计旨在通过虚拟化共享日志 API 来实现虚拟化共识,从而允许服务在不停机的情况下更改共识协议。 论文的出发点是,在生产环境中,系统往往高度集成了共识机制,因此更换共识协议会涉及到非常复杂且深入的系统改动。 以前文提到的 DLedger CommitLog 为例,其分布式共识协议中的数据流(负责容错)和控制流(负责同步共识组配置)是密切相关的。在这种紧密耦合的系统中进行修改是极其困难的,这就使得开发和实施新共识协议的代价变得相当昂贵,甚至单纯的更新微小特性都面临重大挑战。更不用提在这种环境下引入更先进的共识算法(未来如果有的话),这必然花费相当重大的成本。 相比之下,Delos 提出了一种新方法,通过分离控制层和数据层来克服这些挑战。在这个架构中,控制层负责领导选举,而数据层则由 VirtualLog 和下层的 Loglets 组成,用于管理数据。VirtualLog 提供了一个共享日志的抽象层,将不同 Loglets(代表不同共识算法实例)串联起来。这种抽象层在 VirtualLog 和各个 Loglet 之间建立日志条目的映射关系。要切换到新的共识协议时,简单地指示 VirtualLog 将新的日志条目交由新 Loglet 处理以达成共识即可。这种设计实现了共识协议的无缝切换,并显著降低了更换共识协议的复杂性和成本。 (图 5:Delos 设计架构示意图,控制面和数据面被分开,以 VirtualLog 的形式进行协作) 这种设计既然已经在学术界开诚布公,那 RocketMQ 也可以在工业界将其落地并作验证。在 RocketMQ 5.0 中,我们提出了 Raft Controller 的概念:仅在上层协议中使用 Raft 与其它选主算法,而下层数据链路的复制则由 Broker 中的一套数据复制算法负责,用于响应上层的选主结果。 这个设计理念在行业中其实并不罕见,反而已经成为了一种成熟且广泛被验证的实践。以 Apache Kafka 为例,这个高性能的消息传递系统采用了分层的架构策略,在早期版本中使用了 ZooKeeper 来构建其元数据的控制平面,这一控制平面负责管理集群的状态和元数据信息。随着 Kafka 的发展,新版本引入了自研的 KRaft 模式,进一步内部化并提升了元数据管理。此外,Kafka 的 ISR(InSync Replicas) [15] 协议承担了数据传输的重任,提供了高吞吐量和可配置的复制机制,确保了数据平面的灵活性和可靠性。 同样地,Apache BookKeeper,一个低延迟且高吞吐的存储服务,也采用了类似的架构思想。它利用 ZooKeeper 来管理控制平面,包括维护一致的服务状态和元数据信息。在数据平面方面,BookKeeper 利用其自研的 Quorum 协议来处理写操作,并确保读操作的低延迟性能。 类似这种设计,我们的 Broker 不再自己负责自己的选举,而是由上层 Controller 对下层的角色进行指示,下层根据指示结果进行数据的拷贝,从而达到选举与复制分离的目的。 (图 6:来源于 ATA 文章《全新 RocketMQ 5.0 高可用设计解读》) 不过与 Delos 不同的是,我们在这里面其实有三层共识 —— Controller 间的共识协议(Raft),Controller 对 Broker 选主时的共识协议(SyncState Set),Broker 间复制时的共识协议(主备确认复制算法)。我们在这个过程中额外增加了 Controller 间的共识,以保证控制节点也是强一致的。这三种共识算法具体实现在这里不加以赘述,有兴趣的可以看我们 RocketMQ 社区中的 RIP31/32/34/44 几个说明文档。 这个设计也在我们被 ASE 23' 录用的论文 [9] 中得以体现:Controller 组件承担了切换链路中的核心角色,但是又不影响数据链路的正常运行,即便其面临宕机、夯机、网络分区等问题,也不会导致 broker 的数据丢失、不一致。我们在后续测试中甚至模拟了更加严苛的场景,例如 Controller 与 Broker 同时宕机、同时夯机、同时进行随机网络分区等等,我们的设计均有非常好的表现。 下面,我们对 RocketMQ 中的共识协议作展开,深入地剖析 RocketMQ 中的共识是如何实现的。 RocketMQ 中的共识协议 首先我们在这里放一张大图,用于描述 Raft Controller 具体是如何实现控制面、数据面的共识的。 (图 7:Raft Controller 具体设计架构及其运作原理) 上图中,绿色部分的是 Controller,也就是 RocketMQ 划分出来的控制面。它自身包含了两种共识算法,分别是保证 Controller 自身共识的 Raft 算法。以及保证 Broker 共识的选主算法,SyncState Set(后文简称 3S)算法,这个算法是我们参考 PacificA 算法 [10] 提出的一套用于数据面选主的分布式共识协议。红色部分是 Broker,也就是我们最核心的数据面,这里面忽略了 Broker 的其它存储结构,仅保留复制过程中实现共识的核心文件:epoch 文件。我们基于它实现了数据复制过程中的共识协议。 下面我们针对控制面和数据面的共识,分别进行阐述。 控制面共识 控制面共识共有两层: a. Raft Controller 自身的共识——用于保证 Controller 间的数据强一致性 b. 3S 算法——用于保证 Broker 选主结果的强一致。 下面对这两种共识算法分别进行解释。 Raft In Controller Raft 在 RocketMQ 5.0 前,只在 CommitLog 中存在,以 DLedger CommitLog 的方式向外透传角色。但是经过我们前文的分析,可以知道这种方式是有不容忽视的弊端在的:选举、复制过程强耦合,复制过程强依赖 DLedger 存储库,迭代难度高。因此,我们考虑将 Raft 的能力向上移动,让其用于在控制面中实现原数据强一致。这样一来,选举的过程便与日志复制的过程区分开了,而且每次选举的成本相对更低,只需要同步非常有限的数据量。 在 Controller 中,我们将 Raft 算法用于选举 Controller 中的 Active Controller,由它来负责处理数据面的选举、同步等任务,其余的 Controller 则只负责同步 Active Controller 的处理结果。这样的设计能够保证 Controller 本身也是高可用的,且保证了仅有一个 Controller 在处理 Broker 的选举事务。 在最新的 RocketMQ 中,在这一层共提供了两种 Controller 内的分布式共识算法的实现:DLedger 与 JRaft。这两种共识算法可以在 Controller 的配置文件中被非常简单地选择。本质上来说,这两者都是 Raft 算法的具体实现,只不过具体的实现方式有些差异: a. DLedger [2] 是一个基于 raft 协议的 commitlog 存储库,是一个 append only 的日志系统,早期针对 RocketMQ 的诸多场景有过相当多的适配。同时,它是一个轻量级的 java library。它对外提供的 API 非常简单,append 和 get。 b. JRaft 全称为 SOFAJRaft [3],它是一个基于 Raft 一致性算法的生产级高性能 Java 实现,支持 MULTIRAFTGROUP,适用于高负载低延迟的场景。SOFAJRaft 是从百度的 braft [11] 移植而来的,做了一些优化和改进。 这两种 Raft 的具体实现都对外提供了非常简单的 API 接口,所以我们可以把更多的精力放在处理 Active Controller 的事务上。 3S Algorithm 抛开 Controller 本身的共识算法,我们将目光聚焦于 Active Controller 在整个过程中起的作用,这也是我们控制面共识的核心——3S 算法。 3S 算法中的SyncState Set 概念与 Kafka 的 InSync Replica(ISR) [17] 机制类似,都参考了微软的 PacificA 算法。与以分区为维度的 ISR 不同,3S 算法以整个 Broker 的维度发起选举,且针对 RocketMQ 的需要选举场景作了系统的归纳。相比较来说,3S 算法更加简单,选举更加高效,面对大量分区场景能有更加强大的表现。 3S 算法主要作用在 Controller与 Broker 的交互过程中,Active Controller 会处理每个 Broker 的心跳与选举工作。和 Raft 类似的,3S 算法也有心跳机制来实现类似租约的功效——当 Master Broker 一定时长没有上报心跳,就可能触发重新选举。不过与 Raft 不同的是,3S 算法有一个共识处理的核心:Controller。这种中心化的设计能够让数据面的选主更加简单,达成共识更加迅速。 在这种设计下,Broker 的心跳不再向同级别(数据面)发送,而是统一向上(控制面)发送。发送选举请求后,由 Controller 来决定哪个 Broker 可以作为 Master 存在,而其它 Broker 自然退化为 Slave。Controller 的选择原则可以是多样的(同步进度、节点资源等指标),也可以是简单有效的(数据同步进度达到一定阈值),只需要这个节点位于 SyncState Set 中。这也是一种 Strong Leader 的形式,只不过和 Raft 不同的地方在于: Raft 像是小组作业,同学们(Broker)互相投票进行小组长的票选,而 3S 算法则由班主任(Controller)根据举手快慢直接任命。 (图 8:多个 Broker 集群向 Active Controller 汇报集群内的主备角色以及同步情况) 如上图所示,三个 Broker 集群中的 Leader 都会定期向 Active Controller 上报集群的同步状态: a. A 集群的所有节点的同步进度都很良好,因此 Leader 上报的 SyncState Set 是所有节点。 b. B 集群的 Follower2 可能刚刚启动,仍旧在同步历史消息,因此 SyncState Set 并不会包含它——当 Leader 宕机时,Controller 自然也不会选择它。 c. C 集群中,虽然 1 号 Leader 已经宕机,但是 Controller 迅速便能决定 SyncState Set 中的 3 号节点作为替代,提拔为主节点,整个集群便能正常运转,此时,即便 3 号节点又宕机,也能选择 2 号节点为主节点,不影响集群运行状态。 这种设计的好处在哪里呢?Raft 算法的实现原理其实是“投票”,同学间彼此平等,靠投票结果“少数服从多数”。因此,对于一个有 2n+1 节点的集群来说,Raft 最多只能容忍n个节点失效,至少需要保证有 n+1 个节点是持续运行的。但是 3S 算法有一个选举中心,每次选举的 RPC 都向上发送,它不需要得到其它节点的认可便可选举出一个节点。因此对于之前提到的 2n+1 节点的集群来说,最多能容忍 2n 个节点的失效,即副本的数量不需要超过副本总数的一半,不需要满足 “多数派” 原则。通常,副本数大于等于 2 即可, 如此,便在可靠性和吞吐量方面取得平衡。 上面的例子比较简明扼要地介绍了 3S 算法和 Raft 的关系与不同,可以说 3S 算法的设计思想来源于 Raft,但是在特定场景下又优于 Raft。 此外,3S 算法的共识以整个 Broker 为维度,因此我们对选主时机作了优化,有如下几种情形可能触发选举,括号内的红色是更加形象的描述,将选主过程具像化为选小组长的过程,以便理解: a. 控制面,Controller 主动发起 (班主任发起): i.HeartbeatManager 监听到有 broker 心跳失效。(班主任发现有小组同学退学了) ii.Controller 检测到有一组 SyncState Set 不存在 master。(班主任发现有组长虽然在名册里,但是旷课了) b. 数据面,Broker 发起将自己选为 master (同学毛遂自荐): i. Broker 向 controller 查询元数据时,没找到master信息。(同学定期检查小组情况,问班主任为啥没小组长) ii. Broker 向 controller 注册完后,仍未从 controller 获取到 master 信息。(同学报道后发现没小组长,汇报并自荐) c. 运维侧,通过 RocketMQ Admin Tools 发起,是运维能力的一部分 (校长直接任命)。 通过上述两方面优化,3S 算法在 RocketMQ 5.0 中,展露了非常强大的功能性,让 Controller 成为了高可用设计范式中不可或缺的组件。其实 3S 算法在实际使用场景中还有很多细节上的处理优化,能够容忍前文提到的更加严峻的场景:如控制面和数据面同时发生故障,且故障节点超过一半以上的场景。这部分结果会在后文的混沌实验中得以展示。 数据面共识 控制面通过 Raft 算法保障了 Controller 间的角色共识,以及通过 3S 算法保障了 Broker 中的角色共识。那么在 Broker 角色被确定后,其数据面该如何根据选举结果保障数据的强一致呢?这里的算法并不复杂,因此笔者从实现角度介绍一下 RocketMQ 的设计,RocketMQ 的数据面共识主要由下面两个组件构成: HAClient: 每个 Slave 的 HAService 中必备的 client,负责管理同步任务中的读、写操作。 HAConnection: 代表在 Master 中的 HA 连接,每个 connection 理论上对应一个 slave。在该 connection 类中存储了传输过程中的诸多内容,包括 channel、传输状态、当前传输位点等等信息。 为了更形象地描述清楚 RocketMQ 在这方面的设计结构,笔者绘制了下面这幅图,可以看出核心还是数据的传输过程,分别设计了一个 Reader 与一个 Writer: (图 9:数据面复制过程的具体实现,Master 与 Slave 分别设计,但选举完成后可互相切换) 这么简单的设计,是如何确保数据写入时的强一致的呢? 核心的共识其实存在于 HAConnection(也就是图中左下角那个深蓝色框)的建立、维护过程中。每个 Broker 集群的主节点都会维护和所有 Slave 的连接关系,并将其存于 Connection 表中,在每次 Slave 来请求代复制数据后,都会反馈复制的最后位点与结果,因此主节点也可以基于此来确定上报给 Controller 的 SyncState Set。在 HAConnection 的建立过程中,有一个确保数据一致性的 HandShake 阶段。这个阶段能够对 CommitLog 作截断,从而保障复制位点之前的所有数据都是强一致的。这个过程通过 epoch 文件的标记实现:Epoch 文件中包含了每一次选举的状态,每次选举完成后,主节点都会在 epoch 文件上留下自己的痕迹,即当前的选举代数 + 当前的初始位点。 从这里也可以看到,我们数据面的共识算法也有一些 Raft 的影子:Raft 算法在每次选举后也会给任期数自增一,这个任期数的大小决定了后续选主的权威性。而在数据面共识算法中,选主的结果已经认定,任期数被用于多次选主结果的共识表征——当任期数与日志位点一致时,代表这两台 broker 就选主这件事达成过一致,因此可以认为此前的数据是强一致的,只需要保证后续数据的强一致即可。为了方便理解,可以通过下图进行描述: (图 10:来源于文章《全新 RocketMQ 5.0 高可用设计解读》) 类似上面这张图,最上面那个方块长条实际是 RocketMQ 的日志存储形态 MappedFile。下面两条方块组成的长条分别是主和备的 commitlog,备节点会从后向前找到最大的 一致的位点,然后截断到这个位点,开始向后复制。这种复制在 RocketMQ 中有单独的一个 Service 去执行,因此主备节点的复制和选举过程其实是彻底解耦开的,只有当一个备节点尽可能跟上主节点时,这个备才会被纳入 SyncState Set,后续才有资格参加选举。 拥抱故障,把故障当作常态 俗话说的好,“空谈误国,实干兴邦”,设计究竟是先进还是冗余,需要通过各方面的检验。 对于 RocketMQ 这种大规模在生产中被使用的系统,我们必须模拟出足够接近现实情况的故障,才能检验其可用性在现实场景中究竟如何。在这里,我们需要引入一个新的概念——混沌工程。 混沌工程的原始定义 [13] 为:“Chaos engineering is the discipline of experimenting on a system in order to build confidence in the system's capability to withstand turbulent conditions in production. ” 从原始定义看,混沌工程实际上是一种软件工程方法,旨在通过在软件系统的生产环境中故意引入混乱来验证系统的可靠性。混沌工程的基本假设是,生产环境是复杂且不可预测的,而通过模拟各种失败,可以发现并解决潜在的问题。这种方式有助于确保系统能够在面临真实世界中的各种挑战时,持续并有效地运行。 大家都写过代码,也都深知一个精心设计过的系统总是能够巧妙通过各个测试样例,但是上线后总会遇到各种问题。因此对于一个系统来说,能够出色地通过复杂且不可预知的频繁故障的混沌工程的考验,而不是测试样例,才能说明这个系统是高可用的。 下面我们将详细介绍,我们为了验证 RocketMQ 的高可用,对其作过哪些“拷打”。 OpenChaos OpenChaos [14] 作为云原生场景量身定制的混沌“刑具”,位于 OpenMessaging 名下,托管于 Linux 基金会。目前,它支持以下平台的混沌测试:Apache RocketMQ、Apache Kafka、DLedger、Redis、Zookeeper、Etcd、Nacos。 目前 OpenChaos 支持注入多种故障类型,其中最主要的便是: 1. randompartition (fixedpartition):随机(固定)隔离节点与网络的其他部分。 2. randomloss:随机选定的节点丢失网络数据包。 3. randomkill (minorkill, majorkill, fixedkill):终止随机(少数、多数、固定)的进程并重启它们。 4. randomsuspend (minorsuspend, majorsuspend, fixedsuspend):使用 SIGSTOP/SIGCONT 暂停随机(少数、多数、固定)的节点。 在实际场景中,最常见的故障就是这四种:网络分区、丢包、宕机、夯机。此外,OpenChaos 还支持其它更复杂的特定场景,例如 Ring(每个节点能够看到大多数其他节点,但没有节点能看到与任何其他节点相同的多数节点)和 Bridge(网络一分为二,但保留中间的一个节点,该节点与两边的组件保有不间断的双向连通性),形成条件非常严苛,而且它们阻碍共识生效的原理都是“通过阻碍各节点间的可见性,来避免形成全局多数派”,理论上来说,通过足够久的网络分区、丢包,也能模拟出这些情况,甚至更复杂的情况。 因此,我们注入了大量上面罗列的四种混沌故障,观察集群是否有出现消息丢失的情况,并统计了故障恢复时间。 具体测试场景 我们混沌测试的验证实验环境如下: a. namesrv 一台,内含 namesrv 进程,openchaos 的混沌测试进程也在该机器上启动,向 controller/broker 发出控制指令。 b. controller 三台,内含 controller 进程。 c. broker 三台,同属一个集群,分别为主备,内含 broker 进程。 上述 7 台机器的配置为,处理器:8 vCPU,内存:16 GiB,规格族:ecs.c7.2xlarge,公网带宽:5Mbps,内网带宽:5/ 最高 10 Gbps。 在测试中,我们设置了如下的若干种随机测试场景,每种场景都会持续至少 60 秒,且恢复后会保证 60 秒的时间间隔再注入下一次故障: a. 机器宕机,这个混沌故障注入通过 kill 9 命令实现,将会杀死范围内的随机进程。 i. Broker 节点,随机宕机一半以上的节点,至少保留一台 Broker 工作。 ii. Controller 节点,随机宕机一半以上的节点,以及全部宕机。iii. Broker+Controller 节点,随机宕机一半以上的节点。 b. 机器夯机,这个混沌故障注入通过kill SIGSTOP 命令实现,模拟进程暂停的情况。 i. Broker 节点,随机夯机一半以上的节点,至少保留一台 Broker 工作。 ii. Controller 节点,随机夯机一半以上的节点,以及全部夯机。 iii. Broker+Controller 节点,随机夯机一半以上的节点。 c. 机器丢包,这个混沌故障注入通过 iptables 命令实现,可以指定机器间特定比例的丢包事件。 i. Broker 间随机丢包 80%。 ii. Controller 间随机丢包 80%。 iii. Broker 和 Controller 间随机丢包 80%。 d. 网络分区,这个混沌故障注入也是通过 iptables 命令实现,能够将节点间完全分区。 i. Broker 间随机网络分区。 ii. Controller 间随机网络分区。 iii. Broker 合 Controller 间随机网络分区。 实际测试场景、组别远多于上述罗列的所有故障场景,但是存在一些包含关系,例如单台 broker/controller 的启停,便不再单独罗列。此外我们均针对 Broker 的重要参数配置进行了交叉测试。测试的开关有:transientStorePoolEnable(是否使用直接内存),slaveReadEnable(备节点是否提供读消息能力)。我们还针对 Controller 的类型(DLedger/JRaft)也进行了分别的测试,每组场景的测试至少重复 5 次,每次至少持续 60 分钟。 实验结论 针对上述提出的所有场景,混沌测试的总时长至少有: 12(场景数) 2(Broker开关数) 2 (Controller类型) 5(每组测试次数) 60(单组时长) =14400 分钟。 由于设置的注入时长一分钟,恢复时长一分钟,因此至少共计注入故障 14400/2 = 7200 次(实际上注入时长、注入次数远多于上述统计值)。 在这些记录在册的测试结果中,有如下测试结论: a. RocketMQ 无消息丢失,数据在故障注入前后均保持强一致。 b. 恢复时长基本等于客户端的路由间隔时间,在路由及时的情况下,能够保证恢复 RTO 约等于 3 秒。 c. Controller 任意形式下的故障,包括宕机、夯机、网络故障等等,均不影响 Broker 的正常运转。 总结 本文总结了 RocketMQ 与 Raft 的前世今生。从最开始的忠实应用 Raft 发展 DLedger CommitLog,到如今的控制面/数据面分离,并分别基于 Raft 协议作专属于 RocketMQ 的演化。如今 RocketMQ 的高可用已经逐渐趋于成熟:基于三层共识协议,分别实现 Controller 间、Controller&Broker、Broker 间的共识。 在这种设计下,RocketMQ 的角色、数据共识被妥善地划分到了多个层次间,并能够彼此有序地协作。当选主与复制不再耦合,我们便能更好地腾出手脚发展各个层次间的共识协议——例如,当出现比 Raft 更加优秀的共识算法时,我们可以直接将其应用于 Controller 中,且对于我们的数据面无任何影响。 可以说 Raft 的设计给 RocketMQ 的高可用注入了非常多的养分,能够让 RocketMQ 在其基础上吸纳其设计思想,并作出适合自己的改进。RocketMQ 的共识算法与高可用设计 [9] 在 2023 年也得到了学术界的认可,被 CCFA 类学术会议 ASE 23' 录用。期待在未来能够出现更加优秀的共识算法,能够在 RocketMQ 的实际场景中被适配、发扬。 参考链接: _ _ _ [4] Diego Ongaro and John Ousterhout. 2014. In search of an understandable consensus algorithm. In Proceedings of the 2014 USENIX conference on USENIX Annual Technical Conference (USENIX ATC'14). USENIX Association, USA, 305–320. _ _ [8] Mahesh Balakrishnan, Jason Flinn, Chen Shen, Mihir Dharamshi, Ahmed Jafri, Xiao Shi, Santosh Ghosh, Hazem Hassan, Aaryaman Sagar, Rhed Shi, Jingming Liu, Filip Gruszczynski, Xianan Zhang, Huy Hoang, Ahmed Yossef, Francois Richard, and Yee Jiun Song. 2020. Virtual consensus in delos. In Proceedings of the 14th USENIX Conference on Operating Systems Design and Implementation (OSDI'20). USENIX Association, USA, Article 35, 617–632. [9] Juntao Ji, Rongtong Jin, Yubao Fu, Yinyou Gu, TsungHan Tsai, and Qingshan Lin. 2023. RocketHA: A High Availability Design Paradigm for Distributed LogBased Storage System. In Proceedings of the 38th IEEE/ACM International Conference on Automated Software Engineering (ASE 2023), Luxembourg, September 1115, 2023, pp. 18191824. IEEE. _ _ [12] Leslie Lamport. 2001. Paxos Made Simple. ACM SIGACT News (Distributed Computing Column) 32, 4 (Whole Number 121, December 2001), 5158. _ _ _

2024年8月13日

谈谈 RocketMQ 5.0 分级存储背后一些有挑战的技术优化
作者|斜阳 RocketMQ 5.0 提出了分级存储的新方案,经过数个版本的深度打磨,RocketMQ 的分级存储日渐成熟,并成为降低存储成本的重要特性之一。事实上,几乎所有涉及到存储的产品都会尝试转冷降本,如何针对消息队列的业务场景去做一些有挑战的技术优化, 是非常有意思的事。 这篇文章就跟大家探讨下,在消息系统这样一个数据密集型应用的模型下,技术架构选型的分析与权衡,以及分级存储实现与未来演进,让云计算的资源红利真正传达给用户。 1. 背景与需求 RocketMQ 诞生于 2012 年,存储节点采用 sharednothing 的架构读写自己的本地磁盘,单节点上不同 topic 的消息数据会顺序追加写 CommitLog 再异步构建多种索引,这种架构的高水平扩展能力和易维护性带来了非常强的竞争力。 随着存储技术的发展和各种百G网络的普及,RocketMQ 存储层的瓶颈逐渐显现,一方面是数据量的膨胀远快于单体硬件,另一方面存储介质速度和单位容量价格始终存在矛盾。在云原生和 Serverless 的技术趋势下,只有通过技术架构的演进才能彻底解决单机磁盘存储空间上限的问题,同时带来更灵活的弹性与成本的下降,做到 “鱼与熊掌兼得”。 在设计分级存储时,希望能在以下方面做出一些技术优势: 实时: RocketMQ 在消息场景下往往是一写多读的,热数据会被缓存在内存中,如果能做到 “准实时” 而非选用基于时间或容量的淘汰算法将数据转储,可以减小数据复制的开销,利于缩短故障恢复的 RTO。读取时产生冷读请求被重定向,数据取回不需要“解冻时间”,且流量会被严格限制以防止对热数据写入的影响。 弹性: sharednothing 架构虽然简单,缩容或替换节点的场景下待下线节点的数据无法被其他节点读取,节点需要保持相当长时间只读时间,待消费者消费完全部数据,或者执行复杂的迁移流程才能缩容,这种 “扩容很快,缩容很慢” 的形态一点都不云原生,更长久的消息保存能力也会放大这个问题。分级存储设计如果能通过 shareddisk (共享存储) 的方式让在线节点实现代理读取下线节点的数据,既能节约成本也能简化运维。 差异化: 廉价介质随机读写能力较差,类 LSM 的结构都需要大量的 compation 来压缩回收空间。在满足针对不同 topic 设置不同的生命周期(消息保留时间,TTL)等业务需求的前提下,结合消息系统数据不可变和有序的特点,RocketMQ 自身需要尽量少的做格式 “规整” 来避免反复合并的写放大,节约计算资源。 竞争力: 分级存储还应考虑归档压缩,数据导出,列式存储和交互式查询分析能力等高阶技术演进。 2. 技术架构选型 2.1. 不同视角 不妨让我们站在一个新的视角看问题,消息系统对用户暴露的是收发消息,位点管理等一系列的 API,为用户提供了一种能够优雅处理动态数据流的方式,从这个角度说:消息系统拓宽了存储系统的边界。 其实服务端应用大多数是更底层 SQL,POSIX API 的封装,而封装的目的在于简化复杂度的同时,又实现了信息隐藏。 消息系统本身关注的是高可用,高吞吐和低成本,想尽量少的关心存储介质的选择和存储自身的系统升级,分片策略,迁移备份,进一步冷热分层压缩等问题,减少存储层的长期维护成本。 一个高效的、实现良好的存储层应该对不同存储后端有广泛的支持能力,消息系统的存储后端可以是本地磁盘,可以是各类数据库,也可以是分布式文件系统,对象存储,他们是可以轻松扩展的。 2.2. 存储后端调研 幸运的是,几乎所有的“分布式文件系统”或者“对象存储”都提供了“对象一旦上传或复制成功,即可立即读取”的强一致语义,就像 CAP 理论中的描述 “Every read receives the most recent write or an error” 保证了“分布式存储系统之内多副本之间的一致性”。对于应用来说,没有 “拜占庭错误” 是非常幸福的(本来有的数据变没了,破坏了存储节点的数据持久性),更容易做到“应用和分布式存储系统之间是一致的”,并显著减少应用的开发和维护成本。 常见的分布式文件系统有 Ali Pangu,HDFS,GlusterFS,Ceph,Lustre 等。对象存储有 Amazon S3,Aliyun OSS,Azure Blob Storage,Google Cloud Storage,OpenStack Swift 等。他们的简单对比如下: API 支持: 选用对象存储作为后端,通常无法像 HDFS 一样提供充分的 POSIX 能力支持,对于非 KV 型的操作往往存在一定性能问题,例如列出大量对象时需要数十秒,而在分布式文件系统中这类操作只需要毫秒甚至微秒。如果选用对象存储作为后端,弱化的 API 语义要求消息系统本身能够有序管理好这些对象的元数据。 容量与水平扩展: 对于云产品或者大规模企业的存储底座来说,以 HDFS 为例,当集群节点超过数百台,文件达到数亿量级以上时,NameNode 会产生性能瓶颈。一旦底层存储由于容量可用区等因素出现多套存储集群,这种 “本质复杂度” 在一定程度上削弱了 shareddisk 的架构简单性,并将这种复杂度向上传递给应用,影响消息产品本身的多租,迁移,容灾设计。典型的情况就是大型企业为了减少爆炸半径,往往会部署多套 K8s 并定制上层的 Cluster Federation(联邦)。 成本: 以国内云厂商官网公开的典型目录价为例: 本地磁盘,无副本 0.060.08 元/GB/月 云盘,SSD 1元/GB/月,高效云盘 0.35 元/GB/月 对象存储单 AZ 版 0.12 元/GB/月,多 AZ 版本 0.15 元/GB/月,低频 0.08 元/GB/月 分布式文件系统,如盘古 HDFS 接口,支持进一步转冷和 EC。 生态链: 对象存储和类 HDFS 都有足够多的经过生产验证的工具,监控报警层面对象存储的支持更产品化。 2.3. 直写还是转写 方案里,备受瞩目的点在于选择直写还是转写,我认为他们不冲突,两个方案 “可以分开有,都可以做强”。 多年来 RocketMQ 运行在基于本地存储的系统中,本地磁盘通常 IOPS 较高,成本较低但可靠性较差,大规模的生产实践中遇到的问题包括但不限于垂直扩容较难,坏盘,宿主机故障等。 直写: 指使用高可用的存储替换本地块存储,例如使用云盘多点挂载(分布式块存储形态,透明 rdma)或者直写分布式文件系统(下文简称 DFS)作为存储后端,此时主备节点可以共享存储,broker 的高可用中的数据流同步简化为只同步位点,在很大程度上减化了 RocketMQ 高可用的实现。 转写: 对于大部分数据密集型应用,出于故障恢复的考虑必须实时写日志,意味着无法对数据很好的进行攒批压缩,如果仅使用廉价介质,会带来更高的延迟以及更多的内存使用,无法满足生产需要。一个典型思路就是热数据使用容量小的高速介质先顺序写,compation 后转储到更廉价的存储系统中。 直写的目的是池化存储,转写的目的是降低数据的长期保存成本, 所以我认为一个理想的终态可以是两者的结合。RocketMQ 自己来做数据转冷,那有同学就会提出反问了,如果让 DFS 自身支持透明转冷,岂不是更好? 我的理解是 RocketMQ 希望在转冷这个动作时,能够做一些消息系统内部的格式变化来加速冷数据的读取,减少 IO 次数,配置不同 TTL 等。 相对于通用算法,消息系统自身对如何更好的压缩数据和加速读取的细节更加了解。 而且主动转冷的方案在审计和入湖的一些场景下,也可以被用于服务端批量转储数据到不同的平台,到 NoSQL 系统,到 ES,Click House,到对象存储,这一切是如此的自然~ 2.4. 技术架构演进 那么分级存储是一个尽善尽美的最终解决方案吗? 理想很美好,让我们来看一组典型生产场景的数据。 RocketMQ 在使用块存储时,存储节点存储成本大约会占到 30%50%。开启分级存储时,由于数据转储会产生一定的计算开销,主要包括数据复制,数据编解码,crc 校验等,不同场景下计算成本会上升 10%40%,通过换算,我们发现存储节点的总体拥有成本节约了 30% 左右。 考虑到商业和开源技术架构的一致性,选择了先实现转写模式,热数据的存储成本中随着存储空间显著减小,这能够更直接的降低存储成本,在我们充分建设好当前的转写逻辑时再将热数据的 WAL 机制和索引构建移植过来,实现基于分布式系统的直写技术,这种分阶段迭代会更加简明高效,这个阶段我们更加关注通用性和可用性。 可移植性: 直写分布式系统通常需要依赖特定 sdk,配合 rdma 等技术来降低延迟,对应用不完全透明,运维,人力,技术复杂度都有一定上升。保留成熟的本地存储,只需要实现存储插件就可以轻松的切换多种存储后端,不针对 IaaS 做深度绑定在可移植性上会有一定优势。 延迟与性能: 直写模式下存储紧密结合,应用层 ha 的简化也能降低延迟(写多数派成功才被消费者可见),但无论写云盘或者本地磁盘(同区域)延迟都会小于跨可用区的延迟,存储延迟在热数据收发链路不是瓶颈。 可用性: 存储后端往往都有复杂的容错和故障转移策略,直写与转写模式在公有云下可用性都满足诉求。考虑到转写模式下系统是弱依赖二级存储的,更适合开源与非公共云场景。 我们为什么不进一步压缩块存储的磁盘容量,做到几乎极致的成本呢? 事实上,在分级存储的场景下,一味的追求过小的本地磁盘容量价值不大。 主要有以下原因: 故障冗余,消息队列作为基础设施中重要的一环,稳定性高于一切。对象存储本身可用性较高,如果遇到网络波动等问题时,使用对象存储作为主存储,非常容易产生反压导致热数据无法写入, 而热数据属于在线生产业务,这对于可用性的影响是致命的。 过小的本地磁盘,在价格上没有明显的优势。 众所周知,云计算是注重普惠和公平的, 如果选用 50G 左右的块存储,又需要等价 150G 的 ESSD 级别的块存储能提供的 IOPS,则其单位成本几乎是普通块存储的数倍。 本地磁盘容量充足的情况下,上传时能够更好的通过 “攒批” 减少对象存储的请求费用。读取时能够对“温热” 数据提供更低的延迟和节约读取成本。 仅使用对象存储,难以对齐 RocketMQ 当前已经存在的丰富特性, 例如用于问题排查的随机消息索引,定时消息特性等,如果为了节约少量成本,极大的削弱基础设施的能力,反向要求业务方自建复杂的中间件体系是得不偿失的。 3. 分级存储的数据模型与实现 3.1. 模型与抽象 RocketMQ 本地存储数据模型如下: MappedFile:单个真实文件的句柄,也可以理解为 handle 或者说 fd,通过 mmap 实现内存映射文件。是一个 AppendOnly 的定长字节流语义的 Stream,支持字节粒度的追加写、随机读。每个 MappedFile 拥有自己的类型,写位点,创建更新时间等元数据。 MappedFileQueue:可以看做是零个或多个定长 MappedFile 组成的链表,提供了流的无边界语义。Queue 中最多只有最后一个文件可以是 Unseal 的状态(可写)。前面的文件都必须都是 Sealed 状态(只读),Seal 操作完成后 MappedFile 是 immutable(不可变)的。 CommitLog:MappedFileQueue 的封装,每个 “格子” 存储一条序列化的消息到无界的流中。 ConsumeQueue:顺序索引,指向 CommitLog 中消息在 FileQueue 中的偏移量(offset)。 RocketMQ 分级存储提供的数据模型和本地模型类似,改变了 CommitLog 和 ConsumeQueue 的概念: TieredFileSegment:和 MappedFile 类似,描述一个分级存储系统中文件的句柄。 TieredFlatFile:和 MappedFileQueue 类似。 TieredCommitLog 和本地 CommitLog 混合写不同,按照单个 Topic 单个队列的粒度拆分多条 CommitLog。 TieredConsumeQueue 指向 TieredCommitLog 偏移量的一个索引,是严格连续递增的。实际索引的位置会从指向的 CommitLog 的位置改为 TieredCommitLog 的偏移量。 CompositeFlatFile:组合 TieredCommitLog 和 TieredConsumeQueue 对象,并提供概念的封装。 3.2. 消息上传流程 RocketMQ 的存储实现了一个 Pipeline,类似于拦截器链,Netty 的 handler 链,读写请求会经过这个 Pipeline 的多个处理器。 Dispatcher 的概念是指为写入的数据构建索引,在分级存储模块初始化时,会创建 TieredDispatcher 注册为 CommitLog 的 dispatcher 链的一个处理器。每当有消息发送到 Broker 会调用 TieredDispatcher 进行消息分发。下面我们来追踪单条消息进入存储层的流程: 1. 消息被顺序追加到本地 commitlog 并更新本地 max offset(图中黄色部分),为了防止宕机时多副本产生“读摆动”,多副本中多数派的最小位点会作为“低水位”被确认,这个位点被称为 commit offset(图中 2500)。换句话说,commit offset 与 max offset 之间的数据是正在等待多副本同步的。 2. 当 commit offset = message offset 之后,消息会被上传到二级存储的 commitlog 的缓存中(绿色部分)并更新这个队列的 max offset。 3. 消息的索引会被追加到这个队列的 consume queue 中并更新 consume queue 的 max offset。 4. 一旦 commitlog 中缓存大小超过阈值或者等待达到一定时间,消息的缓存将被上传至 commitlog,之后才会将索引信息提交,这里有一个隐含的数据依赖,使索引被晚于原始数据更新。这个机制保证了所有 cq 索引中的数据都能在 commitlog 中找到。宕机场景下,分级存储中的 commitlog 可能会重复构建,此时没有 cq 指向这段数据。由于文件本身还是被使用 Queue 的模型管理的,使得整段数据在达到 TTL 时能被回收,此时并不会产生数据流的“泄漏”。 5. 当索引也上传完成的时候,更新分级存储中的 commit offset(绿色部分被提交)。 6. 系统重启或者宕机时,会选择多个 dispatcher 的最小位点向 max offset 重新分发,确保数据不丢失。 在实际执行中,上传部分由三组线程协同工作。 1. store dispatch 线程,由于该线程负责本地 cq 的分发,我们不能长时间阻塞该线程,否则会影响消息进入本地存储的“可见性延迟”。因此 store dispatch 每次只会尝试对拆分后的文件短暂加锁,如果加锁成功,将消息数据放入拆分后的 commitlog 文件的缓冲区则立即退出,该操作不会阻塞。若获取锁失败则立即返回。 2. store compensate 线程组,负责对本地 cq 进行定时扫描,当写入压力较高时,步骤 1 可能获取锁失败,这个环节会批量的将落后的数据放入 commitlog 中。原始数据被放入后会将 dispatch request 放入 write map。 3. build cq index 线程。write map 和 read map 是一个双缓冲队列的设计,该线程负责将 read map 中的数据构建 cq 并上传。如果 read map 为空,则交换缓冲区,这个双缓冲队列在多个线程共享访问时减少了互斥和竞争操作。 各类存储系统的缓冲攒批策略大同小异,而线上的 topic 写入流量往往是存在热点的,根据经典的二八原则,RocketMQ 分级存储模块目前采用了 “达到一定数据量”,“达到一定时间”两者取其小的合并方式。 这种方式简单可靠,对于大流量的 topic 很容易就可以达到批的最小数据量,对于流量较低的 topic 也不会占用过多的内存。从而减少了对象存储的请求数,其开销主要包括 restful 协议请求头,签名和传输等。诚然,攒批的逻辑仍然存在较大的优化空间,例如 IOT,数据分片同步等各个 topic 流量较为平均的场景使用类似 “滑动窗口” 的加权平均算法,或者基于信任值的流量控制策略可以更好的权衡延迟和吞吐。 3.3. NonStopWrite 特性 NonStopWrite 模型实际上是一致性模型的一部分。实际生产中,后端分布式存储系统的断连和网络问题偶尔会不可避免,而 Append 模型实际上一种强顺序的模型,参考 HDFS 的 23 异步写,我们提出了一种基于 Append 和 Put 的混合模型。 例如:对于如下图片中的 stream,commit / confirm offset = 150,max offset = 200。此时写出缓冲区中的数据包括 150200 的 uncommitted 部分,还有 200 以后源源不断的写入的新数据。 假设后端存储系统支持原子性写入,单个上传请求的数据内容是 150200 这个区间,当单次上传失败时,我们需要向服务端查询上一次写入的位点并进行错误处理。 如果返回的长度是 150,说明上传失败,应用需要重传 buffer。 如果返回的长度是 200,说明前一次上传成功但没有收到成功的 response,提升 commit offset 至 200。 而另一种解决方案是,使用 NonStopWrite 机制立刻新切换一个文件,以 150 作为文件名,立刻重传 150 至 200 的数据,如果有新的数据也可以立刻与这些数据一起上传,我们发现混合模型存在显著优势: 对于绝大部分没有收到成功的响应时,上传是失败的而不是超时,立刻切换文件可以不去 check in 文件长度,减少 rpc 数量。 立刻重传不会阻塞后续新的数据上传,不容易由于后端数据无法写出造成反压,导致前端写失败。 无论 150200 这段数据在第一个文件是到底是写成功还是失败都无关紧要,因为不会去读取这段数据。尤其是对于不支持请求粒度原子写入的模型来说,如果上一次请求的结果是 180,那么错误处理将会非常复杂。 3.4. 随机索引重排 21 年的时候,我第一次听到用“读扩散”或者“写扩散”来描述一个设计方案, 这两个词简洁的概括了应用性能设计的本质。各种业务场景下,我们总是选择通过读写扩散, 选择通过格式的变化,将数据额外转储到一份性能更好或者更廉价的存储, 或者通过读扩散减少数据冗余(减少索引提高了平均查询代价)。 RocketMQ 会在先内存构建基于 hash 的持久化索引文件 IndexFile(非 AppendOnly),再通过 mmap 异步的将数据持久化到磁盘。这个文件是为了支持用户通过 key,消息 ID 等信息来追踪一条消息。 对于单条消息会先计算 hash(topickey) % slot_num 选择 hash slot (黄色部分) 作为随机索引的指针,对象索引本身会附加到 index item 中,hash slot 使用“哈希拉链”的方式解决冲突,这样便形成了一条当前 slot 按照时间存入的倒序的链表。不难发现,查询时需要多次随机读取链表节点。 由于冷存储的 IOPS 代价是非常昂贵的,在设计上我们希望可以面向查询进行优化。新的文件结构类似于维护没有 GC 和只有一次 compation 的 LSM 树,数据结构的调整如下: 1. 等待本地一个 IndexFile 完全写满,规避修改操作,在高 IOPS 的存储介质上异步 compation,完成后删除原来的文件。 2. 从冷存储查询延迟高,而单次返回的数据量大小(不太大的场景)并不会明显改变延迟。compation 时优化数据结构,做到用一次查询连续的一段数据替换多次随机点查。 3. hash slot 的指向的 List 是连续的,查询时可以根据 hash slot 中的 item offset 和 item size 一次取出所有 hashcode 相同的记录并在内存中过滤。 3.5. 消息读取流程 3.5.1 读取策略 读取是写入的逆过程,优先从哪里取回想要的数据必然存在很多的工程考虑与权衡。如图所示,近期的数据被缓存在内存中,稍久远的数据存在与内存和二级存储上,更久远的数据仅存在于二级存储。当被访问的数据存在于内存中,由于内存的速度快速存储介质,直接将这部分数据通过网络写会给客户端即可。如果被访问的数据如图中 request 的指向,存在于本地磁盘又存在于二级存储,此时应该根据一二级存储的特性综合权衡请求落到哪一层。 有两种典型的想法: 1. 数据存储被视为多级缓存,越上层的介质随机读写速度快,请求优先向上层存储进行查询,当内存中不存在了就查询本地磁盘,如果还不存在才向二级存储查询。 2. 由于在转冷时主动对数据做了 compation,从二级存储读取的数据是连续的,此时可以把更宝贵一级存储的 IOPS 留给在线业务。 RocketMQ 的分级存储将这个选择抽象为了读取策略,通过请求中的逻辑位点(queue offset)判断数据处于哪个区间,再根据具体的策略进行选择: DISABLE:禁止从多级存储中读取消息,可能是数据源不支持。 NOT_IN_DISK:不在一级存储的的消息都会从二级存储中读取。 NOT_IN_MEM:不在内存中的消息即冷数据从多级存储读取。 FORCE:强制所有消息从多级存储中读取,目前仅供测试使用。 3.5.2 预读设计 TieredMessageFetcher 是 RocketMQ 分级存储取回数据的具体实现。 为了加速从二级存储读取的速度和减少整体上对二级存储请求数,采用了预读缓存的设计: 即 TieredMessageFetcher 读取消息时会预读更多的消息数据,预读缓存的设计参考了 TCP Tahoe 拥塞控制算法,每次预读的消息量类似拥塞窗口采用加法增、乘法减的流量控制机制。 加法增:从最小窗口开始,每次增加等同于客户端 batchSize 的消息量。 乘法减:当缓存的消息超过了缓存过期时间仍未被全部拉取,此时一般是客户端缓存满,消息数据反压到服务端,在清理缓存的同时会将下次预读消息量减半。 此外,在客户端消费速度较快时,向二级存储读取的消息量较大,此时会使用分段策略并发取回数据。 3.6. 定时消息的分级存储 除了普通消息,RocketMQ 支持设置未来几十天的长定时消息,而这部分数据严重挤占了热数据的存储空间。 RocketMQ 实现了基于本地文件系统的时间轮,整体设计如左侧所示。单节点上所有的定时消息会先写入 rmq_sys_wheel_timer 的系统 topic,进入时间轮,出队后这些消息的 topic 会被还原为真实的业务 topic。 “从磁盘读取数据”和“将消息索引放入时间轮”这两个动作涉及到 IO 与计算,为了减少这两个阶段的锁竞争引入了 Enqueue 作为中转的等待队列,EnqueuGet 和 EnqueuePut 分别负责写入和读取数据,这个设计简单可靠。 不难发现,所有的消息都会进入时间轮,这也是挤占存储空间的根本原因。 写入时,RocketMQ 的分级存储定时消息针对 EnqueuePut 做了一个分流,对于大于当前时间数小时的消息会被写入到基于分级存储的 TimerFlatFile 文件中,我们维护了一个 ConcurrentSkipListMap timerFlatFileTable; 每间隔 1 小时,设置一个 TimerFlatFile,对于 T+n 至 T+n+1 的定时消息,会先被混合追加到 T+n 所对应的文件中。 读取时,当前时间 + 1 小时的消息将被提前出队,这些消息又会重新进入本地 TimerStore 的系统 topic 中/此时,由于定时时间都是将来一小段时间的,他们不再会进入时间轮的结构中。 在这个设计上有一些工程性的考虑: timerFlatFileTable 中的 Key 很多,会不会让分级存储上的数据碎片化?分布式文件系统底层一般使用类 LSM 结构,RocketMQ 只关心 LBA 结构,可以通过优化 Enqueue 的 buffer 让写分级存储时数据达到攒批的效果。 可靠的位点,Enqueue 到“时间轮”和 timerFlatFileTable 可以共用一个 commit offset。对于单条消息来说,只要它进入时间轮或者被上传成功,我们就认为一条消息已经持久化了。由于更新到二级存储本身需要一些攒批缓冲的过程,会延迟 commit offset 的更新,但是这个缓冲时间是可控的。 我们发现偶尔本地存储转储到二级存储会较慢,使用双缓冲队列实现读写分离(如图片中绿色部分)此时消息被放入写缓存,随后转入读缓存队列,最后进入上传流程。 4. 分级存储企业级竞争力 4.1. 冷数据的压缩与归档 压缩是一种经典的时间与空间交换的权衡,其目的在于通过较小的 CPU 开销,实现更少的磁盘占用或网络 I/O 传输。目前,RocketMQ 的热存储在考虑延迟的情况下,仅对单条大于 4K 的消息进行单条压缩存储。对于冷存储服务其实可以做两个层面的压缩与归档处理。 消息队列业务层面,对于大多数业务 Topic,其 Body 通常存在相似性,可将其压缩至原大小的几分之一至几十分之一。 底层存储层面,使用 EC 纠删码,数据被分成若干个数据块,然后再根据一定的算法,生成一些冗余块。当数据丢失时,可以使用其余的数据块和冗余块来恢复丢失的数据块,从而保证数据的完整性和可靠性。典型的 EC 算法后存储空间的使用可以降低到 1.375 副本。 业界也有一些基于 FPGA 实现存储压缩加速的案例,我们将持续探索这方面的尝试。 4.2. 原生的只读挂载能力实现 Serverless 业界对 Serverless 有不同的理解,过去 RocketMQ 多节点之间不共享存储,导致“扩容快,缩容慢”,例如 A 机器需要下线,则必须等普通消息消费完,定时消息全部出队才能进行运维操作。分级存储设计通过 shareddisk的方式实现跨节点代理读取下线节点的数据,如右图所示:A 的数据此时可以被 B 节点读取,彻底释放了 A 的计算资源和一级存储资源。 这种缩容的主要流程如下: 1. RocketMQ 实现了一个简单的选举算法,正常情况下集群内每一个节点都持有对自己数据独占的写锁。 2. 待下线的节点做优雅下线,确保近期定时消息,事务消息,pop retry 消息都已被完整处理。上传自己的元数据信息到共享的二级存储,并释放自己的写锁。 3. 集群使用一定的负载均衡算法,新的节点获取写锁,将该 Broker 的数据以只读的形式挂载。 4. 将原来节点的元数据注册到 NameServer 对客户端暴露。 5. 对于原节点的写请求,例如位点更新,将在内存中处理并周期性快照到共享存储中。 5. 总结 RocketMQ 的存储在云原生时代的演进中遇到了更多有趣的场景和挑战,这是一个需要全链路调优的复杂工程。出于可移植性和通用性的考虑,我们还没有非常有效的使用 DPDK + SPDK + RDMA 这些新颖的技术,但我们解决了许多工程实践中会遇到的问题并构建了整个分级存储的框架。在后续的发展中,我们会推出更多的存储后端实现,针对延迟和吞吐量等细节做深度优化。 参考文档: [1] Chang, F., Dean, J., Ghemawat, S., et al. Bigtable: A distributed storage system for structured data. ACM Transactions on Computer Systems, 2008, 26(2): 4.[2] Liu, Y., Zhang, K., & Spear, M. DynamicSized Nonblocking Hash Tables. In Proceedings of the ACM Symposium on Principles of Distributed Computing, 2014.[3] Ongaro, D., & Ousterhout, J. In Search of an Understandable Consensus Algorithm. Proceedings of the USENIX Conference on Operating Systems Design and Implementation, 2014, 305320.[4] Apache RocketMQ. GitHub, _https://github.com/apache/rocketmq_[5] Verbitski, A., Gupta, A., Saha, D., et al. Amazon aurora: On avoiding distributed consensus for i/os, commits, and membership changes. In Proceedings of the 2018 International Conference on Management of Data, 2018, 789796.[6] Antonopoulos, P., Budovski, A., Diaconu, C., et al. Socrates: The new sql server in the cloud. In Proceedings of the 2019 International Conference on Management of Data, 2019, 17431756.[7] Li, Q. More Than Capacity: Performanceoriented Evolution of Pangu in Alibaba. Fast 2023_https://www.usenix.org/conference/fast23/presentation/liqiangdeployed_[8] Lu, S. Perseus: A FailSlow Detection Framework for Cloud Storage Systems. Fast 2023
查看全部文章
ABOUT US
Apache RocketMQ事件驱动架构全景图
微服务
Higress
Dubbo
Sentinel
Seata
Spring Cloud
Nacos
物联网
家电
汽车
穿戴设备
充电桩
工业设备
手机
事件驱动架构平台
RabbitMQ
Kafka
EventBridge
MQTT
RocketMQ
MNS
Apache RocketMQ as Core
计算
模型服务
函数计算
容器
存储
对象存储
数据库
NoSQL
分析
Flink
Spark
Elastic Search
事件
云服务器
对象存储
云监控
SaaS事件
通知
语音
短信
邮箱
移动推送

产品特点

为什么学习 RocketMQ

云原生
生于云,长于云,无限弹性扩缩,K8S 友好
高吞吐
万亿级吞吐保证,同时满足微服务于大数据场景
流处理
提供轻量、高扩展、高性能和丰富功能的流计算引擎
金融级
金融级的稳定性,广泛用于交易核心链路
架构极简
零外部依赖,Shared-nothing 架构
生态友好
无缝对接微服务、实时计算、数据湖等周边生态
浙ICP备12022327号-1120