高并發(fā)條件下消息隊(duì)列的設(shè)計(jì)與實(shí)現(xiàn)

李方方,周亞鳳,王校建,胡 蕊

(南京信息職業(yè)技術(shù)學(xué)院,南京 210043)

近年來(lái),互聯(lián)網(wǎng)用戶爆發(fā)式增加,這對(duì)行業(yè)軟件系統(tǒng)的高并發(fā)及高處理能力提出了挑戰(zhàn),特別是醫(yī)藥、電子商務(wù)、通信、金融、游戲等行業(yè),用戶基數(shù)大,與系統(tǒng)互動(dòng)頻繁,會(huì)形成巨大的請(qǐng)求數(shù)據(jù),原來(lái)的集群分布式方法已經(jīng)無(wú)法很好地滿足當(dāng)前用戶的需求,受限于網(wǎng)絡(luò)等因素,如果不具有高并發(fā)處理能力,很有可能造成通信請(qǐng)求數(shù)量的損失。為了應(yīng)對(duì)各行業(yè)業(yè)務(wù)量的持續(xù)增長(zhǎng),需在現(xiàn)有系統(tǒng)架構(gòu)上進(jìn)行技術(shù)變革。研究表明,強(qiáng)大高效的及時(shí)通信消息處理能力能減輕服務(wù)器壓力,是保證高并發(fā)、高性能、高可靠的手段之一,因此需對(duì)消息通信進(jìn)行處理,升級(jí)解決方案,引入消息即時(shí)通信技術(shù),替代傳統(tǒng)的HTTP請(qǐng)求預(yù)處理,保證系統(tǒng)吞吐量,滿足系統(tǒng)的高并發(fā)高處理需求[1]。

研發(fā)設(shè)計(jì)了一個(gè)即時(shí)消息隊(duì)列FineMQ,通過(guò)即時(shí)消息列隊(duì)利用中間件的處理能力,解決應(yīng)用解耦及消息異步,實(shí)現(xiàn)高性能及最終的一致性架構(gòu)。

1 消息隊(duì)列

消息隊(duì)列的英文是Message Queue(簡(jiǎn)稱MQ),是一種不同進(jìn)程間或同一進(jìn)程中不同線程間的通信方式,提供了標(biāo)準(zhǔn)的一步通信協(xié)議,使用進(jìn)程間通信交互方式,每一個(gè)軟件的貯列記錄都包含詳細(xì)字段,通過(guò)字段進(jìn)行通信,實(shí)現(xiàn)應(yīng)用間的異步通信,保證應(yīng)用間的高可用,通過(guò)消息組件中的Producer、Broker及Consumer達(dá)到消息的正常消費(fèi)及通信的及時(shí)性與準(zhǔn)確性[2]。

消息隊(duì)列的特點(diǎn)是異步、解耦及廣播。消息隊(duì)列本身是異步的,只需要保證同一時(shí)刻Producer和Consumer有一個(gè)在線即可。Consumer不關(guān)心業(yè)務(wù)的處理流程,只關(guān)心業(yè)務(wù)的處理結(jié)果,需要將Consumer操作結(jié)果實(shí)時(shí)反饋給Consumer,至于是否處理,不是Consumer要關(guān)心的部分。消息共享只需要將消息發(fā)布在一個(gè)固定的平臺(tái),有新Consumer想要看Producer發(fā)布的消息,只需要訂閱這個(gè)平臺(tái)就能看到Producer發(fā)布的消息,不需要每次都重新發(fā)布已經(jīng)發(fā)過(guò)的消息。

1.1 消息

需要兩個(gè)或多個(gè)通信終端/軟件/應(yīng)用參與對(duì)接,其中需要傳遞的信息內(nèi)容稱為消息。消息的定義范疇很廣泛,可以是有聲音頻,也可以是轉(zhuǎn)化的二進(jìn)制,如視頻傳輸、文本傳輸、圖像傳輸、數(shù)字傳輸?shù)?。還有可能含有時(shí)間戳等信息,為了鑒別消息發(fā)送與接受的時(shí)效性,有時(shí)還會(huì)引入sign的鑒權(quán),保證消息發(fā)送與接收者的絕對(duì)安全。

