基于Netty+ActiveMQ 的農(nóng)村生活污水處理設(shè)施監(jiān)測數(shù)據(jù)通信管理平臺設(shè)計(jì)

劉祥宏，潘泉涌，方 寬

（1.浙江省建筑科學(xué)設(shè)計(jì)研究院有限公司，杭州 310000；2. 衢州市住建局，衢州 324000）

1 系統(tǒng)架構(gòu)圖

本系統(tǒng)建立在J2EE 平臺上，將 Netty，ActiveMQ 消息中間件，MySQL 數(shù)據(jù)庫、redis、json 等技術(shù)相結(jié)合，構(gòu)建更加智能、更加穩(wěn)定和并發(fā)更好的的通信管理平臺。通信管理服務(wù)系統(tǒng)搭建在云平臺之上，充分利用現(xiàn)代化信息技術(shù)手段實(shí)現(xiàn)農(nóng)村生活污水處理設(shè)施終端數(shù)據(jù)采集的信息化、集約化，依托云平臺的理念和優(yōu)勢，將已有的專業(yè)系統(tǒng)納入其中，為主管部門、運(yùn)維企業(yè)、其他相關(guān)部門提供統(tǒng)一的數(shù)據(jù)服務(wù)。系統(tǒng)架構(gòu)圖如圖1所示。

圖1 系統(tǒng)架構(gòu)圖

本系統(tǒng)最大的亮點(diǎn)在于使用Netty 作為通信框架，支持海量并發(fā)的同時(shí)，通過其預(yù)制的編碼和解碼器，實(shí)現(xiàn)對不同通信協(xié)議的解析后，將數(shù)據(jù)統(tǒng)一成通用json 格式推送到消息中間件。這樣對于監(jiān)管、運(yùn)維等應(yīng)用平臺而言，數(shù)據(jù)格式統(tǒng)一、規(guī)范，便于使用。

本文通過對比分析國內(nèi)現(xiàn)有的數(shù)據(jù)采集和通信管理解決方案，采用Netty+ActiveMQ 相結(jié)合的方式，降低技術(shù)難度的同時(shí)，實(shí)現(xiàn)農(nóng)村生活污水治理設(shè)施終端數(shù)據(jù)采集、傳輸管理和及時(shí)推送。

2 數(shù)采和控制實(shí)現(xiàn)

平臺可實(shí)現(xiàn)系統(tǒng)用戶的需求，如查看農(nóng)村生活污水處理設(shè)施的流量計(jì)、多功能電表、在線水質(zhì)檢測儀等實(shí)時(shí)數(shù)據(jù)，并根據(jù)設(shè)計(jì)好的指令對上述設(shè)備進(jìn)行控制操作。Netty 與消息中間件的結(jié)合完美解決了通信鏈路和消息推送問題，為實(shí)現(xiàn)平臺的通信管理和數(shù)據(jù)的實(shí)時(shí)推送奠定了基礎(chǔ)。詳細(xì)設(shè)計(jì)圖如圖2所示：

2.1 數(shù)采的設(shè)計(jì)與實(shí)現(xiàn)

平臺的服務(wù)器分為Netty 通信服務(wù)與消息中間件兩個(gè)部分。

（1）Netty 通信服務(wù)包括平臺通用功能和通用接口，用來實(shí)現(xiàn)與各下位機(jī)（數(shù)采儀）的數(shù)據(jù)傳送并將數(shù)據(jù)按照統(tǒng)一json 格式送入消息中間件，Netty 通信服務(wù)在初始化時(shí)與消息中間件（ActiceMQ）建立通信連接；

（2）消息中間件（ActiveMQ）實(shí)現(xiàn)消息的訂閱和推送，主要負(fù)責(zé)將數(shù)據(jù)推送給監(jiān)管平臺、企業(yè)運(yùn)維平臺、其他授權(quán)接入的第三方平臺，實(shí)現(xiàn)了數(shù)據(jù)的實(shí)時(shí)交互。

圖2 通信管理詳細(xì)設(shè)計(jì)圖

通信管理平臺將農(nóng)村生活污水處理設(shè)施終端數(shù)據(jù)推送到應(yīng)用平臺的流程描述：

