基于響應式的服務推送框架設計

胡喜明，胡 淼

(杭州電子科技大學 通信工程學院，浙江 杭州 310018)

0 引 言

隨著物聯(lián)網(wǎng)技術的發(fā)展，人們能夠通過移動終端方便獲取所需要的信息與服務[1]。為了保證網(wǎng)絡設備與物理設備間可靠的傳輸，用戶將傳統(tǒng)的消息拉取模式轉變?yōu)閼孟⑼扑汀Ｅc消息拉取模式相比，由服務器將所產生的消息推送用戶，符合當前互聯(lián)網(wǎng)流量管理以及資源分配的需求，同時在傳輸?shù)膶崟r性、高效性以及流量資源節(jié)約方面都有較大的提升[2]。物聯(lián)網(wǎng)設備需要考慮到硬件相關參數(shù)，例如CPU性能、最大帶寬容量、電池使用時間、網(wǎng)絡最大實時流量等方面的問題，在服務推送方面需要一種功能完善、更加輕量、帶寬占比小且性能高的服務推送技術[3]。

當前主流的服務推送方案有GCM服務、MQTT以及HTTP消息輪循[4]。其中HTTP輪循方式最為簡單，但是在實用性、可靠性以及可擴展性方面有所欠缺。GCM服務是google研發(fā)的一款云推送框架，擁有完善的推送方案，但是受限于網(wǎng)絡原因，無法在國內使用。MQTT協(xié)議具有低延遲、低帶寬、推送速度快等優(yōu)勢，適合大部分物聯(lián)網(wǎng)開發(fā)場景，然而單純的MQTT協(xié)議本身過于簡單，在消息推送、安全性管理、主題化推送方式、數(shù)據(jù)緩存性等方面開發(fā)難度大，因此需要對MQTT協(xié)議進行二次開發(fā)[5]。

本文提出一種基于Reactor-Netty+MQTT的高性能服務推送框架。項目中應用Reactor-Netty技術替代傳統(tǒng)Netty技術對MQTT協(xié)議進行封裝，借助該技術的響應式非阻塞編程、事件驅動等特性構建高性能通信服務。通過Redis進行數(shù)據(jù)緩存和Kafka實現(xiàn)消息代理轉發(fā)進一步提高系統(tǒng)服務處理能力以及線程安全。

1 相關技術

1.1 MQTT協(xié)議

1.1.1 MQTT背景及特點

MQTT是由IBM公司在1999年所提出的一款基于TCP協(xié)議的發(fā)布訂閱協(xié)議[6,7]。MQTT是為了在有限的帶寬以及內存條件下實現(xiàn)消息可靠傳輸。該協(xié)議提供發(fā)布/訂閱形式，實現(xiàn)單點以及一對多的發(fā)布消息推送、應用TCP協(xié)議實現(xiàn)有序可靠雙端連接、報頭以及心跳報文僅占用3個字節(jié)，將帶寬傳輸最小化，有效降低網(wǎng)絡通信流量、支持“最多一次”、”至少一次”，“僅一次“3種服務質量等級、支持遺囑機制，實現(xiàn)客戶端異常斷開后，自動根據(jù)所設置的遺囑機制，發(fā)布相關主題信息通知其它訂閱的客戶端用戶。基于以上特點使得當前物聯(lián)網(wǎng)的開發(fā)中絕大多數(shù)都將MQTT作為消息傳遞協(xié)議的首選[8]。

MQTT協(xié)議主要擁有3個角色：消息發(fā)布者、消息代理、消息訂閱者[9]。消息通過推送的形式從發(fā)布者經過消息代理進行發(fā)布，此時消息已經確定了自身所對應的主題，消息訂閱者可以根據(jù)主題進行相關數(shù)據(jù)的訂閱[10,11]。MQTT工作模型如圖1所示。

圖1 MQTT工作模型

1.1.2 MQTT協(xié)議消息傳輸格式

