基于消息中間件及MongoDB的物聯(lián)網(wǎng)應(yīng)用服務(wù)平臺①

劉 峰，王昭鵬，于 波，郭放達(dá)

1(中國科學(xué)院大學(xué)，北京 100049)

2(中國科學(xué)院 沈陽計算技術(shù)研究所，沈陽 110168)

3(東北大學(xué) 計算機(jī)科學(xué)與工程學(xué)院，沈陽 110819)

當(dāng)今物聯(lián)網(wǎng)(Internet of Things，IoT)技術(shù)儼然已經(jīng)成為蓬勃發(fā)展的技術(shù)體系[1].其應(yīng)用領(lǐng)域廣泛，涉及工業(yè)、農(nóng)業(yè)、交通、城市建設(shè)等諸多領(lǐng)域的諸多場景，因此物聯(lián)網(wǎng)平臺建設(shè)成為業(yè)界熱門課題.

世界各研究組織對物聯(lián)網(wǎng)架構(gòu)劃分相似又不盡相同，在ICT中國.2010 高層論壇上中國電信上海研究院李安民院長指出物聯(lián)網(wǎng)架構(gòu)劃分為感知層、網(wǎng)絡(luò)層、應(yīng)用層三層.目前國內(nèi)外知名物聯(lián)網(wǎng)平臺雖根據(jù)各自的技術(shù)側(cè)重對整體架構(gòu)層次有差異地進(jìn)行詳細(xì)劃分，但仍可認(rèn)為上述三層劃分是具有普適性的基本劃分.

本文依托于智慧城市項目“智能燃?xì)鈹?shù)據(jù)管理系統(tǒng)”[2]，該項目基于CoAP 協(xié)議實(shí)現(xiàn)燃?xì)庠O(shè)備智能管控.

全文首先介紹消息中間件Kafka 和MongoDB 數(shù)據(jù)庫，然后以項目為例，結(jié)合物聯(lián)網(wǎng)平臺特點(diǎn)指出架構(gòu)中存在的性能瓶頸.并針對發(fā)現(xiàn)的問題提出一種基于消息中間件Kafka 和NoSQL 數(shù)據(jù)庫MongoDB的應(yīng)用服務(wù)平臺架構(gòu).最后進(jìn)行新舊架構(gòu)數(shù)據(jù)讀寫性能比較和并發(fā)穩(wěn)定性測試，驗(yàn)證新方案的可行性和有效性.

1 技術(shù)背景

1.1 消息中間件Kafka

在大量數(shù)據(jù)傳遞場景中，消息中間件能夠使傳遞雙方單單關(guān)注數(shù)據(jù)而非如何共享，并通過維護(hù)消息隊列實(shí)現(xiàn)異步工作方式，因此消息中間件往往被簡單看作為消息隊列.其主要有兩種異步消息模式:點(diǎn)對點(diǎn)模式和發(fā)布-訂閱(Pub-Sub)模式.

Kafka 是Apache 軟件基金會開源項目，專為高并發(fā)場景而設(shè)計的分布式消息中間件[3]，它采用基于消息主題(Topic)的發(fā)布-訂閱消息模式，支持不同消費(fèi)者分布式消息讀取.與傳統(tǒng)消息隊列相比，Kafka 單點(diǎn)生產(chǎn)者消費(fèi)者達(dá)到每秒百萬級讀寫，具有更高的吞吐量；它將數(shù)據(jù)寫入磁盤并復(fù)制到集群中保證數(shù)據(jù)持久化和容錯性.

1.2 NoSQL 數(shù)據(jù)庫MongoDB

NoSQL(Not Only SQL)數(shù)據(jù)庫即非關(guān)系型數(shù)據(jù)庫，從數(shù)據(jù)模型角度分為四類:鍵值(Key-Value)型、文檔(Document)型、列式(Column Family)、圖式(Graph)[4].MongoDB 是文檔型數(shù)據(jù)庫的典型代表，兼具關(guān)系型數(shù)據(jù)庫和非關(guān)系型數(shù)據(jù)庫的部分特點(diǎn)和功能，它不需要預(yù)先定義表結(jié)構(gòu)，其數(shù)據(jù)結(jié)構(gòu)較為松散靈活，支持內(nèi)嵌文檔等復(fù)雜結(jié)構(gòu)，適合物聯(lián)網(wǎng)平臺中海量半結(jié)構(gòu)化和非結(jié)構(gòu)化異構(gòu)數(shù)據(jù)的存儲[5].

