基于海量異構傳感器的物聯網水質監測系統

陳 玥 王海峰 賈建鑫 康 凱*

1(東南大學計算機科學與工程學院 江蘇 南京 211189)2(中國科學院上海微系統與信息技術研究所 上海 200050)3(中國科學院上海高等研究院 上海 201210)

0 引 言

當前，隨著我國城市化建設的快速推進，水資源短缺與水環境污染已成為城市化建設面臨的關鍵問題。水資源短缺與水環境污染不僅會阻礙經濟建設，更會對國民健康帶來危害。因此，針對水資源進行有效的監測與管理顯得至關重要。

另一方面，隨著國內外物聯網技術[1-5]的快速發展，物聯網技術被廣泛應用于水資源監測領域[6-9]，其在很大程度上促進了水資源監測技術的發展。物聯網作為一種新興的技術領域，受到國家的高度重視。國家工信部發布了《物聯網“十二五”發展規劃》，將智能環保作為一項重點示范工程。智能環保旨在利用物聯網相關技術實現污染源監控、水質監測、空氣監測、生態監測并建立智能環保信息采集網絡與信息平臺。水質監測和水污染監控無疑是智能環保的一個重要組成部分。因此，研究基于物聯網技術的水資源監測平臺勢在必行。鑒于此，本文基于物聯網平臺相關理論技術，設計并實現了一種基于海量異構傳感器的物聯網水質監測系統，根據圖1所示的云-管-端體系，將該系統分為4層，包括感知層、網絡層、平臺層、應用層。其中感知層主要由各種傳感器節點構成，其主要負責對物理世界進行信息收集與信號處理；網絡層主要由各類傳輸網絡構成，負責終端接入與數據傳輸；平臺層主要由數據庫服務器和業務服務器構成，負責設備通信管理，數據存儲，業務處理規劃等，應用層由各類可視化控件構成，負責數據呈現及客戶交互。

圖1 基于云-管-端體系的物聯網平臺層次劃分

本文的主要貢獻如下：

(1) 基于云-管-端模型提出了物聯網水質監測系統的總體架構，將物聯網水質監測系統分為感知層、網絡層、平臺層、應用層四部分。在感知層，分別采用2G、NB-IoT[10-12]、LoRa[13-15]等多種不同網絡制式的傳感器設備進行水質數據的采集；在網絡層，分別通過2G傳輸網絡、NB-IoT傳輸網絡、LoRa傳輸網絡對水質遙測數據及傳感器設備數據進行傳輸；在平臺層，通過數據庫服務器、云服務器、NB-IoT服務器、以及LoRa服務器對網絡層上報的相關數據進行存儲與處理；在應用層，通過圖表、地圖、視頻等可編程顯示部件來向用戶提供可視化的水質監測信息，并可根據平臺層定義的規則引擎觸發警報實時地向用戶推送短信和郵件通知。

(2) 基于系統總體架構，進一步對系統平臺層的關鍵技術進行了詳細介紹。首先，平臺層的數據庫服務器通過TCP socket模塊以及多線程機制與2G傳感器進行交互，包括將2G網絡傳輸來的水質遙測數據信息與設備信息進行存儲以及針對2G傳感器設備下發指令信息。與此同時，數據庫服務器中的North API client通過調用北向接口North API與NB-IoT服務器及LoRa服務器進行交互，包括從NB-IoT服務器及LoRa服務器中讀取NB-IoT傳感器設備及LoRa傳感器設備采集的遙測信息和設備信息，將讀取后的信息進行解析與存儲，以及針對NB-IoT傳感器設備及LoRa傳感器設備的指令下發。其次，對于數據庫服務器中存儲的設備信息與水質遙測信息，在數據庫服務器中通過Node平臺以Javascript搭配MQTT的方式實現與云服務器中開源IoT平臺-Thingsboard[16-17]的交互，Thingsboard平臺中定義的規則引擎會對數據庫服務器發布的設備遙測數據信息進行過濾，對于異常的遙測數據其會觸發警報并通過SMTP和RESTful向用戶發送警報郵件和警報短信。最后，在Thingsboard平臺中，除其自帶的圖表可顯示部件外，對其地圖部件進行了修改，消除了鑒權失敗的錯誤并且增強了污染范圍的渲染效果，此外，還通過可編程部件Widget加入了視頻顯示控件，其可向用戶提供實時的水質外景監測。

