前言
工業物聯網是一個范圍很大的概念，本文從數據可視化的角度介紹了一個最小化的工業物聯網平台，從 Modbus 數據采集到前端數據可視化呈現的基本實現思路。這里面主要涉及基於 Modbus 通訊規約的數據采集、后台實時數據處理、前端實時數據接收、前端實時數據可視化顯示。物聯網平台架構主要參考了圖撲物聯工業物聯網平台，並從中提取了部分功能進行介紹，前端數據可視化采用的是HT for Web。
由於內容比較多，具體實現上涉及到前端工程師、后台工程師、數據采集工程師等多個開發角色的參與，所以本文重點介紹實現思路和 WebSocket 消息推送的實現，其它環節的具體實現細節作者會在其它文章中進行詳細介紹。
一、物聯網平台架構
物聯網平台主要是B/S模式，工業物聯網平台大都采用的是微服務架構，本文主要涉及兩個微服務：前置數據采集服務和 Web 實時消息推送服務。
前置數據采集服務主要用於現場設備、儀器、儀表、傳感器實時數據的采集，IoTopo工業物聯網平台支持MQTT和透傳雲解析兩種方式，透傳雲解析支持 Modbus 通訊規約。
實時數據采集到平台后，需要推送到瀏覽器端進行顯示，Web 實時消息推送服務采用 Web Socket 進行實時數據推送，可以確保數據的實時性和高效性。
前端可視化技術采用的是HT for Web, HT for Web 是基於HTML5標准的企業應用圖形界面一站式解決方案，其包含通用組件、拓撲組件和3D渲染引擎等豐富的圖形界面開發類庫。雖然 HT for Web 是商業軟件但其提供的一站式解決方案可以極大縮短產品開發周期、減少研發成本、補齊我們在 Web 圖形界面可視化技術上的短板。
二、Modbus 數據采集
Modbus是一種串行通信協議，是Modicon公司（現在的施耐德電氣Schneider Electric）於1979年為使用可編程邏輯控制器（PLC）通信而發表。Modbus已經成為工業領域通信協議的業界標准，並且現在是工業電子設備之間常用的連接方式。Modbus比其他通信協議使用的更廣泛的主要原因有：
- 公開發表並且無版權要求
- 易於部署和維護
- 對供應商來說，修改移動本地的比特或字節沒有很多限制
Modbus允許多個 (大約240個) 設備連接在同一個網絡上進行通信，舉個例子，一個由測量溫度和濕度的裝置，並且將結果發送給計算機。在數據采集與監視控制系統（SCADA）中，Modbus通常用來連接監控計算機和遠程終端控制系統（RTU）。
目前主流的編輯語言都有 Modbus 開發庫，由於 Modbus 相對比較簡單，很多企業也選擇自行開發實現。Modbus 數據采集屬於后台通訊，數據采集到平台后首先會進行數據清理和預處理，過濾掉冗余和無效數據，形成實時數據。平台獲取到實時數據后一般會做 3 項工作：
1. 推送到 Web 前端進行顯示
2. 存儲到時序數據庫
3. 判斷是否產生告警
三、將實時數據推送到 Web 前端
基於 Web 的實時數據推送需要用到 WebSocket，初學者可以學習阮一峰老師的 WebSocket 教程。我們基於 WebSocket 封裝了一套消息傳輸協議，類似於一個消息中間件，前端部分可以訂閱實時數據。考慮到海量實時數據的推送需求，將實時數據分為平台級、站點級、設備級，前端在訂閱實時數據時，可以通過消息主題規則訂閱不同級別的數據。平台側在收到訂閱請求時，可以主動推送一次實時數據。這樣可以確保數據可視化界面在訂閱實時數據成功后，第一時間顯示出正確的界面。
下面給出一個簡化的 WebSocket 消息協議的客戶端代碼，大家可以在些基礎上進行改造以適合自己的業務場景。
消息主題正則表達式，用來匹配消息主題：
1 const matchWildcard = function(str, rule) { 2 return new RegExp('^' + rule.split('*').join('.*') + '$').test(str) 3 }
WebSocket 客戶端，支持消息主題訂閱、取消消息主題訂閱、同一個消息主題支持多個訂閱者：
1 class WebSocketClient { 2 constructor() { 3 this.ws = null 4 this.opts = { 5 debug: false, 6 autoReconnect: true, 7 reconnectInterval: 10000, 8 subscriber: {}, 9 } 10 this.opened = false 11 } 12 13 connect() { 14 if (!this.opened) { 15 return 16 } 17 18 const url = 'ws://www.iotopo.com/msg/v1' 19 console.debug('websocket connect', url) 20 21 let ws = this.ws = new WebSocket(url) 22 ws.onmessage = event => { 23 if (this.opts.debug) { 24 console.log(event) 25 } 26 let data = JSON.parse(event.data) 27 28 for (let topic in this.opts.subscriber) { 29 if (matchWildcard(data.topic, topic)) { 30 let listeners = this.opts.subscriber[topic] 31 if (Array.isArray(listeners)) { 32 listeners.forEach(cb => { 33 if (typeof cb === 'function') { 34 cb(data.payload) 35 } 36 }) 37 } 38 } 39 } 40 } 41 ws.onopen = e => { 42 if (this.opts.debug) { 43 console.log(e) 44 } 45 // 執行訂閱請求 46 for (let topic in this.opts.subscriber) { 47 this._sendSubscribe(topic) 48 } 49 if (typeof this.opts.onopen === 'function') { 50 this.opts.onopen(e) 51 } 52 } 53 ws.onclose = e => { 54 if (this.opts.debug) { 55 console.log(e) 56 } 57 if (typeof this.opts.onclose === 'function') { 58 this.opts.onclose(e) 59 } 60 if (this.opened && this.opts.autoReconnect) { 61 setTimeout(() => { 62 this.connect() 63 }, this.opts.reconnectInterval) 64 } 65 } 66 ws.onerror = e => { 67 if (this.opts.debug) { 68 console.log(e) 69 } 70 if (typeof this.opts.onerror === 'function') { 71 this.opts.onerror(e) 72 } 73 } 74 } 75 76 open(opts) { 77 if (!this.opened) { 78 Object.assign(this.opts, opts || {}) 79 this.opened = true 80 this.connect() 81 } 82 } 83 84 close() { 85 this.opened = false 86 if (this.ws !== null) { 87 this.ws.close() 88 } 89 this.ws = null 90 } 91 92 isOpened() { 93 return this.opened 94 } 95 96 isConnected() { 97 return this.ws !== null 98 } 99 100 _sendSubscribe(topic) { 101 if (this.ws === null) { 102 return Error('websocet not opened') 103 } 104 if (typeof topic !== 'string') { 105 return Error('topic should be a string value') 106 } 107 108 if (this.ws.readyState === WebSocket.OPEN) { 109 let msg = { 110 type: 'subscribe', 111 topic: topic, 112 } 113 this.ws.send(JSON.stringify(msg)) 114 } else { 115 return Error('websocet not connected') 116 } 117 } 118 119 subscribe(topic, cb) { 120 if (this.opts.debug) { 121 console.log('subscribe:', topic) 122 } 123 let listeners = this.opts.subscriber[topic] 124 if (!Array.isArray(listeners)) { 125 listeners = [ 126 cb 127 ] 128 this.opts.subscriber[topic] = listeners 129 } else { 130 listeners.push(cb) 131 } 132 this._sendSubscribe(topic) 133 134 return { topic, cb } 135 } 136 137 unsubscribe({topic, cb}) { 138 if (this.opts.debug) { 139 console.log('unsubscribe:', topic) 140 } 141 142 if (this.ws === null) { 143 return Error('websocet not opened') 144 } 145 146 if (typeof topic !== 'string') { 147 return Error('topic should be a string value') 148 } 149 150 let listeners = this.opts.subscriber[topic] 151 if (cb) { 152 if (Array.isArray(listeners)) { 153 let idx = listeners.indexOf(cb) 154 if (idx >= 0) { 155 listeners.splice(idx, 1) 156 } 157 } 158 } else { 159 delete this.opts.subscriber[topic] 160 } 161 162 if (Array.isArray(listeners) && listeners == 0) { 163 if (this.ws.readyState === WebSocket.OPEN) { 164 let msg = { 165 type: 'unsubscribe', 166 topic: topic, 167 } 168 this.ws.send(JSON.stringify(msg)) 169 } else { 170 return Error('websocet not connected') 171 } 172 } 173 } 174 }
用法舉例：
1 // 初始化客戶端 2 const ws = new WebSocketClient() 3 // 與 WebSocket 服務器建議連接 4 ws.open({ 5 debug: false 6 }) 7 // 訂閱消息 8 ws.subscribe('/foo/bar/*', function(msg) { 9 console.log('recv ws msg:', msg) 10 })
四、數據可視化界面實現
基於 HT for Web 可以簡單快速地搭建一個符合 HTML5 標准的可視化圖形界面，通過 WebSocket 訂閱實時數據，然后驅動圖形界面的變化。數據驅動圖形界面變化的實現方式很多，基本方法是采用數據綁定的方式，具體可以參考 HT for Web 的官方文檔。
在后面的文章中，作者會介紹一種基於 HT for Web 實現的業務數據和圖形數據分離的數據綁定方法。
