基于通用消息的持久化消息隊(duì)列設(shè)計(jì)

郭盛興，　王　晶,　廖建新

(1. 北京郵電大學(xué) 網(wǎng)絡(luò)與交換技術(shù)國家重點(diǎn)實(shí)驗(yàn)室， 北京　100876；2. 東信北郵信息技術(shù)有限公司， 北京　100191)

中間件是一種定義于操作系統(tǒng)之上，應(yīng)用程序之下的一層軟件，它能使應(yīng)用之間進(jìn)行跨網(wǎng)絡(luò)協(xié)同工作，屏蔽了操作系統(tǒng)和網(wǎng)絡(luò)協(xié)議的差異向應(yīng)用提供通信服務(wù)[1]. 消息中間件支持在一個分布式應(yīng)用環(huán)境中多種用途的消息交換. 它所提供的API將不同分布式環(huán)境很好地封裝起來，對外提供統(tǒng)一的接口，使得應(yīng)用能通過統(tǒng)一接口進(jìn)行開發(fā)[2].

通用消息(component packet of realtime application process management and communication,COPART-MACO)是一個抽象得比較好的消息中間件. 對不同類型的上層消息增加同樣的底層消息頭，如圖1，這種消息稱為通用消息.

使用通用消息的進(jìn)程間交互采用統(tǒng)一的協(xié)議，底層采用統(tǒng)一的方式通信，軟件功能差異主要體現(xiàn)在高層的消息處理部分[3]. 通用消息可以滿足不同進(jìn)程間通信需要，但通用消息是面向無連接的，只提供了消息尋址功能，不保證通信的可靠性，對比TCP/IP協(xié)議來說，相當(dāng)于實(shí)現(xiàn)了IP層功能. 為滿足可靠性要求，提出了基于通用消息的持久化隊(duì)列(Ebupt Message Queue簡稱EMQ)設(shè)計(jì)，EMQ由服務(wù)端和客戶端組成，服務(wù)端提供集中式的消息接收、存儲和轉(zhuǎn)發(fā)服務(wù)，客戶端可以通過服務(wù)端發(fā)送和接收消息，客戶端和服務(wù)端之間的通信采用請求應(yīng)答方式，服務(wù)端具有消息持久化功能.

圖1　通用消息的消息結(jié)構(gòu)Fig.1　COPART-MACO packet encapsulation

1　EMQ架構(gòu)設(shè)計(jì)

1.1　EMQ的部署結(jié)構(gòu)和組件

EMQ部署結(jié)構(gòu)如圖2，其中包含3個域：消息隊(duì)列服務(wù)域、生產(chǎn)者域、消費(fèi)者域.

圖2　EMQ部署結(jié)構(gòu)Fig.2　EMQ deployment structure

包含的組件說明：

1) ininit，消息隊(duì)列的守護(hù)進(jìn)程，負(fù)責(zé)啟動本域內(nèi)的所有其他進(jìn)程，在子進(jìn)程異常退出后，重新啟動子進(jìn)程;

2) inaccessd，控制臺接入服務(wù)端進(jìn)程，接入它之后，可以建立與本域內(nèi)的所有進(jìn)程的連接，輸入控制臺命令得到應(yīng)答;

3) msgr，通用消息的消息分發(fā)模塊，用于轉(zhuǎn)發(fā)本域內(nèi)的進(jìn)程和外部進(jìn)程之間的交互消息，根據(jù)實(shí)際需要，也可以有一個或者多個;

4) emqserver，EMQ服務(wù)端，持久化消息隊(duì)列的核心，負(fù)責(zé)接收、存儲和轉(zhuǎn)發(fā)消息;

5) emqclient，與EMQ服務(wù)端交互的客戶端，EMQ客戶端分為生產(chǎn)者和消費(fèi)者，生產(chǎn)者發(fā)送消息給消息隊(duì)列服務(wù)端，消費(fèi)者從通用消息服務(wù)端接收消息并處理.

1.2　EMQ服務(wù)端內(nèi)部結(jié)構(gòu)

EMQ服務(wù)端的結(jié)構(gòu)如圖3：