(3) 通過實驗案例來對水質監測系統進行功能測試，通過收集實時的水質監測數據來觸發警報并向用戶推送警報短信與警報郵件。相關實驗結果表明，該系統可快速有效地向用戶提供水質監測與報警信息。

1 系統總體架構

基于圖2所示的系統總體架構，在感知層，分別采用2G傳感器節點與2G匯聚節點、NB-IoT傳感器節點與NB-IoT匯聚節點，以及LoRa傳感器節點與LoRa匯聚節點對需要監測的水質數據進行采集，包括水的溶解氧濃度(DO)、氫離子濃度指數(PH)、水溫(TEMP)、氧化還原電位(ORP)、電導率(EC)、化學需氧量(COD)、生化需氧量(BOD)、懸浮物(SS)、水質色度(WC)等。

圖2 基于海量異構傳感器的物聯網水質監測系統總體架構

在網絡層，通過2G傳輸網絡、NB-IoT傳輸網絡以及LoRa傳輸網絡對感知層采集的水質監測數據進行傳輸。

在平臺層，數據庫服務器、NB-IoT服務器、LoRa服務器以及云服務器對網絡層傳輸來的遙測數據進行存儲與處理。具體而言，系統整體運行于云服務器之上，主體部分是運行于服務器中的開源IoT平臺——Thingsboard，該平臺基于Java語言構建并可提供完整的Web功能，其利用獨立的外置數據庫Cassandra將數據源與Web平臺隔離，以便為后續大數據處理預留空間。如前文所述，感知層的傳感器通信模組包括LoRa、NB-IoT、以及2G等多種網絡制式，因此本系統分別使用不同的方式對上述網絡制式傳送的遙測數據與設備數據進行處理。為實現數據采集，數據庫服務器中運行了三個程序模塊來實現數據源與云服務器中Cassandra數據庫的交互，包括數據獲取、解析、儲存和設備上報周期指令的下發。其中第一個程序模塊如圖2中數據庫服務器左側的雙向箭頭所示，對于2G網絡傳送的數據，系統通過TCP socket以及多線程機制對2G設備發送的數據進行實時接收并存儲至MySQL數據庫。對于使用NB-IoT與LoRa傳送的數據而言，數據庫服務器通過第二個程序模塊調用North API對NB-IoT服務器與LoRa服務器中緩存的數據進行讀取并存儲至MySQL數據庫。之后，數據庫服務器將MySQL中的數據通過基于JavaScript編寫的第3個程序模塊獲取并使用MQTT的方式發送至Thingsboard平臺的外置Cassandra數據庫。Thingsboard平臺支持HTTP、CoAP和MQTT等多種協議，其中MQTT是被廣泛應用的輕量級M2M/IoT消息協議，其采用訂閱/發布的模式。Thingsboard平臺的規則引擎會對數據庫服務器發送來的數據進行監測，當規則引擎發現異常數據時會向用戶推送報警短信和郵件。

在應用層，系統基于平臺層實現了在Thingsboard平臺內的規則引擎中利用SMTP和RESTful API分別向互聯網中的第三方電子郵件轉發商與短信轉發商推送信息，提供電子郵件和短信通知服務。此外，Thingsboard平臺內置一個Web servlet用于提供Web服務，其默認運行在本地8080端口，Web頁面由多個可編程部件(Widget)構成，Thingsboard平臺預先提供一些物聯網應用常用的顯示部件，如地圖、數據圖表等。可編程部件可在Web頁面內修改HTML、CSS和JavaScript部分代碼并動態編譯運行于Thingsboard平臺中。除了使用已有部件構建Web頁面，本系統還針對騰訊地圖的更新操作進行了修改，以消除鑒權失敗的錯誤并通過地圖渲染進一步增強顯示污染區域的具體范圍。此外，還新建了視頻直播部件，使得用戶可在傳感器終端站點頁面中獲得云平臺提供的攝像頭直播畫面。

