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