作者｜vivo互联网中间件团队-刘润云大量业务使用消息中间件进行系统间的解耦、异步、削峰填谷设计与实现。公司前期实现了一套基于RabbitMQ的高可用消息中间件平台。随着业务的不断增长，消息量也越来越大，这对消息中间件平台提出了更高的要求。此外，在运维过程中，也遇到了难以保证高可用、功能特性不足等诸多问题。基于遇到的这些问题，决定引入RocketMQ进行替换。本文将介绍基于RocketMQ构建消息中间件平台，实现线上业务无感知平滑迁移。一、背景描述vivo互联网中间件团队于2016年开始基于开源的RabbitMQ为业务提供高可用的消息中间件平台服务，为解决业务流量快速增长的问题，我们更好的交付了平台能力通过合理的业务集群拆分和动态调整，满足业务对消息中间件平台的需求。但是，随着长期业务周期的快速发展，消息量也在不断增加。RabbitMQ在高并发、大流量场景下的系统架构设计存在一定的局限性。主要问题如下：1.1高可用架构设计不足存在脑裂风险，默认情况下脑裂后不能自动恢复，人工干预存在数据丢失风险。为了解决脑裂问题，可以选择将网络异常后的处理调整为pause_minority模式，但同时也带来了一个问题，即一个小小的网络抖动也可能导致集群故障无法恢复。1.2.性能不足业务消息发送后，通过exchange路由到对应队列。每个队列实际上由集群中的一个节点承载。在高流量情况下，集群中的某个节点可能成为瓶颈。队列被某个节点承载后无法快速迁移。强制迁移到其他低负载节点可能会导致队列不可用，这也会导致集群增加节点，无法快速提升集群的流量承载能力。集群性能低。经过测试，用三台机器组成一个集群，大概可以承载几万tps，而由于队列实际上是由集群中的一个节点来承载的，不可能继续提升一个队列的性能，所以它不能支持高流量业务。当消息数累积到数千万甚至更多时，集群的性能会下降。即使在大量消息堆积后消费请求的tps特别高，也可能因为磁盘的性能损耗导致发送性能下降，恢复时间长甚至无法恢复。积累了太多的消息。.1.3功能特性不足RabbitMQ默认会对消费异常进行立即重投，少量异常消息也可能导致业务无法消费后续消息。功能特性不支持事务性消息和顺序消息功能。虽然消息追踪逻辑可以自己实现，但是会对集群造成非常大的性能损失。在正式环境下，基于RabbitMQ的原生能力，其实是不可能实现消息追踪功能的。2、消息中间件平台的项目目标基于以上问题，中间件团队于2020年Q4开始对下一代消息中间件平台方案进行研究，为确保下一代消息中间件平台满足新的业务需求，我们首先明确消息中间件平台的建设目标主要包括两部分：业务需求平台需求2.1业务需求分析高性能：可以支持极高的tps，支持横向扩展，可以快速满足流量业务增长的需要。应该成为服务请求链路性能提升的瓶颈点。高可用：平台可用性极高（>99.99%），数据可靠性极高（>99.99999999%）。丰富的功能特性：支持集群、广播消费；支持事务消息、顺序消息、延迟消息、死信消息；支持消息跟踪。2.2平台运维需求分析运维：业务使用权限验证；企业生产消费交通限制；业务流量隔离和快速迁移能力。Observable：丰富的性能指标，可以观察集群的运行情况。精通：可以基于开源组件快速进行二次开发，丰富平台功能，修复相关问题。云原生：未来可以基于容器化提供云原生的消息中间件，提供更高的弹性和扩展性。总结：需要构建一个高性能、高可靠、数据可靠性高、功能特性丰富的下一代消息中间件，并且需要完美兼容目前的RabbitMQ平台，帮助业务快速迁移到新的消息中间件平台，降低业务迁移成本。3、开源组件选择研究基于目前RabbitMQ平台存在的问题和下一代消息中间件平台的项目需求，我们对目前流行的两种消息中间件RocketMQ和Pulsar进行了研究。在研究过程中，主要对比了以下两个方面：3.1高可用能力分析对比3.1.1高可用架构与负载均衡能力对比Pulsar部署架构（来源：Pulsar社区）RocketMQ部署架构（来源：RocketMQ社区）)Pulsar：采用计算与存储分离的架构设计，可以实现海量数据存储，支持冷热数据分离存储。基于ZK和Manager节点控制Broker故障转移，实现高可用。BooKeeper采用分层分片存储设计，天然支持负载均衡。RocketMQ：采用存储计算一体化架构设计，主从模式部署，主节点异常不影响消息读取，topic采用分片设计。需要二次开发支持主从切换，实现高可用。没有实现Broker的自动负载均衡，可以将topn个流量topic分布到不同的Broker上，实现简单的负载均衡。3.1.2伸缩和故障恢复对比PulsarBroker和BooKeeper独立伸缩，伸缩完成后自动负载均衡。Broker节点是无状态的，发生故障后，承载的topic会自动转移到其他Broker节点上，完成故障的秒级恢复。BooKeeper使用自动恢复服务将账本数据对齐并恢复到设定的QW份额。acked消息在故障期间不会丢失，unacked消息需要客户端重新发送。RocketMQBroker扩缩容后，需要人工干预完成Topic流量均衡。可以结合Topic读写权限控制开发自动负载均衡组件，自动完成扩容后的负载均衡。基于主从切换实现高可用性。由于客户端每30秒定期从NameSrv更新路由，故故障恢复时间为30~60秒。客户端可以结合客户端降级策略，主动排除异常Broker节点，实现更快的故障恢复。同步复制异步刷写的部署架构在极端情况下会造成少量的消息丢失。通过同步复制和同步刷写，写入的消息不会丢失。3.1.3性能对比Pulsar可以支持百万主题，但实际上受限于ZK存储元数据。根据内部压力测试，一条1KB的消息可以支持几十万的TPS。从逻辑上讲，RocketMQ可以支持百万级的topic，但是当实际达到数万级时，Broker和NameSrv之间的心跳包传输可能会超时。建议单个集群不超过50000。据压测，可支持1KB消息体TPS可达100000+。3.2功能特点对比3.3总结从高可用架构分析，Pulsar基于Bookkeeper组件实现了该架构的计算和存储分离，可以实现故障快速恢复；RocketMQ采用主从复制架构，故障恢复依赖主从切换。从功能特性分析，Pulsar支持丰富的过期策略，支持消息去重，在实时计算中可以支持消息只消费一次的语义；RocketMQ在事务消息、消息轨迹、消费方式等方面对在线业务有更好的特性。支持。从这两方面的比较来看，最终选择了RocketMQ来搭建我们下一代的消息中间件平台。4.平滑迁移构建通过技术研究，确定基于RocketMQ构建下一代消息中间件平台。为了实现业务从RabbitMQ到RocketMQ的平滑迁移，需要搭建一个消息网关，实现消息从AMQP协议到RocketMQ的转换；RabbitMQ和RocketMQ的元数据语义和存储不同，需要实现元数据语义的映射和元数??据的独立存储。主要完成以下四项：4.1消息网关独立部署和嵌入式部署的区别4.2元数据定义映射和维护4.3互不干扰的高性能消息推送RabbitMQ采用推送方式进行消息消费，虽然RocketMQ也支持消息推送消费，但是由于AMQP协议中的prefetch参数限制了客户端缓存消息的数量，以保证客户端不会因为缓存过多的消息而导致内存异常，所以在使用时也必须满足AMQP协议的语义消息网关实现消息推送。同时，每个消息网关都需要几千甚至几万个队列来推送消息。每个队列的消息消费速率不同，每个队列都可能随时有消息需要推送给客户端消费。需要保证不同队列之间的通信。推送是无干扰且及时的。为了实现高效无干扰的消息推送，有以下策略：每个队列使用独立的线程，保证互不干扰和及时性。缺点是不能支持大量队列的消息推送。基于信号量、阻塞队列等，当感知到有可推送的消息和可消费的服务器时，按需推送消息，从而以较少的线程数完成高效的消息推送。最后选择第二个选项，数据流图如下图所示：一个消息消费流程：客户端开始连接消息网关后，会在消息网关中构建一个RocketMQ推送消费客户端实例，而自定义的ConsumeMessageService会被注入Instance，并使用一个信号量来保存允许客户端推送的消息数量。当消息从集群端推送到消息网关时，消息会根据推送的批次封装成一个任务存储在ConsumeMessageService实例的BlockingQueue中。同时，推送线程会轮询所有的ConsumeMessageService实例。可以消费消息的业务客户端向线程池提交任务，完成消息推送。为了保证其他队列的消息推送时效不会因为少数消费率特别高的队列而降低，每个ConsumeMessageService被限制为只允许推送一定数量的消息，即从其他队列推送消息，这样可以保证所有队列之间的互不干扰和消息推送的及时性。客户端消费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，超过256KB直接返回失败，流量管理未来将进行限制发送大消息体的业务流量5.2平台运维收入业务从RabbitMQ迁移到RocketMQ后，可支持业务流量从10000TPS到100000TPS，支持业务容量从几亿到几百亿。机器资源消耗降低50%以上，运维难度和成本大幅降低。同时，基于消息网关可以实现更丰富的功能和特性。6.未来展望未来，中间件团队计划从消息中间件三个方面迭代演进：基于消息网关能力，丰富现有平台的功能特性，管理业务消息。近五年来，中间件团队基于开源的RabbitMQ进行了RabbitMQ的高可用构建，发现直接让业务方使用基于开源组件的SDK接入会带来升级困难SDK与后端消息中间件类型绑定问题。我们计划实现基于GPRC和消息网关的面向服务的消息队列引擎，业务无需关心底层使用的开源消息中间件的选择。调研RocketMQ5.0计算存储分离架构，升级消息中间件架构。
