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

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

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

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

1 消息隊列

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

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

1.1 消息

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

1.2 隊列

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

2 消息隊列的對比和選擇

目前市面上主流的消息隊列有很多[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ū)別,將各消息隊列的性能指標(biāo)一一列出,如表1所示。

表1 常見的消息隊列比較Tab.1 Comparison of common message queue

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

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

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

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

4 消息隊列的實現(xiàn)

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

表2 性能測試用例Tab.2 Performance test cases

7 結(jié)束語

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