万亿级数据洪峰下的分布式消息引擎

前言

通过简单回顾阿里中间件(Aliware)消息引擎的发展史，本文开篇于双11消息引擎面临的低延迟挑战，通过经典的应用场景阐述可能会面临的问题 - 响应慢，雪崩，用户体验差，继而交易下跌。为了应对这些不可控的洪峰数据，中间件团队通过大量研究和实践，推出了低延迟高可用解决方案，在分布式存储领域具有一定的普适性。在此基础上，通过对现有有限资源的规划，又推出了分级的容量保障策略，通过限流、降级，甚至熔断技术，能够有效保障重点业务的高吞吐，成功的支撑集团包括海外业务平缓舒畅地度过双11高峰。与此同时，在一些对高可靠、高可用要求极为苛刻的场景下，中间件团队又重点推出了基于多副本机制的高可用解决方案，能够动态识别机器宕机、机房断网等灾难场景，自动实现主备切换。整个切换过程对用户透明，运维开发人员无需干预，极大地提升消息存储的可靠性以及整个集群的高可用性。

1. 消息引擎家族史

阿里中间件消息引擎发展到今日，前前后后经历了三代演进。第一代，推模式，数据存储采用关系型数据库。在这种模式下，消息具有很低的延迟特性，尤其在阿里淘宝这种高频交易场景中，具有非常广泛地应用。第二代，拉模式，自研的专有消息存储。能够媲美Kafka的吞吐性能，但考虑到淘宝的应用场景，尤其是其交易链路等高可靠场景，消息引擎并没有一位的追求吞吐，而是将稳定可靠放在首位。因为采用了长连接拉模式，在消息的实时方面丝毫不逊推模式。在前两代经历了数年线上堪比工况的洗礼后，中间件团队于2011年研发了以拉模式为主，兼有推模式的高性能、低延迟消息引擎RocketMQ。并在2012年进行了开源，经历了6年双11核心交易链路检验，愈久弥坚。目前已经捐赠给阿帕奇基金会（ASF），有望成为继ActiveMQ，Kafka之后，Apache社区第三个重量级分布式消息引擎。时至今日，RocketMQ很好的服务了阿里集团大大小小上千个应用，在双11当天，更有不可思议的万亿级消息流转，为集团大中台的稳定发挥了举足轻重的作用。

2. 低延迟可用性探索

疾风吹征帆，倏尔向空没。千里在俄顷，三江坐超忽。—孟浩然

2.1低延迟与可用性

随着Java语言生态的完善，JVM性能的提高，C和C++已经不再是低延迟场景唯一的选择。本章节重点介绍RocketMQ在低延迟可用性方面的一些探索。
应用程序的性能度量标准一般从吞吐量和延迟两方面考量。吞吐量是指程序在一段时间内能处理的请求数量。延迟是指端到端的响应时间。低延迟在不同的环境下有不同的定义，比如在聊天应用中低延迟可以定义为200ms内，在交易系统中定义为10ms内。相对于吞吐量，延迟会受到很多因素的影响，如CPU、网络、内存、操作系统等。
根据Little’s law，当延迟变高时，驻留在分布式系统中的请求会剧增，导致某些节点不可用，不可用的状态甚至会扩散至其它节点，造成整个系统的服务能力丧失，这种场景又俗称雪崩。所以打造低延迟的应用程序，对提升整个分布式系统可用性有很大的裨益。

2.2 低延迟探索之路

RocketMQ作为一款消息引擎，最大的作用是异步解耦和削峰填谷。一方面，分布式应用会利用RocketMQ来进行异步解耦，应用程序可以自如地扩容和缩容。另一方面，当洪峰数据来临时，大量的消息需要堆积到RocketMQ中，后端程序可以根据自己的消费速度来进行数据的读取。所以保证RocketMQ写消息链路的低延迟至关重要。
在今年双11期间，天猫发布了红包火山的新玩法。该游戏对延迟非常敏感，只能容忍50ms内的延迟，在压测初期RocketMQ写消息出现了大量50~500ms的延迟，导致了在红包喷发的高峰出现大量的失败，严重影响前端业务。下图为压测红包集群在压测时写消息延迟热力图统计。

作为一款纯Java语言开发的消息引擎，RocketMQ自主研发的存储组件，依赖Page Cache进行加速和堆积，意味着它的性能会受到JVM、GC、内核、Linux内存管理机制、文件IO等因素的影响。如下图所示，一条消息从客户端发送出，到最终落盘持久化，每个环节都有产生延迟的风险。通过对线上数据的观察，RocketMQ写消息链路存在偶发的高达数秒的延迟。