2 方案研究

2.1 現(xiàn)有方案簡介

根據(jù)物聯(lián)網(wǎng)平臺架構(gòu)整體劃分，現(xiàn)今普遍采用的整體架構(gòu)方案如圖1所示.

設(shè)備接入平臺支持物聯(lián)網(wǎng)終端設(shè)備使用標(biāo)準(zhǔn)物聯(lián)網(wǎng)協(xié)議(CoAP，MQTT 等)完成適配接入，通過協(xié)議解析提取數(shù)據(jù)，最后向應(yīng)用服務(wù)平臺進(jìn)行交付.應(yīng)用服務(wù)平臺在整體架構(gòu)中具有承上啟下的作用，一方面基于應(yīng)用服務(wù)器集群的存儲能力和計算能力對設(shè)備接入平臺提交的數(shù)據(jù)進(jìn)行持久化和數(shù)據(jù)處理，另一方面面向業(yè)務(wù)服務(wù)器提供雙向接口供第三方調(diào)用.業(yè)務(wù)層主要通過B/S Web 應(yīng)用面向各行各業(yè)，實(shí)現(xiàn)設(shè)備信息展示、設(shè)備控制操作、數(shù)據(jù)統(tǒng)計展示等功能.

2.2 基于消息中間件及MongoDB的應(yīng)用服務(wù)平臺方案

大量理論研究[6-8]和工程實(shí)踐經(jīng)驗(yàn)總結(jié)得出物聯(lián)網(wǎng)平臺具有如下特點(diǎn):第一數(shù)據(jù)量大、并發(fā)性高，據(jù)調(diào)研現(xiàn)今大型商用物聯(lián)網(wǎng)平臺需求支持至少千萬級設(shè)備長連接接入和百萬級并發(fā)能力；第二數(shù)據(jù)多源異構(gòu)、結(jié)構(gòu)松散，該特點(diǎn)主要原因是物聯(lián)網(wǎng)設(shè)備的多樣性，從而產(chǎn)生大量異構(gòu)數(shù)據(jù)無法事先統(tǒng)一結(jié)構(gòu).

通過分析現(xiàn)有方案發(fā)現(xiàn)兩處與上述特點(diǎn)不相適應(yīng)之處:第一在大規(guī)模設(shè)備接入場景中，設(shè)備接入層消息代理服務(wù)器有限的消息交付能力和沉重的工作量導(dǎo)致出現(xiàn)性能瓶頸.第二傳統(tǒng)關(guān)系型數(shù)據(jù)庫MySQL 其固定的表結(jié)構(gòu)及字段屬性無法較好地適應(yīng)異構(gòu)松散的設(shè)備數(shù)據(jù)，同時嚴(yán)格的事務(wù)約束也限制了讀寫效率.

針對上述問題，提出一種基于消息中間件Kafka 和MongoDB 數(shù)據(jù)庫的應(yīng)用服務(wù)平臺方案，改變與設(shè)備接入平臺原有的對接方式，改變數(shù)據(jù)持久化方式.該方案中采用性能更優(yōu)的擴(kuò)展鉤子(Hook)將接入平臺中相對分散的消息數(shù)據(jù)匯聚至Kafka.應(yīng)用服務(wù)器集群中各服務(wù)器節(jié)點(diǎn)分布式消費(fèi)Kafka中的消息，從而實(shí)現(xiàn)一次數(shù)據(jù)分區(qū)，并在完成相應(yīng)處理后以分片分頁策略持久化至MongoDB.

3 應(yīng)用服務(wù)平臺設(shè)計實(shí)現(xiàn)

基于本文提出的方案，應(yīng)用服務(wù)平臺架構(gòu)如圖2灰色部分所示