1.2 隊(duì)列

列隊(duì)是一種常用的數(shù)據(jù)結(jié)構(gòu),是存在于內(nèi)存或磁盤中、開(kāi)辟一塊公用存儲(chǔ)空間存放數(shù)據(jù)的一組數(shù)據(jù)結(jié)構(gòu),用來(lái)處理一組數(shù)據(jù)的臨時(shí)空間。為了保證數(shù)據(jù)的有序處理,將所有數(shù)據(jù)存放在列隊(duì)中,在列隊(duì)中依次讀取操作,保證系統(tǒng)處理數(shù)據(jù)的完整性。列隊(duì)一般分為兩種模式,即順序結(jié)構(gòu)和鏈?zhǔn)浇Y(jié)構(gòu)。不同模式的數(shù)據(jù)處理性能有所不同,而消息隊(duì)列是用來(lái)處理海量消息的一種方式。隊(duì)列的排隊(duì)機(jī)制主要分為4種,即先進(jìn)先出(FIFO)、優(yōu)先排隊(duì)、公平排隊(duì)、加權(quán)公平排隊(duì)。

2 消息隊(duì)列的對(duì)比和選擇

目前市面上主流的消息隊(duì)列有很多[3],如2006年發(fā)布的Rabbit MQ,2011年發(fā)布的Kafka,2012年初發(fā)布的RocketMQ及2018年雅虎生產(chǎn)的Apache Pulsar,這些產(chǎn)品得到了大眾的廣泛認(rèn)可,具有很好的穩(wěn)定性,處理高并發(fā)、高可用的能力突出,已應(yīng)用于各行各業(yè)。 為了更加直觀地看出其區(qū)別,將各消息隊(duì)列的性能指標(biāo)一一列出,如表1所示。

表1 常見(jiàn)的消息隊(duì)列比較Tab.1 Comparison of common message queue

面對(duì)這么多的MQ,一般使用以下原則選擇消息隊(duì)列:必須開(kāi)源。萬(wàn)一用戶使用的MQ突然遇到了一個(gè)影響業(yè)務(wù)的bug,用戶可通過(guò)修改源代碼來(lái)規(guī)避這個(gè)bug。要有以下幾個(gè)特性:消息的可靠性、支持集群、性能要好,系統(tǒng)內(nèi)存占用小。

基于這些原則,以電商平臺(tái)系統(tǒng)為例,設(shè)計(jì)一款在高并發(fā)條件下的消息隊(duì)列FineMQ,主要實(shí)現(xiàn)用戶商品的銷售等功能,FineMQ為中小型電商交易平臺(tái)提供了一個(gè)輕量級(jí)、高可用消息隊(duì)列的選擇。

3 消息隊(duì)列的設(shè)計(jì)

消息隊(duì)列系統(tǒng)需要處理大量數(shù)據(jù),包括數(shù)據(jù)處理容錯(cuò)機(jī)制,數(shù)據(jù)恢復(fù)機(jī)制,通信安全機(jī)制,因此設(shè)計(jì)的功能模塊需考慮到這些因素[4],設(shè)計(jì)了以下幾個(gè)功能模塊:連接模塊,用來(lái)處理消息發(fā)布者與消息訂閱者之間的連接或是長(zhǎng)連接。配置管理模塊,用來(lái)處理消息中心的配置信息、消息訂閱主題等。隊(duì)列管理模塊,用來(lái)視圖化輸出當(dāng)前消息處理效率,查看消息隊(duì)列處理響應(yīng)速度,幫助分析排查問(wèn)題。安全模塊,用來(lái)保證消息訂閱、發(fā)送與接收的安全性,達(dá)到安全通信要求。日志模塊,用來(lái)記錄消息處理情況并保留相關(guān)運(yùn)行日志,方便定位排查相關(guān)問(wèn)題。性能管理模塊,用來(lái)監(jiān)控消息隊(duì)列系統(tǒng)是否滿足當(dāng)前設(shè)計(jì)性能,是否達(dá)標(biāo),用于判斷產(chǎn)品是否成功,并可用于性能瓶頸分析。故障回復(fù)模塊,保證在服務(wù)器宕機(jī)、網(wǎng)絡(luò)中斷、系統(tǒng)進(jìn)程卡死、消息丟失等特殊情況下的信息故障恢復(fù)。