2.2.1 JVM停顿

JVM（Java虚拟机）在运行过程中会产生很多停顿，常见的有GC、JIT、取消偏向锁（RevokeBias）、RedefineClasses（AOP）等。对应用程序影响最大的则是GC停顿。RocketMQ尽量避免Full GC，但Minor GC带来的停顿是难以避免的。针对GC调优是一个很伽利略的问题，需要通过大量的测试来帮助应用程序调整GC参数，比如可以通过调整堆大小，GC的时机，优化数据结构等手段进行调优。
对于其它JVM停顿，可以通过-XX:+PrintGCApplicationStoppedTime将JVM停顿时间输出到GC日志中。通过-XX:+PrintSafepointStatistics -XX: PrintSafepointStatisticsCount=1输出具体的停顿原因，并进行针对性的优化。比如在RocketMQ中发现取RevokeBias产生了大量的停顿，通过-XX:-UseBiasedLocking关闭了偏向锁特性。
另外，GC日志的输出会发生文件IO，有时候也会造成不必要的停顿，可以将GC日志输出到tmpfs（内存文件系统）中，但tmpfs会消耗内存，为了避免内存被浪费可以使用-XX:+UseGCLogFileRotation滚动GC日志。
除了GC日志会产生文件IO，JVM会将jstat命令需要的一些统计数据输出到/tmp(hsperfdata)目录下，可通过-XX:+PerfDisableSharedMem关闭该特性，并使用JMX来代替jstat。

2.2.2 锁——同步的“利”器

作为一种临界区的保护机制，锁被广泛用于多线程应用程序的开发中。但锁是一把双刃剑，过多或不正确的使用锁会导致多线程应用的性能下降。
Java中的锁默认采用的是非公平锁，加锁时不考虑排队问题，直接尝试获取锁，若获取失败自动进行排队。非公平锁会导致线程等待时间过长，延迟变高。倘若采取公平锁，又会对应用带来较大性能损失。
另一方面，同步会引起上下文切换，这会带来一定的开销。上下文切换一般是微秒级，但当线程数过多，竞争压力大时，会产生数十毫秒级别的开销。可通过LockSupport.park来模拟产生上下文切换进行测试。
为了避免锁带来的延迟，利用CAS原语将RocketMQ核心链路无锁化，在降低延迟的同时显著提高吞吐量。

2.2.3 内存——没那么快

受限于Linux的内存管理机制，应用程序访问内存时有时候会产生高延迟。Linux中内存主要有匿名内存和Page Cache两种。
Linux会用尽可能多的内存来做缓存，大多数情形下，服务器可用内存都较少。可用内存较少时，应用程序申请或者访问新的内存页会引发内存回收，当后台内存回收的速度不及分配内存的速度时，会进入直接回收（Direct Reclaim），应用程序会自旋等待内存回收完毕，产生巨大的延迟，如下图所示。

另一方面，内核也会回收匿名内存页，匿名内存页被换出后下一次访问会产生文件IO，导致延迟，如下图所示。

上述两种情况产生的延迟可以通过内核参数(vm.extra_free_kbytes和vm.swappiness)调优加以避免。
Linux对内存的管理一般是以页为单位，一页一般为4k大小，当在同一页内存上产生读写竞争时，会产生延迟，对于这种情况，需要应用程序自行协调内存的访问加以避免。

2.2.4 Page Cache——利与弊

Page Cache是文件的缓存，用于加速对文件的读写，它为RocketMQ提供了更强大的堆积能力。RocketMQ将数据文件映射到内存中，写消息的时候首先写入Page Cache，并通过异步刷盘的模式将消息持久化（同时也支持同步刷盘），消息可以直接从Page Cache中读取，这也是业界分布式存储产品通常采用的模式，如下图所示：

该模式大多数情况读写速度都比较迅速，但当遇到操作系统进行脏页回写，内存回收，内存换入换出等情形时，会产生较大的读写延迟，造成存储引擎偶发的高延迟。
针对这种现象，RocketMQ采用了多种优化技术，比如内存预分配，文件预热，mlock系统调用，读写分离等，来保证利用Page Cache优点的同时，消除其带来的延迟。

2.3 优化成果

RocketMQ通过对上述情况的优化，成功消除了写消息高延迟的情形，并通过了今年双11的考验。优化后写消息耗时热力图如下图所示。

优化后RocketMQ写消息延迟99.995%在1ms内，100%在100ms内，如下图所示。

3 容量保障三大法宝