2 物聯網水質監測系統

本節對物聯網水質監測系統平臺層的主要功能及關鍵技術進行介紹，以2G與NB-IoT為例(平臺層針對LoRa的操作與NB類似)展開。

2.1 2G TCP Socket模塊

該模塊部署于數據庫服務器中,其所包含的代碼文件關系圖如圖3所示，主函數所在的Java源文件在運行期間會調用其他5個代碼文件。其中Global Parameter.java主要用于定義2G TCP socket功能模塊相關全局變量。ServerThread.java文件包含與TCP socket相關聯的線程類的公有操作及私有變量的定義。MyDataSource.java描述了與存儲2G相關設備信息以及遙測數據信息的數據庫連接類的私有變量和公有操作的定義。V1Msg.java用于描述2G TCP Socket模塊用到的數據結構類的定義。下文將以主函數為主線來對該模塊的關鍵功能進行介紹。

圖3 2G TCP socket模塊的文件調用關系圖

2G TCP socket模塊main函數的主要邏輯功能如圖4所示，其包含六部分功能。第一部分功能用于設置線程計數器的初始值，主要用于對線程池中的線程進行計數。第二部分功能用于建立一個數據庫連接池對象，與數據庫連接池相關的類為MyDataSource，其相關定義在MyDataSource.java源文件中，該連接池包含了與2G設備信息及遙測數據信息相關的數據庫連接對象，如圖5中雙向箭頭所示。第三部分功能主要用于建立TCP Socket相關的TCP server線程池對象，全局變量GlobalParameter.maxThreadNumber規定了該線程池所包含的最大線程數，該值的定義位于全局變量文件GlobalParameter.java中。第四部分功能用于建立一個保護線程對象，保護線程類定義在ServerThread.java中，保護線程主要用于對線程池中的TCP server線程進行監視以防止某一個TCP server線程執行時間過長所造成的資源占用。第五部分功能用于建立一個TCP server socket對象并監聽指定的端口號。第六部分功能用于建立一個TCP server線程對象并為其賦予TCP socket連接、指定的TCP server線程索引號、以及指定的數據庫資源池ds。

圖4 2G TCP socket模塊main函數的主要邏輯功能

圖5 2G TCP Socket模塊多線程與多數據庫連接的對應關系

根據圖5可以看出TCP server線程資源池中的線程調用數據庫資源池中的任意一個數據庫連接與MySQL數據庫進行交互，具體而言，一個TCP server線程在任一時刻對應一個TCP socket連接與一個數據庫連接。對于右向箭頭，其表示TCP server線程通過TCP socket連接以及數據庫連接將2G網絡傳送來的設備數據與遙測數據寫入數據庫服務器中用于存儲2G設備信息和遙測信息的MySQL數據庫；對于左向箭頭，其表示數據庫服務器根據指定規則下發指令到2G設備來更新其上報數據的周期，目的是為節省設備的能耗。如前文所述，由于TCP server線程池中的線程數量有限，當上報數據的設備逐漸增多時，若沒有保護線程對當前的執行線程進行有效監視與釋放，那么線程池中的可用線程將會逐漸減少，最終可能導致無可用的線程來進行調度。因此，在本系統中，我們設定當線程的執行時間超過預設的最大執行時間timingvalue時就對該線程釋放，也即當線程資源池將線程提交時開始計時，當線程達到最大執行時間timing value時對該線程進行釋放，與保護線程相同，TCP server線程類的相關定義也在代碼文件ServerThread.java中。

2.2 數據庫服務器與NB-IoT服務器交互模塊

本模塊所包含的源代碼文件關系如圖6所示，主函數所在的DataCollectionFromServer.java文件在運行期間會調用其他3個代碼文件。其中DeviceInfoFrom Json.java文件主要包含了DeviceInfoFromJson類的定義，該類主要用于描述NB-IoT傳感器設備節點的主要屬性信息。DataFromJson.java文件包含了DataFrom Json類的定義，該類主要用于描述NB-IoT傳感器設備節點所采集的水質數據信息。ConnectionWithDB.java文件主要定義了與NB-IoT數據庫操作相關的函數，包括建立與數據庫之間的連接getConnection()、關閉與數據庫之間的連接closeConnection()、設備信息的添加與存儲add2Devicelist()、針對NB-IoT傳感器設備的數據上報周期指令查詢checkCommandList()、遙測數據信息的存儲insert2datatable()等。

