基于Netty 的高性能消息中間件設(shè)計與實現(xiàn)

王寧，張娜，于澤川，蘇逸凡，包曉安

（浙江理工大學(xué) 信息學(xué)院，杭州 310018）

0 引言

隨著互聯(lián)網(wǎng)技術(shù)的迅速發(fā)展、物聯(lián)網(wǎng)應(yīng)用的快速普及以及大數(shù)據(jù)時代的來臨，傳統(tǒng)互聯(lián)網(wǎng)應(yīng)用服務(wù)已不能滿足高并發(fā)低延時的需求［1］。當(dāng)海量物聯(lián)網(wǎng)設(shè)備終端在短時間內(nèi)連續(xù)向業(yè)務(wù)服務(wù)器發(fā)送訪問請求時，業(yè)務(wù)服務(wù)器無法及時處理這些訪問請求，繼而造成大量訪問請求在服務(wù)端的堆積，瞬時數(shù)據(jù)量激增，最終導(dǎo)致服務(wù)器無法及時響應(yīng)客戶端，甚至?xí)霈F(xiàn)數(shù)據(jù)丟失的情況，系統(tǒng)服務(wù)的實時性與可靠性出現(xiàn)問題［2］。消息中間件應(yīng)運而生，通過引入消息中間件，應(yīng)用服務(wù)采用異步通信的方式，可以有效提升服務(wù)器的響應(yīng)速率，同時保證了應(yīng)用的可靠性。

本文通過對ActiveMQ、RocketMQ、kafka等當(dāng)前流行的消息中間件研究發(fā)現(xiàn)：ActiveMQ 默認(rèn)情況下使用的是消息推送的方式，當(dāng)數(shù)據(jù)量驟增的情況下，如果其中一個消費者的消費能力較差，可能導(dǎo)致消息在消費者端產(chǎn)生堆積，無法解決互聯(lián)網(wǎng)應(yīng)用服務(wù)目前遇到的問題［3－4］；而RocketMQ 與kafka 使用的消費模式是基于partition 的，其將每個主題分為一個或多個partition，每個消費者與其中一個partition靜態(tài)綁定［5］，若其中一個partition 的消費者消費能力較慢時，則無法通過增加消費者數(shù)量的方式來加快該partition 的消費速度，進而導(dǎo)致消息積壓的問題［6－7］。

綜上考慮，本文設(shè)計了一種基于Netty 框架的消息中間件，能夠較好的解決上述中間件在應(yīng)用服務(wù)中所產(chǎn)生的問題。本文中間件采用發(fā)布／訂閱模式，消息生產(chǎn)者Producer（物聯(lián)網(wǎng)設(shè)備終端）將數(shù)據(jù)消息發(fā)布到中間件服務(wù)器Broker 的消息隊列中，再由消息消費者Consumer 主動從Broker 的消息隊列中拉取并消費。該中間件Broker 與Consumer 通信模塊基于Netty 網(wǎng)絡(luò)框架開發(fā)，能夠滿足短時間內(nèi)連接數(shù)與消息數(shù)據(jù)急速增多的需求。同時利用Netty 框架對自定義協(xié)議的支持，提出了一種簡單的、能夠快速解析的通信協(xié)議，以進一步增強中間件系統(tǒng)的并發(fā)處理能力。本文提出的以Netty 網(wǎng)絡(luò)框架為基礎(chǔ)實現(xiàn)異步通信的中間件，在本實驗室開發(fā)的智慧宿管平臺中進行了廣泛應(yīng)用并得到了很好的驗證。

1 相關(guān)技術(shù)

1.1 Netty 線程模型