他强任他强，清风拂山岗。他横任他横，明月照大江。—九阳真经心法

有了低延迟的优化保障，并不意味着消息引擎就可以高枕无忧。为了给应用带来如丝般顺滑的体验，消息引擎必须进行灵活的容量规划。如何让系统能够在汹涌澎湃的流量洪峰面前谈笑风生？降级、限流、熔断三大法宝便有了用武之地。丢卒保车，以降级、暂停边缘服务、组件为代价保障核心服务的资源，以系统不被突发流量击垮为第一要务。正所谓，他强任他强，清风拂山岗。他横任他横，明月照大江！
从架构的稳定性角度看，在有限资源的情况下，所能提供的单位时间服务能力也是有限的。假如超过承受能力，可能会带来整个服务的停顿，应用的Crash，进而可能将风险传递给服务调用方造成整个系统的服务能力丧失，进而引发雪崩。另外，根据排队理论，具有延迟的服务随着请求量的不断提升，其平均响应时间也会迅速提升，为了保证服务的SLA，有必要控制单位时间的请求量。这就是限流为什么愈发重要的原因。限流这个概念，在学术界又被称之为Traffic Shaping。最早起源于网络通讯领域，典型的有漏桶（leaky bucket）算法和令牌桶（token bucket）算法。

漏桶算法基本思路是有一个桶（会漏水），水以恒定速率滴出，上方会有水滴（请求）进入水桶。如果上方水滴进入速率超过水滴出的速率，那么水桶就会溢出，即请求过载。

令牌桶算法基本思路是同样也有一个桶，令牌以恒定速率放入桶，桶内的令牌数有上限，每个请求会acquire一个令牌，如果某个请求来到而桶内没有令牌了，则这个请求是过载的。很显然，令牌桶会存在请求突发激增的问题。

无论是漏桶、令牌桶，抑或其它变种算法，都可以看做是一种控制速度的限流，工程领域如Guava里的RateLimiter，Netty里的TrafficShaping等也都属于此。除此之外，还有一种控制并发的限流模式，如操作系统里的信号量，JDK里的Semaphore。
异步解耦，削峰填谷，作为消息引擎的看家本领，Try your best本身就是其最初的设计初衷（RPC、应用网关、容器等场景下，控制速度应成为流控首选）。但即便如此，一些必要的流控还是需要考量。不过与前面介绍的不同，RocketMQ中并没有内置Guava、Netty等拆箱即用的速度流控组件。而是通过借鉴排队理论，对其中的慢请求进行容错处理。这里的慢请求是指排队等待时间以及服务时间超过某个阈值的请求。对于离线应用场景，容错处理就是利用滑动窗口机制，通过缓慢缩小窗口的手段，来减缓从服务端拉的频率以及消息大小，降低对服务端的影响。而对于那些高频交易，数据复制场景，则采取了快速失败策略，既能预防应用连锁的资源耗尽而引发的应用雪崩，又能有效降低服务端压力，为端到端低延迟带来可靠保障。
服务降级是一种典型的丢卒保车，二八原则实践。而降级的手段也无外乎关闭，下线等“简单粗暴”的操作。降级目标的选择，更多来自于服务QoS的定义。消息引擎早期对于降级的处理主要来自两方面，一方面来自于用户数据的收集，另一方面来自引擎组件的服务QoS设定。对于前者，通过运维管控系统推送应用自身QoS数据，一般会输出如下表格。而引擎组件的服务QoS，如服务于消息问题追溯的链路轨迹组件，对于核心功能来说，定级相对较低，可在洪峰到来之前提前关闭。

谈到熔断，不得不提经典的电力系统中的保险丝，当负载过大，或者电路发生故障或异常时，电流会不断升高，为防止升高的电流有可能损坏电路中的某些重要器件或贵重器件，烧毁电路甚至造成火灾。保险丝会在电流异常升高到一定的高度和热度的时候，自身熔断切断电流，从而起到保护电路安全运行的作用。
同样，在分布式系统中，如果调用的远程服务或者资源由于某种原因无法使用时，没有这种过载保护，就会导致请求的资源阻塞在服务器上等待从而耗尽系统或者服务器资源。很多时候刚开始可能只是系统出现了局部的、小规模的故障，然而由于种种原因，故障影响的范围越来越大，最终导致了全局性的后果。而这种过载保护就是大家俗称的熔断器(Circuit Breaker)。Netflix公司为了解决该问题，开源了它们的熔断解决方案Hystrix。