圖3　持久化消息隊(duì)列服務(wù)端結(jié)構(gòu)Fig.3　EMQ server structure

服務(wù)端底層通信鏈路層是通用消息層，EmqManager管理EMQ服務(wù)端的兩個最主要的數(shù)據(jù)結(jié)構(gòu)，與客戶端的連接EmqConnection和持久化的消息隊(duì)列EmqQueue，連接與消息隊(duì)列間通過隊(duì)列名QueueName相互關(guān)連.

1.3　EMQ虛擬連接

通用消息是面向無連接的，所以EMQ客戶端與服務(wù)端之間建立的是虛擬連接. 在通用消息的結(jié)構(gòu)下，每個進(jìn)程都有唯一的進(jìn)程地址標(biāo)識，包括三部分：域編號，功能實(shí)體編號，和進(jìn)程實(shí)例編號. 一條EMQ虛擬連接對應(yīng)一對進(jìn)程地址標(biāo)識. 連接建立的時候，客戶端請求參數(shù)中含有隊(duì)列名，服務(wù)端保存與客戶端的連接、進(jìn)程地址標(biāo)識與隊(duì)列的對應(yīng)關(guān)系，客戶端收到服務(wù)端連接響應(yīng)后保存與服務(wù)端的連接信息. 對于JMS[4]中一個客戶端進(jìn)程與服務(wù)端一個隊(duì)列建立多條連接的情況，如Java編程中客戶端進(jìn)程采用多線程，這時每個線程會與服務(wù)端的一個隊(duì)列有一個虛擬連接，將每一個線程作為通用消息的一個實(shí)例.

1.4　EMQ持久化隊(duì)列

考慮通用性，持久化隊(duì)列設(shè)計(jì)成相對獨(dú)立的模塊，按照EMQ消息持久化所要求的接口能力提供操作接口. EMQ需要持久化隊(duì)列實(shí)現(xiàn)提供的接口主要包括：createQueue(創(chuàng)建隊(duì)列)、destroyQueue(刪除隊(duì)列)、enQueue(消息入隊(duì))、deQueue(消息出隊(duì))、recover(恢復(fù)隊(duì)列)等. 研究給出兩種持久化隊(duì)列可行的實(shí)現(xiàn)方式：

1) 基于文件和索引的實(shí)現(xiàn)

定義每條消息的持久化存儲結(jié)構(gòu)，將通用消息整個消息的長度、消息頭和消息體及消息是否已經(jīng)被“消費(fèi)”的標(biāo)識存儲到文件，在內(nèi)存中保存每條消息在文件中的位置索引，讀取消息時直接通過索引根據(jù)存儲結(jié)構(gòu)讀取消息. 此方式在消息量相對比較小的情況下是一種比較好的選擇.

2) 基于開源項(xiàng)目的實(shí)現(xiàn)

基于SQLite來實(shí)現(xiàn)持久化隊(duì)列的存儲，在SQLite基礎(chǔ)上做一層封裝，提供EMQ所需的接口. 基于SQLite實(shí)現(xiàn)的持久化隊(duì)列，對于比較大的數(shù)據(jù)量也能獲得比較好的性能. 除SQLite外，還可以考慮Berkeley DB、redis等其他開源項(xiàng)目，在其開放的API基礎(chǔ)上封裝為EMQ提供所需的接口即可.

2　消息交互流程設(shè)計(jì)

生產(chǎn)者與消費(fèi)者間核心的消息交互流程如圖4.

1) 生產(chǎn)者依據(jù)負(fù)荷分擔(dān)策略從幾個服務(wù)端連接中選擇一個連接，發(fā)送消息請求;

2) 服務(wù)端收到消息后，找到消費(fèi)者進(jìn)程地址標(biāo)識對應(yīng)的持久化隊(duì)列，并將消息持久化保存到隊(duì)列中;

3) 服務(wù)端保存消息后將操作結(jié)果響應(yīng)給生產(chǎn)者;

4) 生產(chǎn)者依據(jù)響應(yīng)結(jié)果決定消息如何處理;