MQTT協(xié)議主要由固定報頭、可變報頭、有效載荷3部分構成。固定報頭是所有MQTT報文段中必須存在的，總長度為2字節(jié)。在第一個字節(jié)中，7-4位標識MQTT控制報文類型，目前報文類型主要分為連接、訂閱、發(fā)布3種。服務質量占兩個標識位，分為Qos0、Qos1、Oos23種。4-0位最為控制報文標志位，可看作一種屬性參數(shù)。可變報頭與有效載荷根據(jù)協(xié)議的不同進行省略[12，13]。

1.2 Reactor-Netty框架

為了應對高并發(fā)場景下流量過大情況，微軟設計了一種異步的編程思想-響應式編程(reactive programming)。響應式編程是一種專注于數(shù)據(jù)流以及變化傳遞的異步編程模式，這也意味著可以應用編程語言進行靜態(tài)和動態(tài)數(shù)據(jù)流的表示。在jdk9中java引入了Reactor的概念，Reactor-Netty作為響應式編程家族的一員，其底層基于Netty框架，對Netty進行響應式編程封裝，將其轉換為異步事件驅動的網(wǎng)絡應用程序框架。Reactor-Netty內部仍然保留了Netty的主從多線程模型，擁有Netty框架的全部優(yōu)勢。

Reactor通過Reactor Streams中的背壓，進行數(shù)據(jù)流量控制，發(fā)布者和訂閱者可以進行數(shù)據(jù)流量協(xié)商，其中背壓分為4種回壓策略：①onBackpressureBuffer：對下游的請求數(shù)據(jù)采用緩存的形式，保證系統(tǒng)不會壓力過大。②OnBackpressureDrop：元素就緒時，根據(jù)下游是否有未滿足的request來判斷是否發(fā)出當前元素。③OnBackpressureLatest：一直發(fā)送當前最新的數(shù)據(jù)。④OnBackpressureError：當前數(shù)據(jù)已經滿了，再次添加請求直接報錯。圖2為OnBackpressureBuffer模式的背壓原理圖，當訂閱者的消費能力遠小于發(fā)布者，訂閱者可以通知發(fā)布者進行服務的取消和終止功能，保證傳輸數(shù)據(jù)流量合理。

圖2 OnBackpressureBuffer模式背壓原理

2 服務推送框架設計

整體服務器分為3部分，消息推送broker模塊，針對MQTT協(xié)議所定義的多種消息類型進行分類處理以及通訊傳輸協(xié)議的搭建。服務認證模塊，為保證服務的安全性提供了接口化驗證，通過將RSA算法與Redis緩存結合，實現(xiàn)安全性加密認證。數(shù)據(jù)緩存模塊，為防止傳輸中服務突然宕機所導致的數(shù)據(jù)丟失，采用Redis對服務中傳輸?shù)臄?shù)據(jù)進行緩存。消息分發(fā)模塊，針對服務中耗時操作以及大型互聯(lián)網(wǎng)數(shù)據(jù)進行流量統(tǒng)計的需求，對接企業(yè)級消息隊列kafka實現(xiàn)消息服務端與數(shù)據(jù)分發(fā)接收端的解耦，便于對海量數(shù)據(jù)的處理。系統(tǒng)架構如圖3所示。

圖3 服務推送系統(tǒng)架構

3 服務推送框架模塊實現(xiàn)

3.1 服務認證模塊

服務認證是對系統(tǒng)進行安全維護的手段，客戶端在接入服務器時需要進行權限認證工作，通過對接入用戶的身份驗證，保證服務傳輸信息不被惡意獲取。驗證流程如圖4所示。

圖4 服務認證流程

為保證服務的安全性，需要對密碼進行加密處理，防止由于密碼泄露所導致的非法客戶端接入竊取信息。這里采用RSA非對稱加密算法實現(xiàn)公私鑰加密校驗。應用項目中提供的RSAUTIL類生成公鑰與私鑰，其內部經過大量的邏輯運算實現(xiàn)加密。系統(tǒng)需要保證每次生成的公鑰與私鑰唯一，這里將服務名稱與當前時間進行拼接作為加密的鹽值。這樣每次獲取生成的公鑰私鑰必然唯一。公鑰私鑰生成后以服務名稱作為key值，明文密碼作為value存儲到Redis中。