4 消息隊(duì)列的實(shí)現(xiàn)

以電商系統(tǒng)為例,前臺(tái)界面的設(shè)計(jì)主要實(shí)現(xiàn)電商的主要功能,包括系統(tǒng)的主頁(yè)商品顯示、商品倒計(jì)時(shí)搶購(gòu)、下單、加入購(gòu)物車、支付、查看消費(fèi)數(shù)據(jù)及統(tǒng)計(jì)報(bào)表等功能。根據(jù)瀏覽導(dǎo)航欄查看商品信息。通過(guò)瀏覽網(wǎng)站界面點(diǎn)擊詳情,便可知道商品的詳細(xì)信息,對(duì)滿意的商品直接加入購(gòu)物車或進(jìn)行商品收藏等操作,進(jìn)入訂單支付界面,多種支付方式簡(jiǎn)單方便。后臺(tái)管理員端主要實(shí)現(xiàn)管理員登錄、發(fā)布商品、修改發(fā)布的商品、管理商品規(guī)格(SKU)、管理商品評(píng)論、修改訂單狀態(tài)、查看商品收藏及用戶足跡等操作。定點(diǎn)銷售作為電商平臺(tái)的核心功能,不允許出現(xiàn)超賣的情況,因此在設(shè)計(jì)中通過(guò)消息隊(duì)列方案保證該功能的正常實(shí)現(xiàn),保證該模塊的高并發(fā)與高可用性能,通過(guò)消息隊(duì)列技術(shù)實(shí)現(xiàn)瞬時(shí)高并發(fā)請(qǐng)求,對(duì)海量的請(qǐng)求響應(yīng)結(jié)果做回應(yīng),保證每個(gè)用戶的請(qǐng)求都是有效且不丟失消息,而響應(yīng)界面的設(shè)計(jì)則是對(duì)高并發(fā)下對(duì)應(yīng)場(chǎng)景業(yè)務(wù)處理能力的最好回應(yīng)。

消息隊(duì)列功能的實(shí)現(xiàn)整體思路是需要確認(rèn)并創(chuàng)建一個(gè)整體的數(shù)據(jù)交互流,確定Producer、Broker與Consumer之間的關(guān)系。Producer發(fā)出消息給Broker,Broker發(fā)出消息給Consumer,Consumer接收消息,處理消息之后回復(fù)消息消費(fèi)確認(rèn)。Broker刪除/備份消息。通過(guò)RPC協(xié)議把整個(gè)數(shù)據(jù)流串起,通過(guò)選用合適的RPC協(xié)議達(dá)到消息列隊(duì)的高可用性,做到數(shù)據(jù)無(wú)狀態(tài)處理,方便消息列隊(duì)水平拓展?？紤]特殊情況下消息堆積的情況處理,如果在合適的時(shí)機(jī)向Consumer投遞消息,而消息推擠的最佳處理方式是通過(guò)存儲(chǔ)方案解決。存儲(chǔ)方案可以是磁盤文件存儲(chǔ)、內(nèi)存存儲(chǔ)等。

目前流行的ActiveMQ、RabbitMQ和ZeroMQ等消息隊(duì)列大多是為了實(shí)現(xiàn)AMQP、STOMP、XMPP之類的協(xié)議,占用太多的內(nèi)存(如新版本ActiveMQ建議分配內(nèi)存達(dá)1 G+),但很多Web應(yīng)用中只是想找到一個(gè)可以緩解高并發(fā)請(qǐng)求的解決方案,一個(gè)輕量級(jí)的消息隊(duì)列實(shí)現(xiàn)方式才是本系統(tǒng)真正需要的。參考RocketMQ[5],自行設(shè)計(jì)了一個(gè)輕量級(jí)消息隊(duì)列(FineMQ)。以下為各個(gè)模塊的設(shè)計(jì):