本方案采用擴(kuò)展鉤子與Kafka 相結(jié)合的方式完成接入平臺與應(yīng)用服務(wù)平臺對接，兩者較好的并發(fā)性能會提高數(shù)據(jù)流轉(zhuǎn)效率.同時所有數(shù)據(jù)統(tǒng)一匯集至Kafka 進(jìn)行緩沖，一方面可取消接入平臺對消息主題路由表的使用和維護(hù)，減少消息轉(zhuǎn)發(fā)的工作量.另一方面可對接入平臺數(shù)據(jù)進(jìn)行容災(zāi)保障.利用MongoDB 數(shù)據(jù)庫集合(Collection)中文檔數(shù)據(jù)結(jié)構(gòu)的靈活性，對半結(jié)構(gòu)化數(shù)據(jù)直接存儲，避免繁瑣的數(shù)據(jù)結(jié)構(gòu)轉(zhuǎn)換，化簡數(shù)據(jù)持久化過程.此外MongoDB 提供的聚合方法aggregate()可以對集合中的文檔進(jìn)行變換組合，而且支持Map-Reduce 計算模式，足以應(yīng)對業(yè)務(wù)平臺對大數(shù)據(jù)分析及海量數(shù)據(jù)高并發(fā)操作的應(yīng)用需求.

圖2 應(yīng)用服務(wù)平臺架構(gòu)

3.1 擴(kuò)展鉤子設(shè)計實(shí)現(xiàn)

擴(kuò)展鉤子負(fù)責(zé)設(shè)備接入平臺與應(yīng)用服務(wù)平臺對接，本質(zhì)是以生產(chǎn)者身份將接入平臺消息按照主題發(fā)布至Kafka 集群.擴(kuò)展鉤子采用職責(zé)鏈設(shè)計模式，使用Erlang語言實(shí)現(xiàn)[9]，通過向鉤子注冊回調(diào)函數(shù)，在接入平臺接收設(shè)備終端上下線、狀態(tài)參數(shù)更新時觸發(fā)鉤子并順序執(zhí)行回調(diào)函數(shù).事件觸發(fā)與回調(diào)函數(shù)對應(yīng)關(guān)系如表1所示.

表1 擴(kuò)展鉤子對照表

3.2 數(shù)據(jù)持久化方案設(shè)計與實(shí)現(xiàn)

本方案主要從數(shù)據(jù)格式、數(shù)據(jù)存儲模式、數(shù)據(jù)操作三方面切入:

數(shù)據(jù)格式選用JSON 格式，因其Key-Value 文本結(jié)構(gòu)與MongoDB的BSON 存儲格式極為類似，便于發(fā)揮MongoDB 松散數(shù)據(jù)結(jié)構(gòu)優(yōu)勢進(jìn)行設(shè)備數(shù)據(jù)存儲[10]，同時易于設(shè)備生成、解析，開發(fā)人員閱讀、編寫.

數(shù)據(jù)存儲選用MongoDB 數(shù)據(jù)庫，根據(jù)MongoDB官方性能測試報告顯示，在讀寫平衡場景中，MongoDB可達(dá)135 000 ops/sec.在此基礎(chǔ)上，面向設(shè)備接入平臺采用Kafka 與MongoDB 相結(jié)合方式，旨在減輕大規(guī)模數(shù)據(jù)寫入時數(shù)據(jù)庫工作負(fù)荷.首先設(shè)備數(shù)據(jù)匯聚至Kafka 緩沖，避免直接對數(shù)據(jù)庫造成沖擊.其次利用Kafka 消息隊列特性，在生產(chǎn)者-消費(fèi)者模式中應(yīng)用服務(wù)器集群各節(jié)點(diǎn)以消費(fèi)者身份消費(fèi)特定主題下消息數(shù)據(jù)，實(shí)現(xiàn)一次數(shù)據(jù)分區(qū)，完成數(shù)據(jù)在特定節(jié)點(diǎn)持久化.主題可根據(jù)具體項目應(yīng)用場景靈活設(shè)置，例如根據(jù)地理區(qū)劃、行業(yè)領(lǐng)域等設(shè)置主題.最后利用MongoDB 支持自定義存儲過程的特性，集合采用邏輯上的雙端隊列，基于最近最久未使用(LRU)算法，使用JavaScript實(shí)現(xiàn)數(shù)據(jù)分頁存儲相應(yīng)的增刪改查方法.以查找為例，按頁號遍歷文檔隊列，找到特定數(shù)據(jù)頁后pageIndex(當(dāng)前頁號)不變，修改字段lastIndex(前頁頁號)，nextIndex(后頁頁號)值，使之位于隊首，示意圖如圖3所示.