客戶端在發(fā)起連接時，需要將通過公鑰加密后的密碼和用戶名稱一并傳輸，服務器接收到信息后，通過私鑰對傳輸密碼進行解密與Redis中存儲的密碼進行對比驗證。

3.2 MQTT核心Broker模塊

Broker模塊是推送服務的核心功能模塊，關于MQTT協(xié)議的邏輯處理部分均在該模塊進行。其中主要包括QoS服務質量等級選擇功能、遺囑消息功能，保留消息功能、客戶端自動重連功能、心跳機制功能、MQTT/WS連接功能以及主題過濾功能。

QoS服務質量等級選擇功能主要是為根據(jù)訂閱端業(yè)務為其提供消息選擇類型的處理，內部根據(jù)選擇等級的不同實現(xiàn)相應邏輯處理。其中主要分為Qos1、Qos2以及Qos3這3種。

(1)Qos0：在該服務質量下，消息至多發(fā)送一次，其消息的發(fā)送完全依賴于OSI七層協(xié)議中的傳輸層進行維護，可能會發(fā)生消息丟失或者重復的情況。該服務質量可用于如下情況：定時推送周圍環(huán)境相關數(shù)據(jù)，丟失一兩組數(shù)據(jù)不會影響服務的應用且在不久后會再次推送。該情況主要用于普通APP定時推送功能，若設備當前推送數(shù)據(jù)時設備并未接入網(wǎng)絡，此時即使再次聯(lián)網(wǎng)也會丟失數(shù)據(jù)。

(2)Qos1：在該服務質量下，消息至少發(fā)送一次，消息發(fā)送到客戶端后，客戶端返回確認信息標志當前服務已送達。該服務由于有一次確認機制，因此在網(wǎng)絡環(huán)境比較好的情況下可以實現(xiàn)數(shù)據(jù)正確流暢的傳輸，但是若網(wǎng)絡環(huán)境出現(xiàn)波動使得服務端無法收到確認信息，可能會造成數(shù)據(jù)多次重發(fā)的問題。

(3)Qos2：該服務質量在三者中處于最高級別，該消息僅會發(fā)送一次。主要用在當消息丟失或者重發(fā)時，對服務端造成業(yè)務上的影響的情況。為了保證傳輸一次的要求，采用了兩階段確認的方式，與Qos1相比開銷較大。

遺囑消息功能主要是在發(fā)生服務斷連問題的時候，訂閱端無法及時了解消息發(fā)布方狀態(tài)導致的持續(xù)等待問題。遺囑消息保證了消息發(fā)布方在網(wǎng)絡波動所導致的服務下線時能夠及時通知訂閱方，這里遺囑的內容以事件監(jiān)聽器的形式實現(xiàn)，一旦出現(xiàn)服務斷連的情況即可通過監(jiān)聽模式進行響應；

保留消息功能主要是對重要信息進行標記，任何標記為保留消息的內容，新接入的訂閱方都可以在連接后收到，并且不需要等待消息發(fā)送方的推送，內部將消息存儲在Redis緩存中，保證數(shù)據(jù)不丟失；

訂閱方斷線重連功能主要是為了將服務自動化管理，保證訂閱方在由于網(wǎng)絡波動導致失去連接，服務器會自動實現(xiàn)斷線重連，降低維護成本，由Reactor-Netty內部采用的心跳檢測機制，通過設定心跳間隔實現(xiàn)斷線檢測功能，并且基于事件驅動，在產生了心跳斷連的情況時觸發(fā)重連機制，通過輪循的方式發(fā)送重連請求保證服務的可靠性；

心跳檢測功能主要是通過底層Netty的心跳檢測模塊，MQTT協(xié)議中有PINGREQ心跳請求，通過實現(xiàn)底層Rea-ctor-Netty的Handler類，對傳入的PINGREQ進行管理，客戶端定時發(fā)送心跳檢測請求，通過Reactor-Netty內置的MONO類進行調用，該類基于事件觸發(fā)，可根據(jù)實際情況進行結束處理。并且MONO類可以與java8中并發(fā)異步響應類進行轉化，服務端將心跳響應包裝為MONO進行返回操作；