下位機(jī)（PLC、單片機(jī)、智能網(wǎng)關(guān)等）采集各傳感器的實(shí)時(shí)數(shù)據(jù)，將這些數(shù)據(jù)上傳到Netty 服務(wù)器，當(dāng)與Netty 服務(wù)器第一次建立連接時(shí)，觸發(fā)channelActive 方法建立通道，該通道在斷開之前一直存在，此后下位機(jī)定時(shí)發(fā)送數(shù)據(jù)，并直接觸發(fā)channelRead 方法接收，接收到的數(shù)據(jù)由平臺統(tǒng)一處理，按照事先設(shè)計(jì)好的數(shù)據(jù)格式組成通用json 數(shù)據(jù)包，再按照設(shè)計(jì)的主題推送至消息中間件（ActiveMQ），監(jiān)管平臺、運(yùn)維平臺訂閱相應(yīng)主題即可獲取推送消息（數(shù)據(jù)）。

Netty 服務(wù)器采用多線程服務(wù)器，對于每一個(gè)連接請求，dispatcher 都會為其創(chuàng)建并分配一個(gè)線程，該線程負(fù)責(zé)這個(gè)請求的處理，優(yōu)點(diǎn)是執(zhí)行粒度是完整的處理流程，處理邏輯清晰，易于開發(fā)。通信過程中，可通過心跳包實(shí)現(xiàn)長連接，通過線程池控制服務(wù)端線程數(shù)的快速增長。

除了消息中間件（ActiveMQ）以外，平臺實(shí)現(xiàn)了一組restful風(fēng)格的通用數(shù)據(jù)接口。監(jiān)管平臺、企業(yè)運(yùn)維平臺等可以在獲取授權(quán)后，通過這些接口獲取歷史數(shù)據(jù)、終端信息，下達(dá)控制指令等。

平臺實(shí)現(xiàn)了對TCP 字節(jié)流（數(shù)據(jù)幀）、MQTT 的數(shù)據(jù)解析。

2.1.1 TCP 字節(jié)流（數(shù)據(jù)幀）

下位機(jī)（數(shù)采儀）可以按照約定的數(shù)據(jù)包格式，將終端監(jiān)測數(shù)據(jù)打包，以TCP 字節(jié)流（數(shù)據(jù)幀，字符集為utf-8）的方式上傳。數(shù)據(jù)包格式可做如下設(shè)計(jì)：服務(wù)端收到數(shù)據(jù)包以后，按照通信協(xié)議解包以后，再將數(shù)據(jù)組合成通用格式的json 數(shù)據(jù)包，送到消息中間件。

Netty 對于TCP 字節(jié)流（數(shù)據(jù)幀）有多種解碼方式，可以采用定長數(shù)據(jù)或者固定結(jié)尾字符（比如以回車換行作為結(jié)尾符）等方式，可以有效解決半包、粘包等問題。

2.1.2 MQTT 消息

根據(jù)農(nóng)村生活污水處理設(shè)施點(diǎn)多面廣、網(wǎng)絡(luò)條件差、監(jiān)測數(shù)據(jù)數(shù)據(jù)量少、實(shí)時(shí)性和安全性低的特點(diǎn)，特別適合采用MQTT技術(shù)。本次設(shè)計(jì)基于MQTT 3.1.1版本，在Netty 服務(wù)端實(shí)現(xiàn)了QoS=1 和QoS=0的消息訂閱/發(fā)布，并根據(jù)實(shí)際需要，設(shè)計(jì)了up、data、ctrl、alarm 四個(gè)主題，分別用來訂閱和發(fā)布終端設(shè)備信息，監(jiān)測數(shù)據(jù)，控制指令和報(bào)警信息。

圖3

MQTT 消息體的數(shù)據(jù)包采用json 數(shù)據(jù)格式，將終端編碼、數(shù)據(jù)等按照協(xié)議打包以后按照約定主題發(fā)布即可。Netty 服務(wù)器在下位機(jī)創(chuàng)建通道時(shí)，可按照終端編碼事先創(chuàng)建四個(gè)主題。如，終端編碼為2018122500001，則創(chuàng)建202018122500001/up、2018122500001/data、2018122500001/ctrl、2018122500001/alarm四個(gè)主題。通信管理平臺提供了設(shè)備注冊功能，通過注冊且處于啟用狀態(tài)的終端，可自動根據(jù)終端編碼創(chuàng)建上述四個(gè)主題。

2.1.3 JSON 數(shù)據(jù)格式