圖6 與NB-IoT服務器交互模塊的文件調用關系圖

以圖7為主線來對該模塊的關鍵功能進行介紹。首先，mian函數的第一部分功能是對運行于數據庫服務器中的North API client進行認證與鑒權，執行成功后，North API client會獲得一個鑒權密鑰accessToken。之后，main函數執行第二與第三部分功能，North API client基于accessToken并通過調用庫函數queryDevices()在NB-IoT服務器端查詢所有NB-IoT的設備信息，通過getTotalCount()函數獲得設備的數量。對于新添加的設備，North API client將該設備信息加入數據庫服務器的DeviceInfo表中，對于已在表中存在的設備，North API client對該設備的相關屬性進行更新。如圖8所示，DeviceInfo首先被打包成JSON格式由NB-IoT服務器傳輸至數據庫服務器端，在數據庫服務器端North API client將JSON格式的數據進行解析并將解析后的設備信息插入/更新至設備信息表中。在設備信息寫入數據庫的同時，mian函數執行第四部分功能，即對于每一臺設備而言，North API client檢查ComandList表以確認是否有針對該設備下發的指令信息，該指令信息用于設備上報周期的更新，其目的是減少設備上報數據時所產生的能耗。

圖7 main函數的主要邏輯功能

圖8 數據庫服務器與NB-IoT服務器的交互

基于上述過程，main函數執行第五部分功能，North API client基于accessToken并通過調用庫函數queryDataHistory在NB-IoT服務器端查詢所有NB-IoT設備采集的遙測數據信息。與設備信息DeviceInfo相同，遙測數據信息DataInfo首先被打包成JSON格式由NB-IoT服務器傳輸至數據庫服務器端，之后North API client將JSON格式的數據進行解析并將解析后的信息根據采集的時間順序插入遙測數據信息表中。最后，main函數執行第六部分功能對鑒權密鑰進行刷新。

2.3 數據庫服務器與云服務器交互模塊

在2.2節已詳細說明了數據庫服務器如何對傳感器上報的水質監測數據及設備數據進行獲取與存儲。為了有效地對數據庫服務器中存儲的數據進行處理，平臺層的業務服務器(即云服務器)需對數據庫服務器中存儲的數據進行獲取。本節將介紹數據庫服務器與業務服務器之間交互模塊的關鍵技術并進一步對業務服務器的關鍵功能進行介紹。

如圖9所示，在數據庫服務器中我們采用Java script搭配MQTT的方式對其存儲的設備數據信息及設備遙測數據信息進行發布，其中數據庫服務器作為數據的發布者而云服務器作為數據的訂閱者。具體而言，圖9中的MQTT(消息隊列遙測傳輸)是ISO標準下基于發布/訂閱范式的消息協議。它工作于TCP/IP協議族之上，而Node是一個可讓JavaScript程序模塊運行于服務端的開發平臺。

圖9 數據庫服務器針對云服務器的數據發布

在云服務器中，Thingsboard平臺會根據定義的規則引擎(Rule Engine)對數據庫服務器發布的遙測數據信息進行過濾并對異常的遙測數據進行報警。其中規則引擎是一個易于使用的框架，主要用于構建基于事件的工作流，其由3個主要組成部分：

(1) 消息——任何傳入的事件，它可以是來自設備的傳入數據、設備的生命周期事件、REST API事件、RPC請求等。

(2) 規則節點——針對傳入消息所執行的功能，有多種不同類型的規則節點可對不同的傳入消息進行過濾、轉換或執行其他操作。

(3) 規則鏈——規則節點通過關系相互連接，因此來自規則鏈中的某一節點的輸入消息為其前驅節點的輸出消息，而該規則節點的輸出消息為其后繼規則節點的輸入消息。