MQTT/WS連接功能在推送服務中，不僅要與采用MQTT協(xié)議的硬件設備進行關聯(lián)，同時還會與互聯(lián)網(wǎng)設備產生連接，因此在服務器接收連接請求時會判斷當前訂閱端類別，根據(jù)MQTT以及WS進行分類處理。底層采用工廠設計模式對服務進行劃分，根據(jù)傳入的標識進行判斷并選取不同的處理方式。對于MQTT協(xié)議，首先需要傳入配置類，通過內置buildServer方法將配置類中信息綁定到當前服務中，為保證安全服務端可以設定SSL驗證保證傳輸安全。在配置類中通過MONO類的鏈式調用，將協(xié)議的處理類，以及客戶端重傳機制進行綁定。WS連接與MQTT不同之處在于底層采用http傳輸，同時WS也不需要重連機制。服務器內部封裝通過Reactor-Netty底層中與網(wǎng)絡傳輸相關的Handler處理類進行半包解析、長連接以及序列化；

主題過濾功能主要是對通配符進行匹配，其中”#“符號表示只要該符號前的信息匹配，后續(xù)字段可以忽略不計。”？“表示當前的占位符，可替代任意符號。服務器通過對通配符進行匹配，實現(xiàn)主題個性化定制，該功能主要針對大批量服務接入時需要根據(jù)主題進行客戶端分類。

borker模塊中Reactor-Netty對MQTT協(xié)議進行封裝，結合Reactor框架響應式以及背壓的特點，提升MQTT整體性能。

以Reactor-Netty框架初始化服務功能為例：

(1)服務啟動后首先調用服務端工廠方法TransportServerFactory進行服務初始化操作，MONO類進行服務的綁定以及初始化。

//初始化類，綁定相關屬性信息

Mono.from(protocolFactory.getProtocol(

ProtocolType.valueOf(config.getProtocol()))

//獲取傳輸信息

.get()>.getTransport()

//綁定Reactor內部類

.start(config,>unicastProcessor)

//netty初始化類

.map(this::>wrapper)

//錯誤信息打印

.doOnError(config.getThrowableConsumer());}

(2)調用底層Reactor-Netty方法進行初始化。創(chuàng)建Netty連接并設置定時屬性。

NettyInbound inbound = connection.getInbound();

Connection c = connection.getConnection();

// 定時關閉

Disposable disposable = Mono.fromRunnable(c::dispose)

.delaySubscription(Duration.ofSeconds(10))

.subscribe();

// 設置connection

c.channel()

.attr(AttributeKeys.connectionAttributeKey)

.set(connection);

// 設置定時關閉

c.channel()

.attr(AttributeKeys.closeConnection)

.set(disposable);

(3)設置心跳檢測以及遺囑消息處理模塊

//心跳檢測

connection.getConnection()>.onReadIdle(config.getHeart(), () -> connection.getConnection()>.dispose())

//設置遺囑消息

connection.getConnection()>.onDispose() ->

{

Optional.ofNullable(connection.getConnection()>.channel()>.attr(AttributeKeys.WILL_MESSAGE))>.map(Attribute::>get)

(4)根據(jù)QOS等級設置不同的消息處理類。

switch (qoS) {

//QOS1等級

case AT_LEAST_ONCE:

co.sendMessage(false, >qoS,>willMessage.isRetain(), willMessage.getTopicName(), willMessage.getCopyByteBuf())>.subscribe();

break;

//QOS2等級以及QOS3等級

case EXACTLY_ONCE:

case AT_MOST_ONCE:

co.sendMessageRetry(false,>qoS,>willMessage.>isRetain(), >willMessage.getTopicName(),>willMessage.getCopyByteBuf())>.

subscribe();

break;

//未傳輸時默認傳輸?shù)燃?/p>

default:

co.sendMessage(false,>qoS,>willMessage.is Retain(), willMessage.getTopicName(),>willMessage.getCopyByteBuf())>.subscribe();

break;}

})));

(5)sendMessage內部綁定了handler處理類，通過對傳輸類型的判斷設置不同的handler。

