孫艷華　王麗娜

摘要：隨著物聯(lián)網(wǎng)建設(shè)的不斷發(fā)展，以實(shí)時傳感數(shù)據(jù)為基礎(chǔ)的各類流數(shù)據(jù)如何及時快速存儲已成為當(dāng)前急需解決的問題。面對連續(xù)不斷到達(dá)的數(shù)據(jù)流，數(shù)據(jù)流存儲系統(tǒng)必須及時快速將其存儲到持久化介質(zhì)中。針對這一需求，本文提出了一種基于緩存的分布式實(shí)時存儲方法DRS。DRS方法依據(jù)窗口閾值，不間斷地從緩存區(qū)接收數(shù)據(jù)，并結(jié)合一致性哈希算法將數(shù)據(jù)分布存儲到數(shù)據(jù)服務(wù)器中。實(shí)驗(yàn)表明，該分布式存儲方法能夠?qū)崟r地將數(shù)據(jù)存儲到持久化介質(zhì)中。

關(guān)鍵字：流數(shù)據(jù)；實(shí)時存儲；閾值；一致性哈希

中圖分類號：TP311 文獻(xiàn)標(biāo)識碼：A 文章編號：1009-3044（2015）19-0005-02

1引言

隨著物聯(lián)網(wǎng)建設(shè)地不斷發(fā)展，流數(shù)據(jù)逐漸成為信息化系統(tǒng)中一類關(guān)鍵數(shù)據(jù)形式。該類數(shù)據(jù)如何及時存儲已成為當(dāng)前物聯(lián)網(wǎng)應(yīng)用建設(shè)中的一個重點(diǎn)。對于實(shí)時數(shù)據(jù)存儲系統(tǒng)而言，首先要確保數(shù)據(jù)及時存儲，從而保證數(shù)據(jù)的實(shí)時性。由于數(shù)據(jù)延遲存儲，造成數(shù)據(jù)處理的非正確性，做出錯誤的決策。

以一個物聯(lián)網(wǎng)環(huán)境下的地鐵站交通監(jiān)管數(shù)據(jù)流存儲任務(wù)為例，該任務(wù)需要監(jiān)管以下幾類數(shù)據(jù)：1）客流；2）基礎(chǔ)設(shè)施：扶梯，售票機(jī)，防火設(shè)備，照明燈等；3）車流：地鐵，公交車，出租車等。收集并將該流數(shù)據(jù)及時存儲。在該背景下，設(shè)計(jì)一種實(shí)時存儲方法成為必要。

2交通樞紐實(shí)時監(jiān)管系統(tǒng)

交通監(jiān)管系統(tǒng)通過具體設(shè)備進(jìn)行數(shù)據(jù)監(jiān)測，存儲，處理，對交通事件做出及時具體的措施。該系統(tǒng)采用了分布式的并行架構(gòu)。其結(jié)構(gòu)如圖1所示。

3DRS方法的基本思想

車輛實(shí)時監(jiān)管系統(tǒng)分布式架構(gòu)中需要存儲大規(guī)模實(shí)時數(shù)據(jù)，而這些數(shù)據(jù)具有實(shí)時性，高速性，因此我們提出的存儲方法首要考慮存儲的實(shí)時性。DRS方法的基本思想如下：

利用Map Reduce思想設(shè)計(jì)DRS方法，其方法的設(shè)計(jì)思路：數(shù)據(jù)接收階段對數(shù)據(jù)做簡單的劃分工作，數(shù)據(jù)寫入階段則對已劃分?jǐn)?shù)據(jù)進(jìn)行處理。以車流數(shù)據(jù)為例，當(dāng)數(shù)據(jù)接收區(qū)監(jiān)聽到數(shù)據(jù)緩沖區(qū)有數(shù)據(jù)時，數(shù)據(jù)接收區(qū)開始接收數(shù)據(jù)。接收數(shù)據(jù)的同時，存儲程序生成多個存放數(shù)據(jù)的隊(duì)列，根據(jù)數(shù)據(jù)的時間標(biāo)示，將該條數(shù)據(jù)放置到對應(yīng)的隊(duì)列中。當(dāng)隊(duì)列中數(shù)據(jù)到達(dá)到窗口閾值后，數(shù)據(jù)寫入?yún)^(qū)解析存儲該隊(duì)列中的數(shù)據(jù)。各個隊(duì)列中數(shù)據(jù)對象發(fā)送給各個數(shù)據(jù)寫入?yún)^(qū)，數(shù)據(jù)寫入?yún)^(qū)解析各個數(shù)據(jù)對象，將其同時存儲到不同數(shù)據(jù)服務(wù)器中，實(shí)現(xiàn)高速率數(shù)據(jù)快速存儲。

