一、背景說(shuō)明

vivo 在 2016 年引入 RabbitMQ，基于開(kāi)源 RabbitMQ 進(jìn)行擴(kuò)展，向業(yè)務(wù)提供消息中間件服務(wù)。

2016~2018年，所有業(yè)務(wù)均使用一個(gè)集群，隨著業(yè)務(wù)規(guī)模的增長(zhǎng)，集群負(fù)載越來(lái)越重，集群故障頻發(fā)。

2019年，RabbitMQ 進(jìn)入高可用建設(shè)階段，完成了高可用組件 MQ 名字服務(wù)以及 RabbitMQ 集群的同城雙活建設(shè)。

同時(shí)進(jìn)行業(yè)務(wù)使用集群的物理拆分，嚴(yán)格按照集群負(fù)載情況和業(yè)務(wù)流量進(jìn)行業(yè)務(wù)使用集群的分配以及動(dòng)態(tài)調(diào)整。

在 2019 年高可用建設(shè)后至今，業(yè)務(wù)流量增加了十倍，集群未出現(xiàn)過(guò)嚴(yán)重故障。

RabbitMQ 是實(shí)現(xiàn)了 AMQP 協(xié)議的開(kāi)源消息代理軟件，起源于金融系統(tǒng)。

具有豐富的特性：

•  消息可靠性保證，RabbitMQ 通過(guò)發(fā)送確認(rèn)保證消息發(fā)送可靠、通過(guò)集群化、消息持久化、鏡像隊(duì)列的方式保證消息在集群的可靠、通過(guò)消費(fèi)確認(rèn)保證消息消費(fèi)的可靠性；
•  RabbitMQ 提供了多種語(yǔ)言的客戶(hù)端；
•  提供了多種類(lèi)型的 exchange，消息發(fā)送到集群后通過(guò)exchange路由到具體的queue中；
•  RabbitMQ 提供了完善的管理后臺(tái)和管理 API，通過(guò)管理API可以快速與自建監(jiān)控系統(tǒng)整合。

RabbitMQ 在具體實(shí)踐中發(fā)現(xiàn)的問(wèn)題：

•  為保障業(yè)務(wù)高可用使用多套集群進(jìn)行物理隔離，多套集群無(wú)統(tǒng)一平臺(tái)進(jìn)行管理；
•  原生RabbitMQ客戶(hù)端使用集群地址連接，使用多套集群時(shí)業(yè)務(wù)需要關(guān)心集群地址，使用混亂；
•  原生RabbitMQ僅有簡(jiǎn)單的用戶(hù)名/密碼驗(yàn)證，不對(duì)使用的業(yè)務(wù)應(yīng)用方進(jìn)行鑒權(quán)，不同業(yè)務(wù)容易混用exchange/queue信息，造成業(yè)務(wù)應(yīng)用使用異常；
•  使用的業(yè)務(wù)應(yīng)用方較多，無(wú)平臺(tái)維護(hù)消息發(fā)送方、消費(fèi)方的關(guān)聯(lián)信息，多個(gè)版本迭代后無(wú)法確定對(duì)接方；
•  客戶(hù)端無(wú)限流，業(yè)務(wù)突發(fā)異常流量沖擊甚至擊垮集群；
•  客戶(hù)端無(wú)異常消息重發(fā)策略，需要使用方實(shí)現(xiàn)；
•  集群出現(xiàn)內(nèi)存溢出等造成集群阻塞時(shí)無(wú)法快速自動(dòng)轉(zhuǎn)移到其它可用集群；
•  使用鏡像隊(duì)列，隊(duì)列的master節(jié)點(diǎn)會(huì)落在具體某個(gè)節(jié)點(diǎn)上，在集群隊(duì)列數(shù)較多時(shí)，容易出現(xiàn)節(jié)點(diǎn)負(fù)載不均衡的情況；
•  RabbitMQ無(wú)隊(duì)列自動(dòng)平衡能力，在隊(duì)列較多時(shí)容易出現(xiàn)集群節(jié)點(diǎn)負(fù)載不均問(wèn)題。