5) 當(dāng)消費(fèi)者連接對應(yīng)的隊(duì)列中有數(shù)據(jù)時，服務(wù)端將每次取出一條消息發(fā)送給消費(fèi)者;

6) 消費(fèi)者收到消息后首先給服務(wù)端消費(fèi)消息響應(yīng);

圖4　消息交互流程Fig.4　Diagram of message flow

7) 消費(fèi)者對收到的消息進(jìn)行“消費(fèi)”處理.

3　接口設(shè)計(jì)

3.1　接口概述

EMQ客戶端和服務(wù)端的通信多數(shù)采用請求—應(yīng)答方式. 定義EMQ客戶端和EMQ服務(wù)端的消息格式如圖5.

圖5　消息格式Fig.5　Message format

1) 操作類型：用于區(qū)分對隊(duì)列的不同操作，為一字節(jié)整型;

2) 操作參數(shù)：可以為任意長度，但是由于EMQ消息使用通用消息的消息體承載，所以操作參數(shù)最大長度受通用消息一個包的最大長度限制.

3.2　操作類型

EMQ客戶端與服務(wù)端之間接口消息的操作類型編碼和操作參數(shù)定義如表1.

3.3　操作結(jié)果

操作結(jié)果0表示成功，非0表示失敗，表2是各種主要操作結(jié)果代碼及說明.

4　流量控制與服務(wù)質(zhì)量

流量控制機(jī)制理想情況下可以指示任意時刻發(fā)送方發(fā)送消息的速率或數(shù)量，確保接受方的資源不被耗盡[5].

表1　操作類型定義Tab.1　Define of message operations

表2　操作結(jié)果代碼定義Tab.2　Result code of operation

服務(wù)端采用基于滑動窗口的流量控制機(jī)制[6]，針對每條消費(fèi)者連接設(shè)置一個發(fā)送窗口，當(dāng)發(fā)送出去但還沒有收到響應(yīng)的消息數(shù)量達(dá)到窗口大小或收到操作結(jié)果代碼為“超過接收窗口大小”的應(yīng)答時，服務(wù)端延遲發(fā)送消息. 對于每條生產(chǎn)者連接，服務(wù)端采用接收窗口來進(jìn)行流量控制，當(dāng)收到一條消息，如果未發(fā)送應(yīng)答的消息數(shù)量達(dá)到窗口大小時，應(yīng)答“超過接收窗口大小”的操作結(jié)果代碼，發(fā)送方則延遲發(fā)送消息.

消費(fèi)者采用服務(wù)端同樣的流量控制機(jī)制和錯誤超時重傳策略.

服務(wù)端與客戶端采用請求-應(yīng)答機(jī)制，當(dāng)應(yīng)答錯誤和應(yīng)答超時通過消息重傳來保證消息不被丟失，而且EMQ隊(duì)列服務(wù)端具有持久化功能，因此可以有效保證服務(wù)端與客戶端之間消息通信的服務(wù)質(zhì)量.

5　結(jié)束語

為發(fā)揮通用消息的作為底層通信模塊的優(yōu)勢，同時彌補(bǔ)其在服務(wù)質(zhì)量方面的不足，在通用消息基礎(chǔ)上，提出了一種消息隊(duì)列EMQ的設(shè)計(jì). EMQ服務(wù)端與客戶端之間基于請求-應(yīng)答方式，當(dāng)請求應(yīng)答錯誤和應(yīng)答超時進(jìn)行消息重傳，服務(wù)端對消息持久化，這些機(jī)制可以更加有效地保證消息隊(duì)列的服務(wù)質(zhì)量.

在實(shí)際應(yīng)用中，設(shè)計(jì)提高了通用消息的服務(wù)質(zhì)量，但由于消息持久化的引入?yún)s付出了性能的代價(jià)，面對當(dāng)前各種應(yīng)用中越來越大量的交互消息及實(shí)時性要求，性能又是非常重要的，所以研究更高效率持久化隊(duì)列，設(shè)計(jì)一種無論對小數(shù)據(jù)量還是大數(shù)據(jù)量消息交互均有良好的性能表現(xiàn)的隊(duì)列，是后續(xù)研究重要的努力方向.