窗口閾值采用時間和數(shù)量閥值，當(dāng)發(fā)送的數(shù)據(jù)在規(guī)定的時間閥值到達(dá)之前已經(jīng)達(dá)到數(shù)量閾值，則立即將該部分?jǐn)?shù)據(jù)進(jìn)行解析存儲，反之亦然。

4DRS方法的實(shí)現(xiàn)

4.1數(shù)據(jù)劃分策略

各個隊(duì)列中數(shù)據(jù)對象在內(nèi)存中采用鏈表結(jié)構(gòu)存儲。根據(jù)第三節(jié)對DRS方法思想的描述可知，數(shù)據(jù)接收階段接收數(shù)據(jù)的同時要對數(shù)據(jù)進(jìn)行劃分操作。該方法使用數(shù)據(jù)的唯一標(biāo)示劃分?jǐn)?shù)據(jù)，將該條數(shù)據(jù)劃分到對應(yīng)的數(shù)據(jù)隊(duì)列中。以車流為例描述該劃分策略：首先，獲取該條數(shù)據(jù)的時間戳屬性值，進(jìn)行hash直接取余，根據(jù)hash值將該條數(shù)據(jù)放置到對應(yīng)的隊(duì)列中。

當(dāng)某個隊(duì)列數(shù)據(jù)滿足某個窗口閾值時，數(shù)據(jù)接收區(qū)會給數(shù)據(jù)對象加上標(biāo)示key。每個數(shù)據(jù)寫入?yún)^(qū)只負(fù)責(zé)某個隊(duì)列數(shù)據(jù)對象的解析存儲。

4.2數(shù)據(jù)寫入

數(shù)據(jù)接收區(qū)將該段時間內(nèi)的數(shù)據(jù)劃分后，數(shù)據(jù)寫入?yún)^(qū)接收分片數(shù)據(jù)。使用一致性哈希算法將各個分片數(shù)據(jù)有組織地存儲到數(shù)據(jù)服務(wù)器中。每個服務(wù)器節(jié)點(diǎn)可以對等均勻地分布數(shù)據(jù)。

使用一致性哈希計(jì)算時，數(shù)據(jù)寫入?yún)^(qū)將數(shù)據(jù)寫入到哪臺數(shù)據(jù)服務(wù)器不僅僅依賴標(biāo)示key本身而將數(shù)據(jù)服務(wù)器的配置（IP地址或者機(jī)器名）也參與hash運(yùn)算。單個數(shù)據(jù)服務(wù)器節(jié)點(diǎn)只需要專注于自身存儲工作，從而實(shí)現(xiàn)將復(fù)雜的分布式架構(gòu)邏輯與單個服務(wù)節(jié)點(diǎn)數(shù)據(jù)處理相分離，最大限度地利用現(xiàn)有的數(shù)據(jù)庫存儲技術(shù)。

數(shù)據(jù)存儲流水線可以看做一個由數(shù)據(jù)監(jiān)測，數(shù)據(jù)接收，數(shù)據(jù)劃分，數(shù)據(jù)寫入四個部分組成的串聯(lián)系統(tǒng)。初始化數(shù)據(jù)發(fā)送速率，隊(duì)列劃分個數(shù)和窗口閾值大小，運(yùn)行存儲程序。

4.3DRS方法實(shí)現(xiàn)

基于上述理論，該DRS方法的實(shí)現(xiàn)如下：

首先，首先給幾類流數(shù)據(jù)設(shè)定標(biāo)示，如給流數(shù)據(jù)設(shè)置屬性值，并配置其值，然后通過屬性值將數(shù)據(jù)分類，即擁有相同屙性值的數(shù)據(jù)為相同類。

其次，數(shù)據(jù)處理機(jī)對某類數(shù)據(jù)處理之后，結(jié)果集也賦予相同的Attribute value值，將其一起發(fā)送到數(shù)據(jù)緩沖區(qū)，最后，存儲程序?qū)邮盏臄?shù)據(jù)進(jìn)行存儲，流程如下：

傳感設(shè)備發(fā)送數(shù)據(jù)或者數(shù)據(jù)處理機(jī)發(fā)送數(shù)據(jù)時，首先將數(shù)據(jù)發(fā)往數(shù)據(jù)緩沖區(qū)；

（1）數(shù)據(jù)接收區(qū)一直監(jiān)聽數(shù)據(jù)緩沖區(qū)的數(shù)據(jù)，監(jiān)測到數(shù)據(jù)緩沖區(qū)中有數(shù)據(jù)，則開始接收數(shù)據(jù)并初始化存放數(shù)據(jù)的隊(duì)列；

（2）如果數(shù)據(jù)緩沖區(qū)監(jiān)測沒有此類數(shù)據(jù)的到來，則一直等待，直到發(fā)送端發(fā)送數(shù)據(jù)；

（3）存儲程序根據(jù)時間戳屬性計(jì)算出該條數(shù)據(jù)發(fā)往的對應(yīng)隊(duì)列；

