作者：vivo 互联网中间件团队- Liu Runyun

大量业务使用消息中间件进行系统间的解耦、异步化、削峰填谷设计实现。公司内部前期基于 RabbitMQ 实现了一套高可用的消息中间件平台。随着业务的持续增长，消息体量随之增大，对消息中间件平台提出了更高的要求，此外在运维过程中也遇到了高可用难以保障，功能特性不足等诸多问题。基于遇到的这些问题，决定引入 RocketMQ 进行替换。本文将介绍基于 RocketMQ 建设消息中间件平台并实现在线业务无感知的平滑迁移。

一、背景说明

vivo 互联网中间件团队于 2016 年开始基于开源 RabbitMQ 向业务提供高可用消息中间件平台服务。

为解决好业务流量快速增长的问题，我们通过合理的业务集群拆分和动态调整，较好的交付了业务对消息中间件平台的平台能力需求。

但是随着业务长周期的迅猛发展，消息体量也越来越大，在高并发、大流量场景下 RabbitMQ 的系统架构设计存在着一定的限制，主要有以下问题：

1.1 高可用能力不足

架构设计存在脑裂风险，并且默认脑裂后无法自动恢复，人工介入恢复存在数据丢失的风险。

为解决脑裂问题，可以选择将网络异常后的处理调整为 pause_minority 模式，但是也带来了可能微小的网络抖动也会导致集群故障无法恢复的问题。

1.2. 性能不足

业务消息发送后通过 exchange 路由到对应的 queue 中，每一个 queue 由集群中的某个节点实际承载流量，高流量下集群中的某个节点可能会成为瓶颈。

queue 由某个节点承载流量后无法快速迁移，强制迁移到其它低负载节点可能会导致 queue 不可用，这也导致了向集群中添加节点并无法快速提升集群的流量承载能力。

集群性能较低，经测试使用三台机器组成集群，可承载大概数万 tps 左右，并且由于 queue 是由集群中某个节点实际承载的，也无法继续提升某个 queue 的性能，这样就无法支撑大流量业务。

消息堆积到千万或更多后会导致集群性能下降，甚至海量堆积后如果消费请求 tps 特别高，可能会因为磁盘的性能损耗导致发送性能下降，并且在消息堆积太多时恢复时间长甚至无法恢复。

1.3 功能特性不足

RabbitMQ 默认情况下消费异常会执行立即重新投递，少量的异常消息也可能导致业务无法消费后续消息。

功能特性上未支持事务消息、顺序消息功能。

虽可自行实现消息轨迹逻辑，但是会对集群产生非常大的性能损耗，在正式环境中实际无法基于 RabbitMQ 原生的能力实现消息轨迹功能。

二、消息中间件平台的项目目标

基于以上问题，中间件团队于 2020 年 Q4 开始进行了下一代消息中间件平台方案的调研，为保证下一代消息中间件平台符合业务新的需求，我们首先明确了消息中间件平台的建设目标，主要包含两部分：

• 业务需求

• 平台需求

2.1 业务需求分析

高性能：可支撑极高的 tps，并且支持水平扩展，可快速满足业务的流量增长需求，消息中间件不应成为业务请求链路性能提升的瓶颈点。

高可用：极高的平台可用性(>99.99%)，极高的数据可靠性(>99.99999999%)。

丰富的功能特性：支持集群、广播消费；支持事务消息、顺序消息、延时消息、死信消息；支持消息轨迹。

2.2 平台运维需求分析

• 可运维：业务使用权限校验；业务生产消费流量限制；业务流量隔离与快速迁移能力。

• 可观测：丰富的性能指标观察集群的运行情况。

• 可掌握：可基于开源组件快速进行二次开发，丰富平台功能特性和进行相关问题修复。

• 云原生：后续可基于容器化提供云原生消息中间件，提供更高的弹性和可伸缩能力。

• 总结：需要建设高性能、高可靠的下一代消息中间件，具备极高的数据可靠性，丰富的功能特性，并且需要完美兼容当前的 RabbitMQ 平台，帮助业务快速迁移到新消息中间件平台，减少业务迁移成本。

三、开源组件选型调研

基于当前 RabbitMQ 平台的问题和对下一代消息中间件平台的项目需求，我们开展了针对当前较流行的两款消息中间件：RocketMQ、Pulsar 的调研。