二、整體架構(gòu)

1、MQ-Portal--支持應(yīng)用使用申請(qǐng)

過(guò)往業(yè)務(wù)團(tuán)隊(duì)適用RabbitMQ時(shí)，應(yīng)用申請(qǐng)的流量以及對(duì)接的應(yīng)用等信息都在線下表格記錄，較為零散，更新不及時(shí)，無(wú)法準(zhǔn)確了解業(yè)務(wù)當(dāng)前真實(shí)的使用情況，因此通過(guò)一個(gè)接入申請(qǐng)的流程可視化、平臺(tái)化建立應(yīng)用使用的元數(shù)據(jù)信息。

通過(guò)MQ-Portal的申請(qǐng)流程（如上圖），確定了消息發(fā)送應(yīng)用、消費(fèi)應(yīng)用、使用exchange/queue、發(fā)送流量等信息使用申請(qǐng)?zhí)峤缓髮⑦M(jìn)入vivo內(nèi)部工單流程進(jìn)行審批。

工單流程審批通過(guò)后，通過(guò)工單的接口回調(diào)，分配應(yīng)用具體使用的集群，并在集群上創(chuàng)建exchange/queue已經(jīng)綁定關(guān)系。

由于采用多集群物理隔離的方式保證業(yè)務(wù)在正式環(huán)境的高可用，無(wú)法簡(jiǎn)單通過(guò)一個(gè)exchange/queue的名稱(chēng)定位到使用的集群。

每一個(gè)exchange/queue與集群之間通過(guò)唯一的一對(duì)rmq.topic.key與rmq.secret.key進(jìn)行關(guān)聯(lián)，這樣SDK啟動(dòng)過(guò)程中即可定位到具體使用的集群。

rmq.topic.key與rmq.secret.key將在工單的回調(diào)接口中進(jìn)行分配。

2、客戶(hù)端SDK能力概述

客戶(hù)端SDK基于spring-message和spring-rabbit進(jìn)行封裝，并在此基礎(chǔ)上提供了應(yīng)用使用鑒權(quán)、集群尋址、客戶(hù)端限流、生產(chǎn)消費(fèi)重置、阻塞轉(zhuǎn)移等能力。

1）應(yīng)用使用鑒權(quán)

開(kāi)源RabbitMQ僅通過(guò)用戶(hù)名密碼的方式判斷是否允許連接集群，但是應(yīng)用是否允許使用exchange/queue是未進(jìn)行校驗(yàn)的。

為了避免不同業(yè)務(wù)混用exchange/queue，需要對(duì)應(yīng)用進(jìn)行使用鑒權(quán)。

應(yīng)用鑒權(quán)由SDK和MQ-NameServer協(xié)同完成。

應(yīng)用啟動(dòng)時(shí)首先會(huì)上報(bào)應(yīng)用配置的rmq.topic.key信息到MQ-NameServer，由MQ-NameServer判斷使用應(yīng)用與申請(qǐng)應(yīng)用是否一致，并且在SDK發(fā)送消息過(guò)程中還會(huì)進(jìn)行二次校驗(yàn)。 

/**  
  * 發(fā)送前校驗(yàn)，并且獲取真正的發(fā)送factory，這樣業(yè)務(wù)可以聲明多個(gè)，  
  * 但是用其中一個(gè)bean就可以發(fā)送所有的消息，并且不會(huì)導(dǎo)致任何異常  
  * @param exchange 校驗(yàn)參數(shù)  
  * @return 發(fā)送工廠  
*/  
public AbstractMessageProducerFactory beforeSend(String exchange) { 
    if(closed || stopped){  
        //上下文已經(jīng)關(guān)閉拋出異常，阻止繼續(xù)發(fā)送，減少發(fā)送臨界狀態(tài)數(shù)據(jù)  
        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("發(fā)送topic校驗(yàn)異常，請(qǐng)勿向無(wú)權(quán)限exchange %s 發(fā)送數(shù)據(jù)，發(fā)送失敗", exchange));  
    }  
    //獲取真正的發(fā)送的bean，避免發(fā)送錯(cuò)誤  
    return PRODUCERS.get(exchange);  
} 