4.1 Broker子模塊的設(shè)計(jì)

Broker既要實(shí)現(xiàn)接收Producer的消息推送,也要向Consumer提供消費(fèi)信息。Broker應(yīng)支持集群且集群地位平等,支持集群可提高系統(tǒng)吞吐量。Broker要內(nèi)置注冊(cè)中心,通過(guò)注冊(cè)中心Broker能動(dòng)態(tài)感知Producer與Consumer的動(dòng)作,自動(dòng)匹配在線的Broker。Broker收到Producer的消息時(shí),第一時(shí)間應(yīng)存入隊(duì)列,而不是直接存儲(chǔ)消息,通過(guò)隊(duì)列的異步將隊(duì)列中的消息存入MySQL中。Broker實(shí)現(xiàn)核心代碼。批量新增消息:在接收Producer生產(chǎn)的消息的PRC調(diào)用時(shí),Broker不會(huì)立刻存儲(chǔ),而是立即push到內(nèi)存隊(duì)列中,同時(shí)立即響應(yīng)PRC調(diào)用,而內(nèi)存隊(duì)列會(huì)通過(guò)異步方式將隊(duì)列中的消息存儲(chǔ)到數(shù)據(jù)庫(kù)。Broker在接收到“消息鎖定”等同步RPC調(diào)用時(shí)會(huì)觸發(fā)同步調(diào)用,采用樂(lè)觀鎖方式鎖定消息。通過(guò)ChannelHandlerContext來(lái)直接跳到上一次終止的位置,不需要每次都要從頭開(kāi)始,減少每次從頭開(kāi)始找指定位置消息的時(shí)間。異常捕獲機(jī)制:當(dāng)捕獲到異常時(shí),自動(dòng)關(guān)閉連接。

4.2 Producer子模塊的設(shè)計(jì)

Producer(消息生產(chǎn)者)兼容異步批量多線程生產(chǎn)+同步生產(chǎn)兩種方式,提升消息發(fā)送性能。消息發(fā)送過(guò)程組成如下:

組裝消息,即對(duì)發(fā)送的消息進(jìn)行按需組裝,包括設(shè)置topic、tag、y延時(shí)及是否有序等。

生成topicPublishInfo,定時(shí)或按需從namesrv中同步該topic的broker消息。選擇隊(duì)列,從topicPublishInfo按照輪詢方式選擇隊(duì)列。發(fā)送消息,通過(guò)異步發(fā)送消息給Broker。

串行消費(fèi),ShardingId 保持一致即可,如消息,可將 ShardingId 設(shè)置為商品ID,則該商戶全部消息固定在一臺(tái)機(jī)器消費(fèi)。廣播消費(fèi),即點(diǎn)擊廣播消息按鈕一次,將會(huì)生產(chǎn)一條廣播消息(消費(fèi)者IMqConsumer注解的group 屬性修改不一致即可。一條消息將會(huì)廣播給該主題全部在線group,每個(gè)group都會(huì)消費(fèi),單個(gè)group只會(huì)消費(fèi)一次)。延時(shí)消費(fèi),EffectTime設(shè)置為固定時(shí)間點(diǎn)即可,如訂單30 min超時(shí)取消,可將EffectTime設(shè)置為30 min后的時(shí)間點(diǎn),到時(shí)將會(huì)自動(dòng)消費(fèi)。失敗重試消費(fèi),RetryCount設(shè)置重試次數(shù)即可。如發(fā)送短信消息,第三方服務(wù)不穩(wěn)定時(shí)失敗很常見(jiàn),可設(shè)置RetryCount為3,失敗時(shí)將會(huì)自動(dòng)重試指定次數(shù)。

4.3 Consumer子模塊的設(shè)計(jì)