调研过程中主要针对以下两方面进行对比：

3.1 高可用能力分析对比

 3.1.1 高可用架构与负载均衡能力对比

 Pulsar 部署架构（来源：Pulsar社区）

RocketMQ 部署架构（来源：RocketMQ社区）

• Pulsar：

• 采用计算与存储分离架构设计，可以实现海量数据存储，并且支持冷热数据分离存储。

• 基于 ZK 和 Manager 节点控制 Broker 的故障切换以实现高可用。

• Zookeeper 采用分层分片存储设计，天然支持负载均衡。

• RocketMQ：

• 采用存算一体架构设计，主从模式部署，master 节点异常不影响消息读取，Topic 采用分片设计。

• 需要二次开发支持主从切换实现高可用。

• 未实现 Broker 的自动负载均衡，可以将 top n 流量 Topic 分布到不同的 Broker 中实现简单的负载均衡。

 3.1.2 扩缩容与故障恢复对比

• Pulsar

• Broker 与 BooKeeper 独立扩缩容，并且扩缩容后会完成自动负载均衡。

• Broker 节点无状态，故障后承载 Topic 会自动转移到其它 Broker 节点，完成故障秒级恢复。

• BooKeeper 由自动恢复服务进行 ledger 数据对齐，并恢复到设置的 QW 份。

• 故障期间已 ack 消息不会丢失，未 ack 消息需要客户端重发。

• RocketMQ

• Broker 扩缩容后需要人工介入完成 Topic 流量均衡，可开发自动负载均衡组件结合 Topic 的读写权限控制自动化完成扩缩容后的负载均衡。

• 基于主从切换实现高可用，由于客户端定期 30 秒从 NameSrv 更新路由，因此故障恢复时间在 30~60 秒，可以结合客户端降级策略让客户端主动剔除异常 Broker 节点，实现更快故障恢复。

• 采用同步复制异步刷盘部署架构，在极端情况下会造成少量消息丢失，采用同步复制同步刷盘，已写入消息不会丢失。

 3.1.3 性能对比

• Pulsar

• 可支撑百万 Topic 数量，实际受到 ZK 存储元数据限制。

• 根据内部压测 1KB 消息可支撑 TPS 达数十万。

• RocketMQ

• 逻辑上可支撑百万 Topic，实际在达到数万时 Broker 与 NameSrv 传输心跳包可能超时，建议单集群不超过 5 万。

• 根据压测可支撑 1KB 消息体 TPS 达 10 万+。

3.2 功能特性对比

3.3  总结

从高可用架构分析，Pulsar 基于 Bookeeper 组件实现了架构的计算与存储分离，可以实现故障的快速恢复；RocketMQ 采用了主从复制的架构，故障恢复依赖主从切换。

从功能特性分析，Pulsar 支持了丰富的过期策略，支持了消息去重，可以支持实时计算中消息只消费一次的语义；RocketMQ 在事务消息、消息轨迹、消费模式等特性对在线业务有更好的支持。

从这两方面对比，最终选择了 RocketMQ 构建我们下一代的消息中间件平台。

四、平滑迁移建设

通过技术调研，确定了基于 RocketMQ 建设下一代消息中间件平台。

为了实现业务从 RabbitMQ 平滑迁移到 RocketMQ，就需要建设消息网关实现消息从 AMQP 协议转换到 RocketMQ；RabbitMQ 与 RocketMQ 的元数据语义与存储存在差异，需要实现元数据语义的映射与元数据的独立存储。

主要有以下四个事项需要完成：

4.1 消息网关独立部署与嵌入式部署差异对比

4.2 元数据定义映射与维护

4.3 互不干扰的高性能消息推送

RabbitMQ 采用推模式进行消息消费，虽然 RocketMQ 也支持消息推送消费，但是因为 AMQP 协议中通过 prefetch 参数限制了客户端缓存消息数量以保证不会因缓存太多消息导致客户端内存异常，因此在消息网关实现消息推送时也需要满足 AMQP 协议的语义。

同时每个消息网关都需要数千甚至数万的 queue 的消息推送，每个 queue 消息消费速率存在差异，并且每个队列可能随时有消息需要推送到客户端进行消费，要保证不同 queue 之间的推送互不干扰且及时。

为了实现高效的、互不干扰的消息推送，有以下策略：