2）集群尋址

前文說(shuō)過(guò)，應(yīng)用使用RabbitMQ嚴(yán)格按照集群的負(fù)載情況和業(yè)務(wù)流量進(jìn)行集群的分配，因此具體某個(gè)應(yīng)用使用的的不同的exchange/queue可能是分配在不同的集群上的。

為了提升業(yè)務(wù)的開(kāi)發(fā)效率， 需要屏蔽多集群對(duì)業(yè)務(wù)的影響，因此按照應(yīng)用配置的rmq.topic.key信息進(jìn)行集群的自動(dòng)尋址。

3）客戶(hù)端限流

原生SDK客戶(hù)端不進(jìn)行發(fā)送流量限流，在部分應(yīng)用存在異常持續(xù)向MQ發(fā)送消息時(shí)，可能會(huì)沖垮MQ集群。并且一個(gè)集群為多應(yīng)用共同使用，單一應(yīng)用造成集群影響將會(huì)影響使用異常集群的所有應(yīng)用。

因此需要在SDK中提供客戶(hù)端限流的能力，必要時(shí)可以限制應(yīng)用向集群發(fā)送消息，保障集群的穩(wěn)定。

4）生產(chǎn)消費(fèi)重置

①隨著業(yè)務(wù)規(guī)模增長(zhǎng)，集群負(fù)載持續(xù)增加，此時(shí)需要進(jìn)行集群的業(yè)務(wù)拆分。為了減少在拆分過(guò)程中避免業(yè)務(wù)重啟，需要有生產(chǎn)消費(fèi)重置功能。

②集群出現(xiàn)異常，可能會(huì)造成消費(fèi)者掉線，此時(shí)通過(guò)生產(chǎn)消費(fèi)重置可以快速拉起業(yè)務(wù)消費(fèi)。

為了實(shí)現(xiàn)生產(chǎn)消費(fèi)重置，需要實(shí)現(xiàn)一下流程：

•  重置連接工廠連接參數(shù)；
•  重置連接；
•  建立新的連接；
•  重新啟動(dòng)生產(chǎn)消費(fèi)。 

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

同時(shí)MQ-SDK中有異常消息重發(fā)策略，可以避免在生產(chǎn)重置過(guò)程中導(dǎo)致的消息發(fā)送異常。

5）阻塞轉(zhuǎn)移

RabbitMQ在內(nèi)存使用超過(guò)40%，或是磁盤(pán)使用超限制時(shí)會(huì)阻塞消息發(fā)送。

由于vivo中間件團(tuán)隊(duì)已經(jīng)完成了RabbitMQ同城雙活的建設(shè)，因此在出現(xiàn)一個(gè)集群發(fā)送阻塞時(shí)可以通過(guò)生產(chǎn)消費(fèi)重置到雙活集群完成阻塞的快速轉(zhuǎn)移。

6）多集群調(diào)度

隨著應(yīng)用的發(fā)展，單集群將無(wú)法滿(mǎn)足應(yīng)用的流量需求，并且集群隊(duì)列均為鏡像隊(duì)列，無(wú)法簡(jiǎn)單的通過(guò)增加集群節(jié)點(diǎn)的方式實(shí)現(xiàn)業(yè)務(wù)支撐流量單集群的水平擴(kuò)容。

因此需要SDK支持多集群調(diào)度能力，通過(guò)將流量分散到多個(gè)集群上滿(mǎn)足業(yè)務(wù)大流量需求。

3、MQ-NameServer--支持MQ-SDK實(shí)現(xiàn)故障快速切換

MQ-NameServer為無(wú)狀態(tài)服務(wù)，通過(guò)集群部署即可保障自身高可用，主要用于解決以下問(wèn)題：