由于水質監測系統監測的水質指標較多，在這里以其中5項指標為例進行相關說明，包括水的溶解氧濃度(DO)、氫離子濃度指數(PH)、水溫(TEMP)、氧化還原電位(ORP)、電導率(EC)。我們首先定義了如圖10所示的規則鏈根實體，其中規則鏈根實體中的五個長方形分別標識針對5項水質監測指標的規則鏈節點。

圖10 規則鏈根實體

以DO為例對其規則鏈的建立進行說明。如圖11所示，可以看到對于數據庫服務器發布的水質監測信息， DO規則鏈首先對DO值進行檢查并判斷該值是否觸發黃色警報或紅色警報，若該值異常并足以觸發黃色警報，則系統會向河長、部門、公司發送警報郵件及警報短信；若該值異常足以觸發紅色警報，則系統會向河長、部門、公司、領導發送警報郵件。其中觸發黃色警報的具體動作定義如圖12所示，紅色警報觸發動作的定義與黃色警報相同。

圖11 DO規則鏈的建立

圖12 觸發黃色警報動作的定義

2.4 云服務器中Thingsboard平臺地圖顯示

本系統還對云服務器中Thingsboard平臺的地圖顯示部件進行了修改，消除了鑒權失敗的錯誤并且通過污染范圍的繪制及渲染增強了水質污染區域的顯示。本節對該關鍵技術進行簡單介紹。

云服務器中的Thingsboard平臺可預先提供一些物聯網應用常用的顯示部件，如地圖、數據圖表等。這些顯示部件可通過可編程部件Widgets進行編輯與動態編譯。具體而言，其可通過在Web頁面內修改HTML、CSS、和JavaScript部分代碼并動態編譯運行于Thingsboard平臺中。

如2.3節所述，對于數據庫服務器發布來的最新遙測數據，Thingsboard平臺可對數據進行觸發報警短信與報警郵件的邏輯判斷。進一步，為了更清晰地在地圖中定位異常數據發生的位置，我們需對Thingsboard平臺前端及后臺的部分代碼進行更新。由于地圖顯示部件本身并不直接讀取Thingsboard平臺外置數據庫中的遙測數據，因此需要通過調用Thingsboard后臺代碼中的結構體dataSource進行遙測數據的讀取并進行正確的解析。具體而言，每個傳感器設備有其自身的屬性值，如經度、緯度、測量值和閾值，而傳感器數量眾多，發送的數據量較大，因此需要在后臺代碼中根據內置的key值對每個傳感器對應的屬性值進行區分，然后放入dataSource結體并發送至地圖顯示部件的API。

之后，前端代碼根據解析的數據判斷當前讀入的遙測數據是否為異常數據，對于異常數據，前端代碼根據對應數據采集設備的經緯度值進行定位并繪制出污染范圍。用戶顯示界面只需在Thingsboard平臺中的dashboard地圖數據源配置部分設置時間序列為最新的遙測值，進行刷新，若測量值大于閾值，雙擊地理坐標點則可根據污染范圍繪制如圖13所示的示意圖，再次雙擊則會更清楚地顯示繪制效果。

圖13 污染范圍繪制示意圖

本節加入的異常數據地圖渲染功能，搭配前文所述的報警短信與報警郵件功能，可對用戶提供更快速有效的水質監測決策支持。

3 實驗案例

本節通過具體的實驗案例來對物聯網水質監測系統平臺層功能進行測試。該實驗選取的監測區域位于上海市奉賢區的大治河，在河中部署了2G與NB-IoT兩種通信模組的傳感器節點與匯聚節點。兩種設備定時對水質監測數據進行上報，系統平臺層的數據庫服務器對上報的遙測數據進行接收并存儲，再將存儲的遙測數據發布至業務服務器做進一步處理，業務服務器對發布的數據進行處理并根據數據是否異常為應用層用戶提供預警服務(包括發送報警短信與報警郵件)，此外，應用層可提供多種顯示控件供用戶進行數據查詢與實時監測。

功能測試實驗步驟如下：

(1) 在數據庫服務器中的Node平臺中運行test.java模塊，該模塊基于javascript搭配MQTT協議編寫，其主要功能是將數據庫服務器中存儲的傳感器遙測數據發送至云服務器。

