高并發條件下消息隊列的設計與實現

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

(南京信息職業技術學院,南京 210043)

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

研發設計了一個即時消息隊列FineMQ,通過即時消息列隊利用中間件的處理能力,解決應用解耦及消息異步,實現高性能及最終的一致性架構。

1 消息隊列

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

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

1.1 消息

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

1.2 隊列

列隊是一種常用的數據結構,是存在于內存或磁盤中、開辟一塊公用存儲空間存放數據的一組數據結構,用來處理一組數據的臨時空間。為了保證數據的有序處理,將所有數據存放在列隊中,在列隊中依次讀取操作,保證系統處理數據的完整性。列隊一般分為兩種模式,即順序結構和鏈式結構。不同模式的數據處理性能有所不同,而消息隊列是用來處理海量消息的一種方式。隊列的排隊機制主要分為4種,即先進先出(FIFO)、優先排隊、公平排隊、加權公平排隊。

2 消息隊列的對比和選擇

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

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

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

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

3 消息隊列的設計

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

4 消息隊列的實現

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

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

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

4.1 Broker子模塊的設計

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

4.2 Producer子模塊的設計

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

組裝消息,即對發送的消息進行按需組裝,包括設置topic、tag、y延時及是否有序等。

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

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

4.3 Consumer子模塊的設計

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

Consumer通過多線程自適應輪詢,定時向Broker拉取消息進行順序消費,消息消費結束后調用RPC修改消息狀態,追加消息日志,Broker利用內存隊列的方式通過異步將消息隊列中的變更存儲到數據庫中。

5 消息隊列調度算法

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

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

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

6 系統的并發性能測試

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

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

7 結束語

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