Netty 是一個高性能的異步事件驅(qū)動模型的網(wǎng)絡(luò)框架，其高性能主要得益于其I／O 模型和線程模型。Netty 的實現(xiàn)主要基于Reactor 主從多線程模型，將接收客戶端請求與處理連接產(chǎn)生的事件這兩部分進行了拆分。Netty 服務(wù)端使用兩個NioEventLoopGroup，一個叫作bossGroup 作為主線程池，另一個叫作workerGroup 作為子線程池。主線程池負(fù)責(zé)接收客戶端請求，子線程池負(fù)責(zé)處理所有連接產(chǎn)生的事件。當(dāng)bossGroup 接收了客戶端的連接請求后，會將這個請求封裝成NioSocketChannel，然后注冊到workerGroup 上，此后這個連接產(chǎn)生的所有事件都由 workerGroup 來處理。Netty 中 的NioEventLoopGroup 中包含多個EventLoop，每個EventLoop 都聚合了一個多路復(fù)用器Selector，相當(dāng)于一個Reactor 單線程模型，從而只需少量的EventLoop 就可以同時并發(fā)處理大量客戶端連接。并且為了提升性能，Netty 在I／O 線程內(nèi)部采用了串行操作，避免了多線程競爭情況下頻繁加鎖導(dǎo)致的性能下降問題。因此，每個客戶端連接在其生命周期中只會與一個EventLoop 進行綁定，該連接產(chǎn)生的所有的讀寫事件都由與之綁定的EventLoop 進行處理，無需進行線程切換，從而避免了多線程操作導(dǎo)致的鎖競爭問題。Netty 線程模型如圖1 所示。

圖1 Netty 線程模型Fig.1 Netty thread model

1.2 消息傳輸模型

Java 消息服務(wù)是針對Java 平臺之間消息傳輸所制定的一套規(guī)范，其中提出了點對點（Point To Point，p2p）與發(fā)布／訂閱（Publish／Subscribe，PUB／SUB）兩種消息傳輸模型［8］。兩種傳輸模型結(jié)構(gòu)如圖2 所示。

圖2 兩種傳輸模型結(jié)構(gòu)圖Fig.2 Two kinds of transmission model structure diagram

（1）點對點模型：點對點模型主要由發(fā)送者、接收者和消息隊列3 部分組成［9－10］。每條消息產(chǎn)生后，都由發(fā)送者將其發(fā)送到消息隊列中，消息隊列在收到消息后會將消息保存，并通知接收者，接收者接到隊列發(fā)送的消費通知后，從消息隊列中拉取消息進行消費，并在消費結(jié)束后通知消息隊列［11］。若消息已經(jīng)被接收者消費或是超過時間，消息隊列便會將這個消息刪除。這個模型下的消息，只能被一個接收者消費，并且接收者與發(fā)送者之間不存在時間相關(guān)性［12］。

（2）發(fā)布／訂閱模型：發(fā)布／訂閱模型是由發(fā)布者、主題、訂閱者3 個部分組成［13］。消息發(fā)布者將一條消息發(fā)送到對應(yīng)主題后，消息訂閱者可采用兩種方式獲取該條消息：一是訂閱者主動詢問消息服務(wù)器，對應(yīng)主題中是否有未消費的消息；二是消息服務(wù)器接收到這條消息后，立即推送給所有訂閱了這個主題的訂閱者，而無需訂閱者去詢問。發(fā)布／訂閱模型允許一條消息被所有的訂閱者同時消費，具有一對多的特點。

2 系統(tǒng)設(shè)計

2.1 系統(tǒng)結(jié)構(gòu)模型

本文設(shè)計的消息中間件由消息生產(chǎn)者Producer、消息服務(wù)器Broker 以及消息消費者Consumer 3 部分組成。其中，生產(chǎn)者角色主要為智慧宿管平臺系統(tǒng)中智能門鎖以及移動終端；消費者角色則為智慧宿管平臺系統(tǒng)的后臺服務(wù)端。中間件與智慧宿管平臺的關(guān)系模型如圖3 所示。

圖3 中間件與智慧宿管平臺關(guān)系模型圖Fig.3 Relationship model diagram between middleware and intelligent residential management platform

本文中間件的消息服務(wù)器Broker 分為4 個部分：協(xié)議制定、消息接收模塊、消息消費模型以及消息持久化方式。

2.1.1 協(xié)議制定

本文設(shè)計的消息中間件是面向智慧宿管平臺的，消息生產(chǎn)者主要是智能門鎖等終端設(shè)備。由于智能門鎖這一類的硬件設(shè)備具有低功耗、低內(nèi)存以及芯片級設(shè)計等特點，所支持的網(wǎng)絡(luò)協(xié)議也都較為簡單，相對于其它網(wǎng)絡(luò)設(shè)備也更接近底層［14］。因此，本文設(shè)計的消息中間件系統(tǒng)之間的數(shù)據(jù)傳輸都是通過TCP（Transmission Control Protocol，傳輸控制協(xié)議）協(xié)議實現(xiàn)。TCP 協(xié)議是一種流式服務(wù)，所以各端接收消息數(shù)據(jù)時，會出現(xiàn)網(wǎng)絡(luò)傳輸中最常見的粘包拆包問題。為此本文設(shè)計了一種自定義協(xié)議，并通過實現(xiàn)Netty為自定義協(xié)議所提供的兩個編解碼器抽象類來解決粘包拆包問題。自定義協(xié)議組成見表1。

