劉 峰，王昭鵬，于 波，郭放達

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

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

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

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

世界各研究組織對物聯網架構劃分相似又不盡相同，在ICT中國.2010 高層論壇上中國電信上海研究院李安民院長指出物聯網架構劃分為感知層、網絡層、應用層三層.目前國內外知名物聯網平臺雖根據各自的技術側重對整體架構層次有差異地進行詳細劃分，但仍可認為上述三層劃分是具有普適性的基本劃分.

本文依托于智慧城市項目“智能燃氣數據管理系統”[2]，該項目基于CoAP 協議實現燃氣設備智能管控.

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

1 技術背景

1.1 消息中間件Kafka

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

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

1.2 NoSQL 數據庫MongoDB

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

2 方案研究

2.1 現有方案簡介

根據物聯網平臺架構整體劃分，現今普遍采用的整體架構方案如圖1所示.

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

2.2 基于消息中間件及MongoDB的應用服務平臺方案

大量理論研究[6-8]和工程實踐經驗總結得出物聯網平臺具有如下特點:第一數據量大、并發(fā)性高，據調研現今大型商用物聯網平臺需求支持至少千萬級設備長連接接入和百萬級并發(fā)能力；第二數據多源異構、結構松散，該特點主要原因是物聯網設備的多樣性，從而產生大量異構數據無法事先統一結構.

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

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

3 應用服務平臺設計實現

基于本文提出的方案，應用服務平臺架構如圖2灰色部分所示

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

圖2 應用服務平臺架構

3.1 擴展鉤子設計實現

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

表1 擴展鉤子對照表

3.2 數據持久化方案設計與實現

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

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

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

圖3 雙端隊列LRU 示意圖

數據操作主要面向業(yè)務層，采用REST 架構風格的Web 服務封裝簡單易用的接口并設計實現相應URL 定位，通過HTTP 請求完成數據的增刪改查操作.

3.3 應用服務平臺數據流轉過程

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

4 測試

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

本文基于消息中間件及MongoDB，針對原有物聯網平臺架構中應用服務平臺進行改進，并結合智能燃氣數據管理系統具體性能需求完成搭建部署.系統運行如圖5所示.

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

圖4 數據流轉時序圖

圖5 系統運行效果

圖6 數據庫讀寫性能對比

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

5 結論

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

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