上述三幅图，描述了系统从初始的健康状态到高并发场景下阻塞在下游的某个关键依赖组件的场景。这种情况很容易诱发雪崩效应。而通过引入Hystrix的熔断机制，让应用快速失败，继而能够避免最坏情况的发生。

借鉴Hystrix思路，中间件团队自研了一套消息引擎熔断机制。在大促压测备战期间，曾经出现过由于机器硬件设备导致服务不可用。如果采用常规的容错手段，是需要等待30秒时间，不可用机器才能从列表里被摘除。但通过这套熔断机制，能在毫秒范围内识别并隔离异常服务。进一步提升了引擎的可用性。

4. 高可用解决方案

昔之善战者，先为不可胜，以待敌之可胜。不可胜在己，可胜在敌。故善战者，能为不可胜，不能使敌之必可胜。故曰：胜可知，而不可为。—孙武

虽然有了容量保障的三大法宝作为依托，但随着消息引擎集群规模的不断上升，到达一定程度后，集群中机器故障的可能性随之提高，严重降低消息的可靠性以及系统的可用性。与此同时，基于多机房部署的集群模式也会引发机房断网，进一步降低消息系统的可用性。为此，阿里中间件（Aliware）重点推出了基于多副本的高可用解决方案，动态识别机器故障、机房断网等灾难场景，实现故障自动恢复；整个恢复过程对用户透明，无需运维人员干预，极大地提升了消息存储的可靠性，保障了整个集群的高可用性。
高可用性几乎是每个分布式系统在设计时必须要考虑的一个重要特性，在遵循CAP原则（即：一致性、可用性和分区容错性三者无法在分布式系统中被同时满足，并且最多只能满足其中两个）基础上，业界也提出了一些针对分布式系统通用的高可用解决方案，如下图所示：

其中，行代表了分布式系统中通用的高可用解决方案，包括冷备、Master/Slave、Master/Master、两阶段提交以及基于Paxos算法的解决方案；列代表了分布式系统所关心的各项指标，包括数据一致性、事务支持程度、数据延迟、系统吞吐量、数据丢失可能性、故障自动恢复方式。
从图中可以看出，不同的解决方案对各项指标的支持程度各有侧重。基于CAP原则，很难设计出一种高可用方案能同时够满足所有指标的最优值，以Master/Slave为例，一般满足如下几个特性：
1) Slave是Master的备份，可以根据数据的重要程度设置Slave的个数。
数据写请求命中Master，读请求可命中Master或者Slave。
2) 写请求命中Master之后，数据可通过同步或者异步的方式从Master复制到Slave上；其中同步复制模式需要保证Master和Slave均写成功后才反馈给客户端成功；异步复制模式只需要保证Master写成功即可反馈给客户端成功。
数据通过同步或者异步方式从Master复制到Slave上，因此Master/Slave结构至少能保证数据的最终一致性；异步复制模式下，数据在Master写成功后即可反馈给客户端成功，因此系统拥有较低的延迟和较高的吞吐量，但同时会带来Master故障丢数据的可能性；如期望异步复制模式下Master故障时数据仍不丢，Slave只能以Read-Only的方式等待Master的恢复，即延长了系统的故障恢复时间。相反，Master/Slave结构中的同步复制模式会以增大数据写入延迟、降低系统吞吐量的代价来保证机器故障时数据不丢，同时降低系统故障恢复时间。

5. RocketMQ高可用架构

RocketMQ基于原有多机房部署的集群模式，利用分布式锁和通知机制，借助Controller组件，设计并实现了Master/Slave结构的高可用架构，如下图所示：

其中，Zookeeper作为分布式调度框架，需要至少在A、B、C三个机房部署以保证其高可用，并为RocketMQ高可用架构提供如下功能：
1) 维护持久节点（PERSISTENT），保存主备状态机；
2) 维护临时节点（EPHEMERAL），保存RocketMQ的当前状态；
3) 当主备状态机、服务端当前状态发生变更时，通知对应的观察者。
RocketMQ以Master/Slave结构实现多机房对等部署，消息的写请求会命中Master，然后通过同步或者异步方式复制到Slave上进行持久化存储；消息的读请求会优先命中Master，当消息堆积导致磁盘压力大时，读请求转移至Slave。
RocketMQ直接与Zookeeper进行交互，体现在：
1) 以临时节点的方式向Zookeeper汇报当前状态；
2) 作为观察者监听Zookeeper上主备状态机的变更。当发现主备状态机变化时，根据最新的状态机更改当前状态；
RocketMQ HA Controller是消息引擎高可用架构中降低系统故障恢复时间的无状态组件，在A、B、C三个机房分布式部署，其主要职责体现在：
1) 作为观察者监听Zookeeper 上RocketMQ当前状态的变更；
2) 根据集群的当前状态，控制主备状态机的切换并向Zookeeper汇报最新主备状态机。
出于对系统复杂性以及消息引擎本身对CAP原则适配的考虑，RocketMQ高可用架构的设计采用了Master/Slave结构，在提供低延迟、高吞吐量消息服务的基础上，采用主备同步复制的方式避免故障时消息的丢失。数据同步过程中，通过维护一个递增的全局唯一SequenceID来保证数据强一致。同时引入故障自动恢复机制以降低故障恢复时间，提升系统的可用性。