(2) 如圖14所示，在數據庫服務器中打開命令窗口，對發送至云服務器的遙測數據進行觀察。

圖14 云服務器接收的遙測數據中發現異常

(3) 如圖14高亮區域所示，由于發現異常數據，因此用戶將收到報警短信與報警郵件，如圖15與圖16所示。

圖15 異常數據觸發的報警短信

圖16 異常數據觸發的報警郵件

可以看出，由于云服務器收到的異常數據DO的值超出了規則引擎所規定的合理范圍，因此觸發了報警短信與報警郵件。報警短信與報警郵件中包含了采集該異常數據的傳感器設備地址。此外，在報警郵件中還包含了騰訊地圖的鏈接，通過鏈接可以打開如圖17所示的用戶查詢監測界面。

圖17 異常數據在數據庫服務器中的位置

(4) 圖17中，可以根據地圖找到相應位置的設備數據信息與遙測數據信息，此外由于通過可編程部件加入了視頻顯示插件，因此可以通過視頻實時監測此時河流周圍的具體環境信息。

(5) 根據報警短信與報警郵件顯示的異常數據信息，到數據庫服務器中定位該值所在的數據庫文件，如圖18所示，其位于數據庫表device5中，可以看出該設備編號與報警短信與報警郵件中的設備編號一致。

圖18 異常數據在用戶查詢與監測界面的顯示

通過上述實驗測試可以看出，平臺層可以成功地銜接網絡層與應用層，其可將網絡層傳輸來的遙測數據通過數據庫服務器進行有效存儲并能通過云服務器對存儲的遙測數據進行有效處理，為應用層的用戶提供決策服務。此外，應用層可通過地圖、圖表、視頻等可顯示部件來為用戶提供可視化的水質監測服務。

4 結 語

本文基于物聯網平臺的云-管-端體系設計并實現了一種基于海量異構傳感器的物聯網水質監測系統。該系統基于2G傳感器節點、NB-IoT傳感器節點以及LoRa傳感器節點對水質監測數據進行實時采集，并通過平臺層的數據庫服務器、NB-IoT服務器、LoRa服務器以及業務服務器之間的交互來實現水質遙測數據的高效存儲與處理。通過平臺層的處理，該系統可以圖表、地圖、視頻的形式為應用層的用戶提供水資源監測數據的實時查詢與監控并可通過報警短信與報警郵件的方式向用戶提供異常水質監測數據信息，以便迅速作出決策。總體而言，本文主要完成了以下工作：

(1) 提出了一種系統的總體設計架構，基于云-管-端體系對系統架構進行層次劃分并總結了系統各個層的主要組成單元及功能。

(2) 詳細介紹了系統平臺層的主要功能及關鍵技術，包括數據庫服務器通過TCP socket模塊對2G傳感器設備采集的遙測數據信息進行接收與存儲，數據庫服務器通過North API client模塊從NB-IoT服務器與LoRa服務器提取及存儲與NB-IoT傳感器和LoRa傳感器設備相關的遙測數據信息，以及數據庫服務器通過Node平臺以Javascript搭配MQTT的方式將其存儲的數據發布至云服務器供其進行數據的處理與提供面向用戶的Web服務。

(3) 通過實驗案例對水質監測系統進行了功能測試并對測試結果進行了分析，可以看出該系統可通過多種方式為用戶提供水質數據的查詢與監測并可及時發送水質預警短信及郵件來為用戶快速決策提供支持。

本文提出的水質監測系統是一項較為復雜的系統工程，文中的相關內容還需再做進一步完善與補充。今后的研究工作應包含以下相關內容：

(1) 考慮到未來海量傳感器節點的部署，海量數據的傳輸、存儲、處理，以及更為豐富的Web應用，應通過相關技術進一步增強系統的穩定性、可擴展性及開發便捷性。

(2) 考慮到未來傳感器節點采集的海量數據，應考慮采用服務器集群的方式對數據進行存儲與處理。

(3) 考慮到系統的商用穩定性及長期維護的便捷性，應采用備份服務器或服務器集群的方式實現服務器功能失效后的快速切換。