•  MQ-SDK啟動(dòng)鑒權(quán)以及應(yīng)用使用集群定位；
•  處理MQ-SDK的定時(shí)指標(biāo)上報(bào)（消息發(fā)送數(shù)量、消息消費(fèi)數(shù)量），并且返回當(dāng)前可用集群地址，確保SDK在集群異常時(shí)按照正確地址進(jìn)行重連；
•  控制MQ-SDK進(jìn)行生產(chǎn)消費(fèi)重置。

4、MQ-Server高可用部署實(shí)踐

RabbitMQ 集群均采用同城雙活部署架構(gòu)，依靠MQ-SDK和MQ-NameServer提供的集群尋址、故障快速切換等能力保障集群的可用性。

1）集群腦裂問(wèn)題處理

RabbitMQ官方提供了三種集群腦裂恢復(fù)策略。

①ignore

忽略腦裂問(wèn)題不處理，在出現(xiàn)腦裂時(shí)需要進(jìn)行人為干預(yù)才可恢復(fù)。由于需要人為干預(yù)，可能會(huì)造成部分消息丟失，在網(wǎng)絡(luò)非常可靠的情況可以使用。

②pause_minority

節(jié)點(diǎn)在與超過(guò)半數(shù)集群節(jié)點(diǎn)失聯(lián)時(shí)將會(huì)自動(dòng)暫停，直到檢測(cè)到與集群超半數(shù)節(jié)點(diǎn)的通信恢復(fù)。極端情況下集群內(nèi)所有節(jié)點(diǎn)均暫停，造成集群不可用。

③autoheal

少數(shù)派節(jié)點(diǎn)將自動(dòng)重啟，此策略主要用于優(yōu)先保證服務(wù)的可用性，而不是數(shù)據(jù)的可靠性，因?yàn)橹貑⒐?jié)點(diǎn)上的消息會(huì)丟失。

由于RabbitMQ集群均為同城雙活部署，即使單集群異常業(yè)務(wù)流量也可自動(dòng)遷移到雙活機(jī)房集群，因此選擇使用了pause_minority策略避免腦裂問(wèn)題。

2018年多次因網(wǎng)絡(luò)抖動(dòng)造成集群腦裂，在修改集群腦裂恢復(fù)策略后，已未再出現(xiàn)腦裂問(wèn)題。

2）集群高可用方案

RabbitMQ采用集群化部署，并且因?yàn)榧耗X裂恢復(fù)策略采用pause_minority模式，每個(gè)集群要求至少3個(gè)節(jié)點(diǎn)。

推薦使用5或7節(jié)點(diǎn)部署高可用集群，并且控制集群隊(duì)列數(shù)量。

集群隊(duì)列均為鏡像隊(duì)列，確保消息存在備份，避免節(jié)點(diǎn)異常導(dǎo)致消息丟失。

exchange、queue、消息均設(shè)置為持久化，避免節(jié)點(diǎn)異常重啟消息丟失。

隊(duì)列均設(shè)置為lazy queues，減少節(jié)點(diǎn)內(nèi)存使用的波動(dòng)。

3）同城雙活建設(shè)

雙機(jī)房部署等價(jià)集群，并且通過(guò)Federation插件將雙集群組成聯(lián)盟集群。

本機(jī)房應(yīng)用機(jī)器優(yōu)先連接本機(jī)房MQ集群，避免因?qū)＞€抖動(dòng)造成應(yīng)用使用異常。

通過(guò)MQ-NameServer心跳獲取最新的可用集群信息，異常時(shí)重連到雙活集群中，實(shí)現(xiàn)應(yīng)用功能的快速恢復(fù)。

三、未來(lái)挑戰(zhàn)與展望

目前對(duì)RabbitMQ的使用增強(qiáng)主要在MQ-SDK和MQ-NameServer側(cè)，SDK實(shí)現(xiàn)較為復(fù)雜，后期希望可以構(gòu)建消息中間件的代理層，可以簡(jiǎn)化SDK并且對(duì)業(yè)務(wù)流量做更加細(xì)致化的管理。