messageTypeCollection.computeIfAbsent(messageType,

type->{

//根據(jù)傳入的類型進行判斷所需要的handler

switch(type){

//數(shù)據(jù)傳輸響應

case PUBACK:

case PUBREC:

case PUBREL:

case PUBLISH:

case PUBCOMP:

return new PubHandler();

//連接請求

case CONNECT:

case DISCONNECT:

return new ConnectHandler();

//心跳檢測

case PINGREQ:

return new HeartHandler();

//服務訂閱

case SUBSCRIBE:

case UNSUBSCRIBE:

return new SubHandler();

}

經過以上操作，服務器初始化成功，等待相關請求。服務器在接收到請求后，根據(jù)數(shù)據(jù)請求類型進行相應處理。

3.3 服務器數(shù)據(jù)緩存模塊

隨著服務連接數(shù)的增多，交互數(shù)據(jù)量也逐漸增大，如果將數(shù)據(jù)都存儲在內存中，將降低整體服務的性能。項目中將數(shù)據(jù)存儲在Redis中，作為當前服務數(shù)據(jù)的緩存。Redis內部對存入信息按照key-value形式進行存儲，保證數(shù)據(jù)的唯一性。Redis是一款Nosql數(shù)據(jù)庫，在存儲方面采用的單線程處理，因此訪問時不需要進行并發(fā)維護，并且Redis在并發(fā)數(shù)據(jù)訪問以及讀取方面的性能均好于數(shù)據(jù)庫。為防止Redis系統(tǒng)崩潰導致的數(shù)據(jù)丟失，內部開啟數(shù)據(jù)持久化形式，將數(shù)據(jù)定時保存到磁盤文件中，同時采用sentinel監(jiān)控下的Redis集群模型，sentinel作為服務監(jiān)控組件主要功能為服務監(jiān)控、故障提示、故障自動轉移，通過對當前Redis集群的心跳監(jiān)控，保證當Redis主節(jié)點崩潰時，系統(tǒng)自動在其它從服務器中進行選舉，產生新的主節(jié)點，實現(xiàn)服務的可用性。Redis集群模式采用哈希槽算法對每個存入的key進行CRC16校驗后對16 384進行取模來決定放入那個服務器中。相對于傳統(tǒng)集群形式，該結構更容易擴展，如果擴充一個節(jié)點D，只需要將A、B、C節(jié)點中的部分槽放置在D上；如果想移除節(jié)點A，只需要將A的數(shù)據(jù)轉移到B和C節(jié)點上。由于將哈希槽從一個節(jié)點移動到另一個節(jié)點不需要停止服務，只需要通過命令直接再分配，因而上述擴展不會造成集群不可用，保證了服務的高可用性。

如表1所示，項目中Redis主要對以下7個內容進行存儲。

表1 Redis存儲信息

publish以及pubrel：主要是為了滿足Qos質量等級，質量等級的不同導致數(shù)據(jù)需要重傳，因此需要對傳輸當前數(shù)據(jù)進行緩，等待重新發(fā)送。

session：主要是對當前客戶端與服務器所建立的會話進行存儲。MQTT的設計主要是為了對信號不穩(wěn)定的網(wǎng)絡提供服務，因此有時會出現(xiàn)網(wǎng)絡波動導致的斷開連接情況，此時客戶端就可以在Session中將遺囑信息也進行存儲，服務器會定期對客戶端狀態(tài)進行檢查，如果出現(xiàn)異常可以向Topic中發(fā)送遺囑信息通知訂閱方，避免服務宕機而導致的訂閱方長時間等待問題。

client：訂閱者根據(jù)主題的不同實現(xiàn)訂閱服務，Redis中需要針對每個訂閱者進行唯一的標識ClientId，該標識應用Redis內部的incr方法實現(xiàn)標識號的遞增操作并與服務名進行拼接。在高并發(fā)場景下，Redis內部實現(xiàn)id的同步生成，保證線程安全。將ClientId與所訂閱的主題進行綁定，方便后續(xù)服務推送時查詢。

subWild與NotsubWild:發(fā)布方采用個性化的主題模式進行服務發(fā)布，其中可以通過#與？進行多字段匹配，與當前單獨字段匹配，因此在存儲的時候需要將有無通配符的主題進行分別存儲。

retain：該緩存主要為了broker模塊中保留信息功能，可以讓新訂閱的客戶端獲取發(fā)布方當前最新的信息，不需要進行接收等待。

3.4 消息代理分發(fā)模塊

消息代理分發(fā)模式是在硬件設備之間通信的基礎上，將數(shù)據(jù)傳入大型互聯(lián)網(wǎng)設備中進行處理。這里采用kafka作為消息代理分發(fā)的中間件，kafka用于海量數(shù)據(jù)的場景，通過分布式架構所提供的消息處理機制對數(shù)據(jù)進行流式存儲，消費者通過訂閱不同的主題可以實現(xiàn)毫秒級延遲的信息接收，保證數(shù)據(jù)順序性消費。大型企業(yè)項目中可通過kafka將該框架的推送數(shù)據(jù)轉發(fā)到hdfs歸檔或者elasticsearch做全文檢測，為大型互聯(lián)網(wǎng)設備提供高性能的數(shù)據(jù)獲取途徑。

4 測試結果

4.1 安全性測試

客戶端輸入錯誤的密碼并向服務器發(fā)送連接請求，在已知服務器的IP地址以及運行端口號的情況下進行訪問測試。測試結果如圖5所示，經過rsa私鑰解碼顯示解析后的密碼與Redis中所存儲的密碼不符，返回錯誤提示。因此該推送框架可以保證服務的安全性。

圖5 安全認證失敗

4.2 性能測試

為檢測推送服務的性能，需要對其在不同并發(fā)量情況下進行性能測試，測試采用Jmeter進行服務推送性能檢測，根據(jù)并發(fā)數(shù)量以及響應時間對框架進行分析。測試設備采用兩臺阿里云服務器，配置見表2。

Jmeter端每間隔1 s發(fā)送一個350字節(jié)的數(shù)據(jù)，在不同的并發(fā)數(shù)情況下進行測試。如圖6所示，在低并發(fā)情況下，由于系統(tǒng)中自身處理速度比較快，無法清楚看出MQTT與本框架的性能，但隨著并發(fā)數(shù)的增長，單純采用MQTT的響應時間上升較快，當并發(fā)數(shù)達到了2000時可以看到MQTT的響應時間已經遠遠超過Reactor-Netty+MQTT，同時從與Netty+MQTT的響應時間對比可以看出，該框架的響應時間也相對較低。這說明采用Reactor-Netty+MQTT的組合形式能夠極大降低高并發(fā)場景下的響應時間。

圖6 平均響應時間對比

從吞吐量方面考慮，圖7可知當并發(fā)數(shù)達到1500左右MQTT達到了性能的瓶頸期，隨著并發(fā)數(shù)的增高，服務吞吐量不再上升。而在并發(fā)數(shù)量提升到3000左右時，Netty+MQTT方案的吞吐量提升速度不如Reactor-Netty+MQTT方案快。實驗結果表明在4000并發(fā)數(shù)的時候，系統(tǒng)仍然有較好的吞吐量，并且其上升趨勢遠好于原生MQTT以及Netty+MQTT方案。這說明Reactor-Netty+MQTT的方案有良好的并發(fā)性能，適用于高并發(fā)訪問。

圖7 平均響應速度對比

5 結束語

本文介紹了一種高性能消息推送框架，基于Reactor-Netty與MQTT相結合的方案，通過Reactor-Netty響應式編程思想對MQTT協(xié)議進行封裝，實現(xiàn)異步高性能響應。應用Redis進行消息緩存，保證服務消息的實時性，設計kafka內容分發(fā)代理模式，可將數(shù)據(jù)推送服務與大型互聯(lián)網(wǎng)架構相結合，實現(xiàn)服務數(shù)據(jù)高質量處理。經過實驗驗證發(fā)現(xiàn)，Reactor-Netty實現(xiàn)的消息推送服務可以提高整體的并發(fā)處理能力，相對于傳統(tǒng)MQTT框架以及Netty框架性能有極大的提高。