1. 每个 queue 采用独立的线程，保证互不干扰和时效性，缺点是无法支撑海量 queue 的消息推送。

2. 基于信号量、阻塞队列等，在感知到有可推送消息和可消费服务端时按需进行消息的推送，这样可使用少量的线程即可完成高效的消息推送。

最终选择了第 2 种方案，数据流转图如下图所示：

一个消息消费过程：客户端在启动连接到消息网关后，在消息网关中会构建 RocketMQ 推送消费客户端实例，并且注入自定义的 ConsumeMessageService 实例，同时使用一个信号量保存客户端允许推送的消息数量。

当消息从集群侧推送到消息网关时，将消息按照推送的批次封装为一个任务保存在 ConsumeMessageService 实例的 BlockingQueue 中，同时推送线程会轮询所有的 ConsumeMessageService 实例，如果发现本地缓存有待消费的消息并且有可消费消息的业务客户端，将任务提交到线程池中完成消息的推送。

为了保证不会因为少量消费速率特别高的 queue 导致其它 queue 的消息推送时效性降低，会限制每一个 ConsumeMessageService 只允许推送一定数量的消息即转到推送其它 queue 的消息，以此即可保证所有 queue 的消息推送的互不干扰和时效性。

在客户端消费 ack/uack 后再次通过信号量通知下一次推送，这样也保证了使用少量的线程资源即可完成海量消息的推送需求。

4.4 消费启停与消费限流能力实现

基于消息网关，可以在消息推送逻辑中增加消费启停和消费限流逻辑。

消费启停可以帮助业务快速实现消费的暂停或是部分异常节点停止消息消费。

消费限流可以帮助业务控制消息消费速率，避免对底层依赖产生太大压力。

4.5 平台架构

• 最终形成了以上的平台架构。新建设了一个 AMQP-proxy 消息网关服务实现 AMQP 消息转换到 RocketMQ，支持业务的消息生产消费。

• 建设了 mq-meta 服务维护集群的元数据信息。

• 通过 mq-controller 控制集群的主从切换，实现集群的高可用，同时增加了集群监控，负载均衡模块保障集群的高可用。

五、平台建设进展与迁移收益

5.1 业务使用收益

 5.1.1 更高、更稳定的消息发送性能

原生 RabbitMQ 集群业务压测性能

使用消息网关后业务压测性能

 5.1.2 更丰富的功能特性

• 统一的消息过期时间

• 消费异常消息将按照梯度延时重投递

• 直接支持广播消费模式

• 全环境按需提供消息轨迹功能

• 支持消费重置到以前的某个位点

 5.1.3 业务使用特性变化

• 消息将不再无限期保留，默认保留 3~7 天（实际保留时间根据集群配置决定）

• 消费异常将不再立即重投递，将按照一定的梯度延时重投递，多次异常后将变为死信消息

• 直接支持广播消费，注意广播消费模式消费无异常重投递，每个消息每个节点只消费一次

• 业务生产消费性能可支持水平扩展

• 不支持消费优先级功能

• 默认消费超时时间 15 分钟，消费超时后消息重新投递，消费超时时间可按需调整

• 支持消费启停（全局或限制部分节点消费）

• 支持全局消费限流

• 限制消息体大小，当前限制为 256KB，超过将直接返回失败，后续将进行流量治理，限制发送大消息体业务流量

5.2 平台运维收益

业务从 RabbitMQ 迁移到 RocketMQ 后，可支撑业务流量从万 TPS 级别提升到十万 TPS 级别，可支撑业务容量从数亿提升至百亿级别。耗用机器资源下降 50%以上，运维难度和成本均大大降低，同时可以基于消息网关实现更加丰富的功能特性。

六、未来展望

未来，中间件团队计划在三个方面对消息中间件进行迭代演进：

1. 基于消息网关能力丰富现有平台功能特性，进行业务消息治理。

2. 过去五年中间件团队基于开源 RabbitMQ 进行了 RabbitMQ 的高可用建设，发现直接让业务方使用基于开源组件的 SDK 接入会带来 SDK 升级困难，与后端消息中间件类型绑定的问题，未来我们计划基于 GPRC 和消息网关，实现消息队列引擎服务化，业务无需关心底层具体使用的开源消息中间件选型。

3. 调研 RocketMQ5.0 计算与存储分离构架，进行消息中间件架构的再升级。