一、背景说明

vivo 在 2016 年引入 RabbitMQ，基于开源 RabbitMQ 进行扩展，向业务提供消息中间件服务。

2016~2018年，所有业务均使用一个集群，随着业务规模的增长，集群负载越来越重，集群故障频发。

2019年，RabbitMQ 进入高可用建设阶段，完成了高可用组件 MQ 名字服务以及 RabbitMQ 集群的同城双活建设。

同时进行业务使用集群的物理拆分，严格按照集群负载情况和业务流量进行业务使用集群的分配以及动态调整。

在 2019 年高可用建设后至今，业务流量增加了十倍，集群未出现过严重故障。

RabbitMQ 是实现了 AMQP 协议的开源消息代理软件，起源于金融系统。

具有丰富的特性：

消息可靠性保证，RabbitMQ 通过发送确认保证消息发送可靠、通过集群化、消息持久化、镜像队列的方式保证消息在集群的可靠、通过消费确认保证消息消费的可靠性。

RabbitMQ 提供了多种语言的客户端。

提供了多种类型的 exchange，消息发送到集群后通过exchange路由到具体的queue中。

RabbitMQ 提供了完善的管理后台和管理 API，通过管理API可以快速与自建监控系统整合。

RabbitMQ 在具体实践中发现的问题：

为保障业务高可用使用多套集群进行物理隔离，多套集群无统一平台进行管理。

原生RabbitMQ客户端使用集群地址连接，使用多套集群时业务需要关心集群地址，使用混乱。

原生RabbitMQ仅有简单的用户名/密码验证，不对使用的业务应用方进行鉴权，不同业务容易混用exchange/queue信息，造成业务应用使用异常。

使用的业务应用方较多，无平台维护消息发送方、消费方的关联信息，多个版本迭代后无法确定对接方。

客户端无限流，业务突发异常流量冲击甚至击垮集群。

客户端无异常消息重发策略，需要使用方实现。

集群出现内存溢出等造成集群阻塞时无法快速自动转移到其它可用集群。

使用镜像队列，队列的master节点会落在具体某个节点上，在集群队列数较多时，容易出现节点负载不均衡的情况。

RabbitMQ无队列自动平衡能力，在队列较多时容易出现集群节点负载不均问题。

二、整体架构

vivo 基于原生 RabbitMQ 的高可用架构实践

1、MQ-Portal--支持应用使用申请

过往业务团队适用RabbitMQ时，应用申请的流量以及对接的应用等信息都在线下表格记录，较为零散，更新不及时，无法准确了解业务当前真实的使用情况，因此通过一个接入申请的流程可视化、平台化建立应用使用的元数据信息。

vivo 基于原生 RabbitMQ 的高可用架构实践

通过MQ-Portal的申请流程（如上图），确定了消息发送应用、消费应用、使用exchange/queue、发送流量等信息使用申请提交后将进入vivo内部工单流程进行审批。

vivo 基于原生 RabbitMQ 的高可用架构实践

工单流程审批通过后，通过工单的接口回调，分配应用具体使用的集群，并在集群上创建exchange/queue已经绑定关系。

由于采用多集群物理隔离的方式保证业务在正式环境的高可用，无法简单通过一个exchange/queue的名称定位到使用的集群。

每一个exchange/queue与集群之间通过唯一的一对rmq.topic.key与rmq.secret.key进行关联，这样SDK启动过程中即可定位到具体使用的集群。

rmq.topic.key与rmq.secret.key将在工单的回调接口中进行分配。

vivo 基于原生 RabbitMQ 的高可用架构实践

2、客户端SDK能力概述

客户端SDK基于spring-message和spring-rabbit进行封装，并在此基础上提供了应用使用鉴权、集群寻址、客户端限流、生产消费重置、阻塞转移等能力。

2.1、应用使用鉴权

开源RabbitMQ仅通过用户名密码的方式判断是否允许连接集群，但是应用是否允许使用exchange/queue是未进行校验的。

为了避免不同业务混用exchange/queue，需要对应用进行使用鉴权。

应用鉴权由SDK和MQ-NameServer协同完成。

应用启动时首先会上报应用配置的rmq.topic.key信息到MQ-NameServer，由MQ-NameServer判断使用应用与申请应用是否一致，并且在SDK发送消息过程中还会进行二次校验。

/**
  * 发送前校验，并且获取真正的发送factory，这样业务可以声明多个，
  * 但是用其中一个bean就可以发送所有的消息，并且不会导致任何异常
  * @param exchange 校验参数
  * @return 发送工厂
*/
public AbstractMessageProducerFactory beforeSend(String exchange) {
    if(closed || stopped){
        //上下文已经关闭抛出异常，阻止继续发送，减少发送临界状态数据
        throw new RmqRuntimeException(String.format("producer sending message to exchange %s has closed, can't send message", this.getExchange()));
    }
    if (exchange.equals(this.exchange)){
        return this;
    }
    if (!VIVO_RMQ_AUTH.isAuth(exchange)){
        throw new VivoRmqUnAuthException(String.format("发送topic校验异常，请勿向无权限exchange %s 发送数据，发送失败", exchange));
    }
    //获取真正的发送的bean，避免发送错误
    return PRODUCERS.get(exchange);
}

2.2、集群寻址

前文说过，应用使用RabbitMQ严格按照集群的负载情况和业务流量进行集群的分配，因此具体某个应用使用的的不同的exchange/queue可能是分配在不同的集群上的。

为了提升业务的开发效率，需要屏蔽多集群对业务的影响，因此按照应用配置的rmq.topic.key信息进行集群的自动寻址。

2.3、客户端限流

原生SDK客户端不进行发送流量限流，在部分应用存在异常持续向MQ发送消息时，可能会冲垮MQ集群。并且一个集群为多应用共同使用，单一应用造成集群影响将会影响使用异常集群的所有应用。

因此需要在SDK中提供客户端限流的能力，必要时可以限制应用向集群发送消息，保障集群的稳定。

2.4、生产消费重置

（1）随着业务规模增长，集群负载持续增加，此时需要进行集群的业务拆分。为了减少在拆分过程中避免业务重启，需要有生产消费重置功能。

（2）集群出现异常，可能会造成消费者掉线，此时通过生产消费重置可以快速拉起业务消费。

为了实现生产消费重置，需要实现一下流程：

重置连接工厂连接参数

重置连接

建立新的连接

重新启动生产消费

CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
connectionFactory.setAddresses(address);
connectionFactory.resetConnection();
rabbitAdmin = new RabbitAdmin(connectionFactory);
rabbitTemplate = new RabbitTemplate(connectionFactory);

同时MQ-SDK中有异常消息重发策略，可以避免在生产重置过程中导致的消息发送异常。