表1 協(xié)議組成Tab.1 Composition of the agreement

2.1.2 消息接收模塊

消息服務(wù)器Broker 基于Netty 開發(fā)，核心類為SmartDepBrokerServer，類結(jié)構(gòu)見表2。首先初始化啟動類ServerBootstrap，綁定主從線程池bossGroup 與workerGroup，并為workerGroup 預(yù)設(shè)了Channel 初始化器。

表2 SmartDepBrokerServer 類結(jié)構(gòu)Tab.2 Class structure of SmartDepBrokerServer

Broker 啟動后，通過bossGroup 監(jiān)聽所有客戶端的連接請求，進行連接建立。連接建立時，bossGroup將連接封裝成一個NioSocketChannel，并將其注冊到workerGroup 上；workerGroup 通過Channel 初始化器為這個連接的NioSocketChannel 添加預(yù)設(shè)的編解碼器以及消息處理器MessageBrokerHandler，并監(jiān)聽該SocketChannel 所有的讀寫事件。當(dāng)有數(shù)據(jù)可讀時，先將讀取的數(shù)據(jù)通過解碼器按照自定義協(xié)議解碼，然后通過調(diào)用MessageBrokerHandler 中的handleMessage 方法對消息數(shù)據(jù)進行處理。消息處理器MessageBrokerHandler類結(jié)構(gòu)見表3。

表3 MessageBrokerHandler 類結(jié)構(gòu)Tab.3 Class structure ofMessageBrokerHandler

2.1.3 消息消費模型

目前Kafka 與RocketMQ等一些流行的中間件使用的都是基于partition 的消費模型，即將一個主題Topic 分為多個partition，而每一個partition 則與一個消費者Consumer 進行靜態(tài)綁定，消費者需要消費消息時，只會從與之綁定的Partition 中獲取消息。但是當(dāng)消費者數(shù)量多于partition 數(shù)量時，會導(dǎo)致有些消費者被閑置，浪費了系統(tǒng)資源［15］。并且一旦遇到消息發(fā)送方突然大量發(fā)送消息的場景，消費者處理速度較慢，而基于partition 的消費模型無法通過消費者擴容來提高處理速度，從而導(dǎo)致消息大量堆積。

本文的中間件采用發(fā)布／訂閱模型，并將所有同屬一個主題Topic 的消息集合起來，形成一個主題隊列Queue。每個主題隊列對應(yīng)一個消費者集群ConsumerCluster，每個消費者集群中的消費者數(shù)量能夠根據(jù)消息數(shù)量動態(tài)擴容與縮容。當(dāng)消息生產(chǎn)者發(fā)布消息到消息服務(wù)器的對應(yīng)主題隊列后，消息服務(wù)器首先調(diào)用消息持久化模塊，將消息持久化到日志文件中，持久化完成后才會去輪詢訂閱了該主題的消費者集群ConsumerCluster，通知集群中的消費者消費，再由消費者主動去消息服務(wù)器中拉取消息進行處理消費。由于通知消費者消費消息之前，消息服務(wù)器已經(jīng)將消息持久化到內(nèi)存中，確保消息不會丟失，實現(xiàn)了消息中間件的高可靠性。消費模型如圖4 所示。

圖4 消費模型圖Fig.4 Consumption model chart

2.1.4 消息持久化