在設(shè)計(jì)Consumer時(shí)要考慮以下幾個(gè)因素:支持用戶組,消費(fèi)主題(topic),消費(fèi)開(kāi)關(guān),避免重復(fù)消費(fèi)。Consumer子模塊初始化主要包括構(gòu)建consumer訂閱和消費(fèi)分組的重試隊(duì)列。創(chuàng)建Rebalance服務(wù), 該服務(wù)負(fù)責(zé)messageQueue的消費(fèi)。啟動(dòng)消費(fèi)偏移量獲取服務(wù),獲取上一次消費(fèi)位移。啟動(dòng)定時(shí)任務(wù),核心任務(wù)之一是定時(shí)去namesrv拉取broker信息。啟動(dòng)pullMessageService,從broker拉取等待消費(fèi)的消息。啟動(dòng)rebalanceService,負(fù)責(zé)定期調(diào)整consumer端負(fù)載均衡。

Consumer通過(guò)多線程自適應(yīng)輪詢,定時(shí)向Broker拉取消息進(jìn)行順序消費(fèi),消息消費(fèi)結(jié)束后調(diào)用RPC修改消息狀態(tài),追加消息日志,Broker利用內(nèi)存隊(duì)列的方式通過(guò)異步將消息隊(duì)列中的變更存儲(chǔ)到數(shù)據(jù)庫(kù)中。

5 消息隊(duì)列調(diào)度算法

當(dāng)系統(tǒng)接收到新消息時(shí),系統(tǒng)調(diào)度算法確定消息是否為響應(yīng)消息。如果是,帶有和信息相同ID的信息表示傳輸將正常刪除緩存隊(duì)列中存儲(chǔ)的信息。確定持續(xù)比特是否為1,消息被永久保存。再次確定接收到消息的用戶是本地還是需要路由,判斷用戶是否處于在線狀態(tài),如果是在線直接發(fā)送給用戶。例如,可以使用前進(jìn)程與后臺(tái)進(jìn)程兩個(gè)隊(duì)列。在前隊(duì)列中只能使用RR調(diào)度算法,但在端隊(duì)列中只能使用FCFS調(diào)整算法。除此之外,還需在隊(duì)列內(nèi)進(jìn)行調(diào)整。一般情況下,使用固定優(yōu)先權(quán)預(yù)設(shè)調(diào)整。因此前臺(tái)隊(duì)列應(yīng)優(yōu)先于后臺(tái)隊(duì)列[3]。消息隊(duì)列調(diào)度算法流程如圖1所示:

圖1 消息隊(duì)列調(diào)度算法流程Fig.1 Flow of message queue scheduling algorithm

每個(gè)隊(duì)列都是雙向鏈表,信件從隊(duì)列的頭部中提取,并在末尾進(jìn)行分立。出隊(duì)和入隊(duì)可以并行工作,互不干涉。對(duì)隊(duì)列的操作以消息為單位,對(duì)每個(gè)隊(duì)列采用FIFO原理。

6 系統(tǒng)的并發(fā)性能測(cè)試

為了保證消息隊(duì)列的性能要求,需對(duì)該功能進(jìn)行一定的高并發(fā)環(huán)境壓力測(cè)試。通過(guò)Jemeter工具進(jìn)行用例壓力測(cè)試。用例如表2所示。

表2 性能測(cè)試用例Tab.2 Performance test cases

7 結(jié)束語(yǔ)

使用的消息隊(duì)列是通過(guò)參考RocketMQ而實(shí)現(xiàn)的面向小型電商平臺(tái)的輕量級(jí)消息隊(duì)列。該消息隊(duì)列解決了并發(fā)量不高、對(duì)服務(wù)器要求較高、占用系統(tǒng)資源過(guò)多等問(wèn)題,增加了系統(tǒng)的穩(wěn)定性及安全性,減輕了服務(wù)器壓力,減少了內(nèi)存占用,為小型電商平臺(tái)提供了一個(gè)輕量級(jí)消息隊(duì)列的選擇。