JSON 是一種輕量級數(shù)據(jù)交換格式，它采用完全獨(dú)立于語言的文本格式，此特性使JSON 成為理想的數(shù)據(jù)交換語言，易于閱讀和編寫，同時(shí)也易于機(jī)器解析和生成，提升網(wǎng)絡(luò)傳輸速率。本平臺的各數(shù)據(jù)通信環(huán)節(jié)均采用JSON 格式通信，使用對象和數(shù)組兩種結(jié)構(gòu)。對象在JSON 格式表示為“{ }”中的內(nèi)容，數(shù)組在JSON 格式是中括號“[ ]”中的內(nèi)容，通過這兩種結(jié)構(gòu)可以表示各種復(fù)雜的結(jié)構(gòu)。

2.2 控制實(shí)現(xiàn)

當(dāng)用戶發(fā)送控制信息時(shí)，數(shù)據(jù)又是如何從上位機(jī)（服務(wù)端）到達(dá)控制設(shè)備的呢？這個(gè)流程和2.1中介紹的數(shù)采流程相反。值得一提的是，采用MQTT 消息訂閱和發(fā)布方式時(shí)，如果沒有根據(jù)終端（下位機(jī)）的編碼進(jìn)行主題創(chuàng)建，那么指令下發(fā)時(shí)，所有訂閱了ctrl 主題的終端都會收到指令消息，然后終端需要根據(jù)消息里面的終端編碼或者具體指令來判斷該指令是否需要執(zhí)行。

3 平臺功能設(shè)計(jì)

通信管理平臺功能效果圖如下：

圖4

3.1 接入安全控制

下位機(jī)在接入通信管理平臺時(shí)，可以通過驗(yàn)證用戶名、密碼以及注冊設(shè)備編碼等方式進(jìn)行接入端身份驗(yàn)證。平常通信過程中，可以通過到包數(shù)量和數(shù)采頻率對通信進(jìn)行動態(tài)監(jiān)控，如果到包數(shù)量異常，可以人為關(guān)閉通道和加入通信黑名單等手段進(jìn)行通信管理。

3.2 接入設(shè)備管理

通信管理平臺通過設(shè)備白名單對接入設(shè)備進(jìn)行驗(yàn)證。農(nóng)村生活污水處理設(shè)施終端（下位機(jī)）在平臺注冊且啟用時(shí)，才有權(quán)限接入本平臺。一旦判定接入設(shè)備存在風(fēng)險(xiǎn)，管理員可以手動將風(fēng)險(xiǎn)設(shè)備禁用或者踢出白名單，Netty 服務(wù)將關(guān)閉該設(shè)備的通信通道，切斷連接，中止接入。該設(shè)備再次接入時(shí)，會因?yàn)椴辉诎酌麊味痪芙^。

3.3 監(jiān)測設(shè)備及數(shù)據(jù)管理

以農(nóng)村生活污水治理設(shè)施終端為最小單位，管理終端的監(jiān)測設(shè)備（傳感器）。通過列表結(jié)合圖表的方式，可以動態(tài)展示各終端監(jiān)測數(shù)據(jù)在時(shí)間維度的變化情況，結(jié)合報(bào)警數(shù)據(jù)。

3.4 報(bào)警數(shù)據(jù)

下位機(jī)發(fā)生異常時(shí)，向平臺傳輸報(bào)警數(shù)據(jù)。平臺本身根據(jù)調(diào)度任務(wù)分時(shí)段計(jì)算各下位機(jī)的數(shù)據(jù)到達(dá)率，如果掉包率過高或者到包數(shù)遠(yuǎn)超預(yù)計(jì)，可以對用戶進(jìn)行警告提示。

4 結(jié)束語

本平臺使用Netty+消息中間件技術(shù)解決了下位機(jī)與服務(wù)器雙向通訊的問題，后臺可以隨時(shí)向監(jiān)管平臺、企業(yè)運(yùn)維平臺實(shí)時(shí)推送消息（數(shù)據(jù)），以保證終端數(shù)據(jù)的真實(shí)、實(shí)時(shí)。下一步工作將在通信協(xié)議（數(shù)據(jù)格式）上進(jìn)行進(jìn)一步細(xì)化、分類和規(guī)范，將數(shù)據(jù)采集、管理、控制、實(shí)時(shí)推送做到最優(yōu)。