為了保證中間件的高可靠性，避免消息服務(wù)器因意外宕機導(dǎo)致的消息丟失問題，消息服務(wù)器需要在重啟后能夠恢復(fù)原先的消息隊列，消息中間件一般都需有消息持久化機制［16－17］。本文的消息中間件使用的持久化方式是基于消費模型設(shè)計，采用文件存儲系統(tǒng)，將消息持久化到日志文件中。每次消息服務(wù)器接收到新消息時，會按照順序添加的方式，將消息存儲到日志文件message.log 中。相比于其它訪問文件的方式，本文采用順序添加的方式具有更高的效率。由于將所有的消息都存儲到一個日志文件中，為了區(qū)分不同主題的消息，本文通過引入一個索引文件offset－topicName.log，來記錄不同主題隊列中的消息在message.log 文件中的偏移量，每個主題隊列對應(yīng)一個索引文件。最后使用一個消費記錄日志consume－clusterId.log 記錄每個消費者集群的消費情況，保存所消費的消息在索引日志中的序號，并采用順序消費的方式，無需為每個消息單獨記錄消費狀態(tài)，極大地降低了成本。存儲模型如圖5 所示。

圖5 存儲模型圖Fig.5 Storage model diagram

2.1.5 冪等性處理

正常情況下，消費者在消費一條消息之后，都會給消息服務(wù)器回發(fā)一條確認(rèn)消息。當(dāng)消息服務(wù)器收到確認(rèn)消息后，消費者才能拉取下一條消息。但是，由于網(wǎng)絡(luò)環(huán)境的不確定性，消費者所發(fā)送的確認(rèn)數(shù)據(jù)包在網(wǎng)絡(luò)傳輸中可能會丟失，從而導(dǎo)致消息服務(wù)器誤認(rèn)為還沒有消費者消費這條消息，繼而對其進行重發(fā)，消費者就會重復(fù)消費這條消息，最終造成數(shù)據(jù)異常［18－19］。

對于系統(tǒng)冪等性處理，本文做法是首先保證每條消息都擁有一個全局唯一標(biāo)識，然后在數(shù)據(jù)庫中建立一張存儲已處理的消息表，并將每條消息的全局唯一標(biāo)識作為表的主鍵。每當(dāng)消費者獲取一個新消息時，先嘗試向數(shù)據(jù)庫中插入這條消息記錄，如果插入失敗，說明已經(jīng)處理過該消息，并向消息服務(wù)器返回一條確認(rèn)應(yīng)答，以便獲取下一條消息。

2.2 數(shù)據(jù)消息處理

消息生產(chǎn)者（終端設(shè)備）、消息中間件和消息消費者（智慧宿管平臺服務(wù)端）3 者之間的數(shù)據(jù)交互主要是消息生產(chǎn)與消息消費這兩個過程。其中，消息生產(chǎn)主要是消息生產(chǎn)者與中間件之間的數(shù)據(jù)交互，消息消費則是消息中間件與消息消費者之間的數(shù)據(jù)交互。終端設(shè)備發(fā)送數(shù)據(jù)消息時，會先向消息中間件發(fā)送建立連接的請求，中間件通過Netty 通信模塊與終端設(shè)備建立連接。連接建立后，終端設(shè)備便會將數(shù)據(jù)消息上傳至消息中間件，中間件在接收到數(shù)據(jù)信息后，先將消息持久化存儲到數(shù)據(jù)日志文件中，以保證消息的可靠性，然后再將數(shù)據(jù)消息根據(jù)其Topic 放入到對應(yīng)隊列中，并通知消費者讀取消息進行處理。整個消息處理流程如圖6 所示。

圖6 數(shù)據(jù)消息處理流程圖Fig.6 Data message processing flow chart

3 實驗結(jié)果與分析

3.1 實驗環(huán)境

本文系統(tǒng)運行于一臺四核高性能阿里云服務(wù)器上，服務(wù)器配置參數(shù)為：CPU 主頻為2.5 GHz、采用Intel（R）Xeon（R）Platinum 8163（Skylake）處理器、8 G內(nèi)存、40 G 高效云盤、操作系統(tǒng)版本為CentOS 6.9；實驗通過開源壓力測試工具Jmeter 模擬大量終端設(shè)備，同時喚醒業(yè)務(wù)場景，向系統(tǒng)發(fā)起大規(guī)模TCP連接請求；壓測機器使用一臺雙核心四線程的計算機，其具體配置參數(shù)為：Intel（R）Core（TM）i5－4210M 處理器、CPU 主頻為2.60 GHz、8 G 內(nèi)存、128 G固態(tài)硬盤、操作系統(tǒng)Windows 10。

3.2 實驗設(shè)計與結(jié)果分析