圖3 雙端隊列LRU 示意圖

數(shù)據(jù)操作主要面向業(yè)務(wù)層，采用REST 架構(gòu)風(fēng)格的Web 服務(wù)封裝簡單易用的接口并設(shè)計實(shí)現(xiàn)相應(yīng)URL 定位，通過HTTP 請求完成數(shù)據(jù)的增刪改查操作.

3.3 應(yīng)用服務(wù)平臺數(shù)據(jù)流轉(zhuǎn)過程

本文設(shè)計實(shí)現(xiàn)的應(yīng)用服務(wù)平臺數(shù)據(jù)流轉(zhuǎn)過程主要包括數(shù)據(jù)采集匯聚、數(shù)據(jù)處理、數(shù)據(jù)分區(qū)、數(shù)據(jù)持久化.擴(kuò)展鉤子通過事件觸發(fā)機(jī)制完成數(shù)據(jù)采集；Kafka采用生產(chǎn)者消費(fèi)者模式完成數(shù)據(jù)匯聚和數(shù)據(jù)分區(qū)，同時可對數(shù)據(jù)進(jìn)行處理；MongoDB 基于文檔模型采用分頁方式完成數(shù)據(jù)持久化，為業(yè)務(wù)層提供可用數(shù)據(jù)；整體數(shù)據(jù)流轉(zhuǎn)時序如圖4所示.

4 測試

測試機(jī)配置:Intel(R)Core(TM)i5-4200U CPU @1.60 GHz，RAM 8 GB DDR3，硬盤240 GB SATA，系統(tǒng)CentOS7.

本文基于消息中間件及MongoDB，針對原有物聯(lián)網(wǎng)平臺架構(gòu)中應(yīng)用服務(wù)平臺進(jìn)行改進(jìn)，并結(jié)合智能燃?xì)鈹?shù)據(jù)管理系統(tǒng)具體性能需求完成搭建部署.系統(tǒng)運(yùn)行如圖5所示.

為驗(yàn)證本文方案的有效性，通過使用Web Bench壓測工具進(jìn)行百萬級數(shù)據(jù)讀寫性能對比測試，結(jié)果如圖6所示.使用emqtt_benchmark 工具進(jìn)行平臺穩(wěn)定性測試，結(jié)果如表2所示.以上結(jié)果表明本方案中數(shù)據(jù)庫持久化效率雖有波動但得到明顯提升；在當(dāng)前測試條件下平臺服務(wù)器節(jié)點(diǎn)CPU 利用率收斂于59%，內(nèi)存占用處于合理范圍，平臺吞吐量總體趨于平穩(wěn)，可見平臺具有較好的性能穩(wěn)定性.

圖4 數(shù)據(jù)流轉(zhuǎn)時序圖

圖5 系統(tǒng)運(yùn)行效果

圖6 數(shù)據(jù)庫讀寫性能對比

表2 平臺穩(wěn)定性測試

5 結(jié)論

本文依據(jù)物聯(lián)網(wǎng)平臺特點(diǎn)，分析常見物聯(lián)網(wǎng)平臺架構(gòu)性能瓶頸成因，針對應(yīng)用服務(wù)層提出一種基于消息中間件Kafka 及NoSQL 數(shù)據(jù)庫MongoDB的應(yīng)用服務(wù)平臺方案，設(shè)計并實(shí)現(xiàn)了擴(kuò)展鉤子和數(shù)據(jù)持久化模式，改變了接入平臺與應(yīng)用服務(wù)平臺之間的對接方式，改變了數(shù)據(jù)流轉(zhuǎn)方式和持久化方式.實(shí)驗(yàn)結(jié)果證明，本文方案切實(shí)有效，平臺并發(fā)性能穩(wěn)定、持久化效率得到提高，并且后續(xù)開發(fā)維護(hù)難度降低，更好地支持業(yè)務(wù)擴(kuò)展.

后期工作將充分利用Kafka 在數(shù)據(jù)流實(shí)時處理過程中的優(yōu)勢，進(jìn)一步提升平臺性能并為業(yè)務(wù)層提供更多數(shù)據(jù)支持.