f4）如果某個隊(duì)列數(shù)據(jù)到達(dá)時間窗口閾值，無論數(shù)據(jù)量是否滿足數(shù)據(jù)量閾值，數(shù)據(jù)接收區(qū)均給該數(shù)據(jù)對象配置標(biāo)示key；

（5）數(shù)據(jù)寫入?yún)^(qū)獲取該數(shù)據(jù)對象，根據(jù)標(biāo)示key和數(shù)據(jù)服務(wù)器列表，進(jìn)行一致性哈希計(jì)算，根據(jù)hash值將該數(shù)據(jù)對象存放到對應(yīng)數(shù)據(jù)服務(wù)器中；

（6）如果到達(dá)數(shù)據(jù)量窗口閾值，無論時間是否滿足時間窗口閾值，數(shù)據(jù)寫入?yún)^(qū)都將進(jìn)行取數(shù)據(jù)對象操作。

5實(shí)驗(yàn)設(shè)計(jì)及結(jié)果

以物聯(lián)網(wǎng)下交通樞紐監(jiān)測系統(tǒng)下車流的存儲為例測試驗(yàn)證DRS方法。

實(shí)驗(yàn)準(zhǔn)備：2臺雙核3.0GHz CPU和4GB內(nèi)存的機(jī)器上安裝Load Runner9.0，模擬數(shù)據(jù)流發(fā)送端；1臺2x4核2.4GHzCPU，16GB內(nèi)存的機(jī)器上安裝第三方緩存ActiveMQ中間件，作為緩存服務(wù)器；一臺2x4核2.4GHz CPU，16GB內(nèi)存的機(jī)器上安裝存儲程序；6臺2x4核2.4GHz CPU，16GB內(nèi)存1TBRAID5磁盤陣列的機(jī)器上安裝oracle10g，作為持久化存儲。

實(shí)驗(yàn)1：使用DRS方法進(jìn)行低中速率數(shù)據(jù)存儲測試，查看效果：使用Load Runner服務(wù)器模擬數(shù)據(jù)流發(fā)送時，設(shè)置不同的數(shù)據(jù)發(fā)送速率，窗口時間閾值設(shè)定為2s，數(shù)據(jù)量閾值設(shè)定為2000條。持續(xù)運(yùn)行10小時。

實(shí)驗(yàn)結(jié)果：

圖2中，橫軸為數(shù)據(jù)發(fā)送速率，即每秒發(fā)送的數(shù)據(jù)量，單位為條/秒，表示速率不斷提高，縱軸為延遲時間，單位為小時。由圖2得知，將流數(shù)據(jù)劃分并考慮將數(shù)據(jù)分布存儲之后，低中速數(shù)據(jù)無延遲的存儲到數(shù)據(jù)服務(wù)器中，這大大提高了系統(tǒng)性能，保證了數(shù)據(jù)的完整性，實(shí)時性。

實(shí)驗(yàn)2：使用DRS方法進(jìn)行萬級別發(fā)送速率的存儲測試，查看效果：使用Load Runner服務(wù)器模擬數(shù)據(jù)流發(fā)送時，設(shè)置不同的數(shù)據(jù)發(fā)送速率與不同的數(shù)據(jù)量閾值，每秒數(shù)據(jù)發(fā)送量依次設(shè)置為1w，2w，3w，4w，5w，6w，7w，8w。數(shù)據(jù)量閾值分別設(shè)為2000條，3000條，5000條，在這三種閾值下分別進(jìn)行測試。時間閾值設(shè)定為2s，持續(xù)運(yùn)行10小時。

圖3中，橫軸為數(shù)據(jù)發(fā)送速率，即每秒發(fā)送的數(shù)據(jù)量，單位為條/秒，表示速率不斷提高，縱軸為延遲時間，單位為小時。由結(jié)果圖可知使用DRS方法進(jìn)行存儲也出現(xiàn)延遲現(xiàn)象，延遲時間隨著速率不斷提高而增長。當(dāng)數(shù)據(jù)速率提高到更高級別時，由于數(shù)據(jù)劃分策略緣故延遲現(xiàn)象不會完全消除，這也是本論文下一步將要解決的問題。

6結(jié)論

本文提出了分布式實(shí)時存儲方法DRS，該方法基于實(shí)時車輛數(shù)據(jù)進(jìn)行了一系列實(shí)驗(yàn)。結(jié)果證明該方法在大部分情況下?lián)碛辛己玫男阅?，并?yīng)用于某市交通監(jiān)管實(shí)時系統(tǒng)中，成效顯著，提高了系統(tǒng)的性能。

本文下一步工作將針對于閾值的設(shè)定，數(shù)據(jù)的劃分與發(fā)送數(shù)據(jù)的速率之間的關(guān)系做進(jìn)一步分析，使該方法更有使用價值。