實驗將通過對智慧宿管平臺結(jié)合NIO 通信進行壓測，對智慧宿管平臺結(jié)合高性能中間件ActiveMQ 進行壓測，以及對智慧宿管平臺結(jié)合本文設(shè)計的中間件進行壓測3 種方案進行對比。從高并發(fā)情況下系統(tǒng)的總響應(yīng)時間，以及數(shù)據(jù)準(zhǔn)確率兩方面來進行實驗結(jié)果的對比分析。通過設(shè)置壓測工具Jmeter 的線程數(shù)，模擬實現(xiàn)不同數(shù)量的終端設(shè)備在1 s內(nèi)發(fā)送TCP 請求，并設(shè)置請求參數(shù)為一串16 進制數(shù)字，大小為24 B，表示某一個門鎖設(shè)備的開鎖記錄。

3 種方案的系統(tǒng)總響應(yīng)時間如圖7 所示，并發(fā)量在4 000 以下時，兩種使用中間件的方案優(yōu)勢較為明顯，其系統(tǒng)總響應(yīng)時間明顯少于傳統(tǒng)NIO 通信方案的總響應(yīng)時間。由于傳統(tǒng)NIO 通信方案需要等待服務(wù)端將數(shù)據(jù)處理完畢后再返回，而數(shù)據(jù)處理是一個耗時操作，因此傳統(tǒng)NIO 通信方案的總響應(yīng)時間最久。而當(dāng)并發(fā)量達到6 000時，ActiveMQ 無法在運行過程中實現(xiàn)消費者集群的動態(tài)擴容，從而導(dǎo)致消息堆積并會阻塞生產(chǎn)者，響應(yīng)時間隨之增加。而本文設(shè)計的中間件方案通過動態(tài)改變消費者數(shù)量，加快了消息的處理，縮短了系統(tǒng)的響應(yīng)時間，表現(xiàn)較為穩(wěn)定。

圖7 不同并發(fā)場景下系統(tǒng)總響應(yīng)時間Fig.7 Total response time in different concurrent scenarios

在數(shù)據(jù)準(zhǔn)確率方面，3 種方案的系統(tǒng)平均數(shù)據(jù)接收準(zhǔn)確率如圖8 所示。當(dāng)并發(fā)量小于2 000時，3種方案的數(shù)據(jù)接收準(zhǔn)確率表現(xiàn)相差不大；當(dāng)并發(fā)量大于5 000時，NIO 通信以及ActiveMQ 方案的數(shù)據(jù)接收準(zhǔn)確率明顯下降，出現(xiàn)較多數(shù)據(jù)丟失的情況，隨著并發(fā)量上升，系統(tǒng)響應(yīng)時間增長，導(dǎo)致了服務(wù)端響應(yīng)異常。而本文中間件的表現(xiàn)則優(yōu)于另外2 種方案，數(shù)據(jù)丟失的情況較少，準(zhǔn)確率穩(wěn)定在98%以上。

圖8 不同并發(fā)場景下系統(tǒng)接收數(shù)據(jù)的準(zhǔn)確率Fig.8 The accuracy of receiving data in different concurrent scenarios

4 結(jié)束語

在當(dāng)下大數(shù)據(jù)、大流量環(huán)境中，消息中間件的應(yīng)用前景十分廣闊。為解決智慧宿管平臺在高并發(fā)情況出現(xiàn)的響應(yīng)不及時、數(shù)據(jù)丟失的問題，本文設(shè)計的基于Netty框架的消息中間件和自定義協(xié)議進行通信，使得消息數(shù)據(jù)更加簡單，數(shù)據(jù)解析更加高效。為了應(yīng)對不同的消費場景，采用可擴展的消費者集群的算法來實現(xiàn)消費者數(shù)量的動態(tài)變化。

目前，本文設(shè)計的中間件系統(tǒng)已經(jīng)在智慧宿管平臺的實際業(yè)務(wù)場景中應(yīng)用，提高了響應(yīng)速率，解決了數(shù)據(jù)丟失等問題，系統(tǒng)運行效果良好。但是由于硬件設(shè)備以及實際業(yè)務(wù)場景的局限性，更高并發(fā)場景下該消息中間件的表現(xiàn)與改進還有待進一步的研究與實踐。