5.1 可用性评估

系统可用性（Availability）是信息工业界用来衡量一个信息系统提供持续服务的能力，它表示的是在给定时间区间内系统或者系统某一能力在特定环境中能够正常工作的概率。简单地说， 可用性是平均故障间隔时间（MTBF）除以平均故障间隔时间（MTBF）和平均故障修复时间（MTTR）之和所得的结果， 即：

通常业界习惯用N个9来表征系统可用性，比如99.9%代表3个9的可用性，意味着全年不可用时间在8.76小时以内；99.999%代表5个9的可用性，意味着全年不可用时间必须保证在5.26分钟以内，缺少故障自动恢复机制的系统将很难达到5个9的高可用性。

5.2 RocketMQ 高可用保障

通过可用性计算公式可以看出，要提升系统的可用性，需要在保障系统健壮性以延长平均无故障时间的基础上，进一步加强系统的故障自动恢复能力以缩短平均故障修复时间。RocketMQ高可用架构设计并实现了Controller组件，按照单主状态、异步复制状态、半同步状态以及最终的同步复制状态的有限状态机进行转换。在最终的同步复制状态下，Master和Slave任一节点故障时，其它节点能够在秒级时间内切换到单主状态继续提供服务。相比于之前人工介入重启来恢复服务，RokcetMQ高可用架构赋予了系统故障自动恢复的能力，能极大缩短平均故障恢复时间，提升系统的可用性。

下图描述了RocketMQ高可用架构中有限状态机的转换：

1） 第一个节点启动后，Controller控制状态机切换为单主状态，通知启动节点以Master角色提供服务。
2） 第二个节点启动后，Controller控制状态机切换成异步复制状态。Master通过异步方式向Slave复制数据。
3） 当Slave的数据即将赶上Master，Controller控制状态机切换成半同步状态，此时命中Master的写请求会被Hold住，直到Master以异步方式向Slave复制了所有差异的数据。
4） 当半同步状态下Slave的数据完全赶上Master时，Controller控制状态机切换成同步复制模式，Mater开始以同步方式向Slave复制数据。该状态下任一节点出现故障，其它节点能够在秒级内切换到单主状态继续提供服务。
Controller组件控制RocketMQ按照单主状态，异步复制状态，半同步状态，同步复制状态的顺序进行状态机切换。中间状态的停留时间与主备之间的数据差异以及网络带宽有关，但最终都会稳定在同步复制状态下。

展望

虽然经历了这么多年线上堪比工况的苛刻检验，阿里中间件消息引擎仍然存在着优化空间，如团队正尝试通过优化存储算法、跨语言调用等策略进一步降低消息低延迟存储。面对移动物联网、大数据、VR等新兴场景，面对席卷全球的开放与商业化生态，团队开始着手打造第4代消息引擎，多级协议QoS，跨网络、跨终端、跨语言支持，面向在线应用更低的响应时间，面向离线应用更高的吞吐，秉持取之于开源，回馈于开源的思想，相信RocektMQ朝着更健康的生态发展。

参考文献

[1]Ryan Barrett. http://snarfed.org/transactions_across_datacenters_io.html
[2]http://www.slideshare.net/vimal25792/leaky-bucket-tocken-buckettraffic-shaping
[3]http://systemdesigns.blogspot.com/2015/12/rate-limiter.html
[4]Little J D C, Graves S C. Little’s law[M]//Building intuition. Springer US, 2008: 81-100.
[5]https://access.redhat.com/documentation/en-US/Red_Hat_Enterprise_Linux/6/html-single/Performance_Tuning_Guide/index.html
[6]http://highscalability.com/blog/2012/3/12/google-taming-the-long-latency-tail-when-more-machines-equal.html
[7]https://www.azul.com/files/EnablingJavaInLatencySensitiveEnvs_DotCMSBootcamp_Nashville_23Oct20141.pdf