摘要：本文是其中的一個(gè)解決方案。地址客戶端服務(wù)端前端網(wǎng)頁介紹，消息隊(duì)列遙測(cè)傳輸是開發(fā)的一個(gè)即時(shí)通訊協(xié)議，有可能成為物聯(lián)網(wǎng)的重要組成部分。必須用于在頂層分隔符之后，除了當(dāng)自己指定時(shí)。

最近，本實(shí)驗(yàn)室大量上馬云測(cè)量，云監(jiān)控方面的項(xiàng)目，大概是屬于物聯(lián)網(wǎng)應(yīng)用的一個(gè)分支。老板也有將舊有儀器改造的想法，所以要實(shí)現(xiàn)儀器設(shè)備的云控制。本文是其中的一個(gè)解決方案。

消息隊(duì)列：MQTT，服務(wù)器使用centos，安裝mosquitto

客戶端使用C#，窗體框架使用WPF，MQTT的客戶端使用MQTTNet

服務(wù)端采用spring-cloud微服務(wù)框架

前端采用Vue，使用Element-admin-ui后臺(tái)框架，使用MQTTJS組件（MQTTJS采用websocket連接方式）

客戶端的測(cè)量采集數(shù)據(jù)程序使用TPL Dataflow

這個(gè)解決方案加入如圖(請(qǐng)?jiān)徫业膽卸?：

以上只是本demo的架構(gòu)，很簡單，但是也有很大的問題。

客戶端跟隨儀器，原則上一臺(tái)儀器一個(gè)控制程序，當(dāng)然也可以有多個(gè)?？蛻舳藢?shí)現(xiàn)了對(duì)儀器所有硬件設(shè)備的控制，對(duì)所有數(shù)據(jù)的采集?？蛻舳丝梢杂薪缑?，也可以沒有界面，一般來說，我們是需要一個(gè)界面的，客戶端可以獨(dú)立完成測(cè)量任務(wù)。

客戶端集成了網(wǎng)絡(luò)通信功能，可以完全替代用戶對(duì)客戶端的操作使用。具體架構(gòu)如下（請(qǐng)?jiān)徫业膽卸瑁?br>

其中，客戶端界面和網(wǎng)絡(luò)接口是等效的，可以多帶帶控制，也可以共同控制。

服務(wù)端基于spring-cloud微服務(wù)框架，主要提供服務(wù)發(fā)現(xiàn)，用戶管理，權(quán)限管理，設(shè)備管理，MQTT節(jié)點(diǎn)管理等管理功能

前端網(wǎng)頁是用戶通過網(wǎng)絡(luò)操作儀器設(shè)備的交互接口。采用日前流行的Vue框架，由于是后臺(tái)管理模式，就使用Element-admin-ui這個(gè)框架。

客戶端： https://github.com/spartajet/IotWpfClient
服務(wù)端：https://github.com/spartajet/iot-demo-server
前端網(wǎng)頁：https://github.com/spartajet/iot-demo-web

MQTT（Message Queuing Telemetry Transport，消息隊(duì)列遙測(cè)傳輸）是IBM開發(fā)的一個(gè)即時(shí)通訊協(xié)議，有可能成為物聯(lián)網(wǎng)的重要組成部分。該協(xié)議支持所有平臺(tái)，幾乎可以把所有聯(lián)網(wǎng)物品和外部連接起來，被用來當(dāng)做傳感器和制動(dòng)器（比如通過Twitter讓房屋聯(lián)網(wǎng)）的通信協(xié)議。

MQTT協(xié)議是為大量計(jì)算能力有限，且工作在低帶寬、不可靠的網(wǎng)絡(luò)的遠(yuǎn)程傳感器和控制設(shè)備通訊而設(shè)計(jì)的協(xié)議，它具有以下主要的幾項(xiàng)特性：
1、使用發(fā)布/訂閱消息模式，提供一對(duì)多的消息發(fā)布，解除應(yīng)用程序耦合；
2、對(duì)負(fù)載內(nèi)容屏蔽的消息傳輸；
3、使用 TCP/IP 提供網(wǎng)絡(luò)連接；
4、有三種消息發(fā)布服務(wù)質(zhì)量：
“至多一次”，消息發(fā)布完全依賴底層 TCP/IP 網(wǎng)絡(luò)。會(huì)發(fā)生消息丟失或重復(fù)。這一級(jí)別可用于如下情況，環(huán)境傳感器數(shù)據(jù)，丟失一次讀記錄無所謂，因?yàn)椴痪煤筮€會(huì)有第二次發(fā)送。
“至少一次”，確保消息到達(dá)，但消息重復(fù)可能會(huì)發(fā)生。
“只有一次”，確保消息到達(dá)一次。這一級(jí)別可用于如下情況，在計(jì)費(fèi)系統(tǒng)中，消息重復(fù)或丟失會(huì)導(dǎo)致不正確的結(jié)果。
5、小型傳輸，開銷很?。ü潭ㄩL度的頭部是 2 字節(jié)），協(xié)議交換最小化，以降低網(wǎng)絡(luò)流量；
6、使用 Last Will 和 Testament 特性通知有關(guān)各方客戶端異常中斷的機(jī)制；

詳細(xì)的MQTT協(xié)議內(nèi)容請(qǐng)參考：https://mcxiaoke.gitbooks.io/mqtt-cn/content/mqtt/01-Introduction.html

本文使用開源的MQTT 服務(wù)器mosquitto，介紹如下：

Eclipse Mosquitto is an open source (EPL/EDL licensed) message broker that implements the MQTT protocol versions 3.1 and 3.1.1. Mosquitto is lightweight and is suitable for use on all devices from low power single board computers to full servers.
The MQTT protocol provides a lightweight method of carrying out messaging using a publish/subscribe model. This makes it suitable for Internet of Things messaging such as with low power sensors or mobile devices such as phones, embedded computers or microcontrollers.

CentOS 7沒有mosquitto包。要安裝它，我們將首先安裝一個(gè)名為Extra Packages for Enterprise Linux或EPEL的額外軟件存儲(chǔ)庫。

sudo yum -y install epel-release

然后安裝

sudo yum -y install mosquitto

直接brew開路：

brew install mosquitto

配置文件mosquitto.conf路徑：

mac ：/usr/local/etc/mosquitto/mosquitto.conf

centos：/etc/mosquitto/mosquitto.conf

bind_address 127.0.0.1
port 1883
protocol mqtt

bind_address可以不配置，

listener 1884
protocol websockets

首先，要禁止匿名登錄

allow_anonymous false

配置密碼文件：

password_file usr/local/etc/mosquitto/pwfile

配置密碼的方法：

mosquitto提供了mosquitto_passwd工具設(shè)置密碼,使用方法如下：

?  ~ mosquitto_passwd help
mosquitto_passwd is a tool for managing password files for mosquitto.

Usage: mosquitto_passwd [-c | -D] passwordfile username
       mosquitto_passwd -b passwordfile username password
       mosquitto_passwd -U passwordfile
 -b : run in batch mode to allow passing passwords on the command line.
 -c : create a new password file. This will overwrite existing files.
 -D : delete the username rather than adding/updating its password.
 -U : update a plain text password file to use hashed passwords.

See http://mosquitto.org/ for more information.

我的做法是：先在usr/local/etc/mosquitto/pwfile中把用的密碼寫下來，如下：

user1:passwd1
user2:passwd2
user3:passwd3
user4:passwd4
user5:passwd5
user6:passwd6

然后使用mosquitto_passwd -U usr/local/etc/mosquitto/pwfile指令生成密碼

更多配置請(qǐng)參考 https://mosquitto.org

# Config file for mosquitto 
# 
# See mosquitto.conf(5) for more information. 
# 
# Default values are shown, uncomment to change. 
# 
# Use the 
# character to indicate a comment, but only if it is the 
# very first character on the line. 
# ================================================================= 
# General configuration 
# ================================================================= 
# 重新發(fā)送已經(jīng)發(fā)出去的Qos 為1或者2的消息的等待時(shí)間 
retry_interval 20 
# 系統(tǒng)狀態(tài)的刷新時(shí)間，設(shè)置為0表示不刷新 
sys_interval 10 
#清除在內(nèi)部消息存儲(chǔ)里面的未引用的消息的時(shí)間。 
#較低的值將占用較少的內(nèi)存，但處理器時(shí)間較長， 
#越高的值將產(chǎn)生相反的效果。 
#設(shè)置值為0意味著未引用的消息將以盡可能快的速度處理。 
store_clean_interval 10
#pid_file 
# 以什么用戶啟動(dòng) mosquitto，此配置在 windows 下無效，以非 root 運(yùn)行無效 
#user mosquitto 
#當(dāng)前每個(gè)客戶端正在傳輸?shù)腝o1和2消息的最大數(shù)量。 
#這包括通過握手信息，以及那些正在重試的信息。 
#默認(rèn)為20。設(shè)置為0表示無上限。 
#設(shè)置為1將保證QoS 1 和2的消息按順序傳遞 max_inflight_messages 20 
#當(dāng)前正在進(jìn)行的隊(duì)列中Qos 1和2條消息的最大數(shù)量。默認(rèn)為100。 
#設(shè)置到0表示沒有上限(不推薦)。同樣可參見queue_qos0_messages max_queued_messages 100 
#設(shè)置為true，當(dāng)一個(gè)持久客戶端被斷開連接時(shí)，以Qos為0將消息放到隊(duì)列中。 
#這些消息受max_queued_messages限制 queue_qos0_messages false 
#此選項(xiàng)設(shè)置被代理允許發(fā)布的消息的大小。 
#超過這個(gè)尺寸的消息將不會(huì)被代理接受。 
#默認(rèn)值為0，這意味著所有有效的MQTT消息都被接受。 
#MQTT的最大有效大小為268435455字節(jié) 
message_size_limit 0 
# 用于設(shè)置客戶端長連接的過期時(shí)間，默認(rèn)永不過期，必須以h d w m y為單位 
#分別代表 小時(shí)，天，星期，月，念 
#persistent_client_expiration 
# 如果客戶端訂閱了多個(gè)重疊的訂閱，例如foo/
#和foo/+/baz，然后MQTT期望當(dāng)代理接 
#收到一個(gè)與兩個(gè)訂閱相匹配的主題的消息時(shí)，例如foo/bar/baz，那么客戶端應(yīng)該只接 
#收一次消息。為了滿足這一要求，mosquitto不斷跟蹤發(fā)送給客戶的消息。允許重復(fù)的 
#消息選項(xiàng)允許禁用此行為，如果您有大量的客戶端訂閱相同的主題集合，并且非常關(guān)注 
#最小化內(nèi)存使用的情況，那么這個(gè)選項(xiàng)可能是有用的。如果你事先知道你的客戶端永不 
#會(huì)有重疊的訂閱，那么你的客戶必須能夠正確處理重復(fù)的信息，即使在Qo = 2的時(shí)候， 
#你的客戶端也必須能夠正確地處理重復(fù)的信息 
#allow_duplicate_messages false 
# ================================================================= 
# Default listener 
# ================================================================= 
# 服務(wù)綁定的IP地址 
#bind_address 
# 服務(wù)綁定的端口號(hào) 
#port 1883 
# 允許的最大連接數(shù)，-1表示沒有限制 
#max_connections -1 
# cafile：CA證書文件 
# capath：CA證書目錄 
# certfile：PEM證書文件 
# keyfile：PEM密鑰文件 
#cafile 
#capath 
#certfile 
#keyfile 
# 必須提供證書以保證數(shù)據(jù)安全性 
#require_certificate false 
# 若require_certificate值為true，use_identity_as_username也必須為true #use_identity_as_username false 
# 啟用PSK（Pre-shared-key）支持 
#psk_hint 
# SSL/TSL加密算法，可以使用“openssl ciphers”命令獲取 
# as the output of that command. 
#ciphers 
# ================================================================= 
# Persistence 
# ================================================================= 
# 消息自動(dòng)保存的間隔時(shí)間 
#autosave_interval 1800 
# 消息自動(dòng)保存功能的開關(guān) 
#autosave_on_changes false 
# 持久化功能的開關(guān) persistence true 
# 持久化DB文件
#persistence_file mosquitto.db 
# 持久化DB文件目錄 
#persistence_location /var/lib/mosquitto/ 
# ================================================================= 
# Logging 
# ================================================================= 
# 4種日志模式：stdout、stderr、syslog、topic 
# none 則表示不記日志，此配置可以提升些許性能 log_dest none 
# 選擇日志的級(jí)別（可設(shè)置多項(xiàng)） 
#log_type error 
#log_type warning 
#log_type notice 
#log_type information 
# 是否記錄客戶端連接信息 
#connection_messages true 
# 是否記錄日志時(shí)間 
#log_timestamp true 
# ================================================================= 
# Security 
# ================================================================= 
# 客戶端ID的前綴限制，可用于保證安全性 
#clientid_prefixes 
# 允許匿名用戶 
#allow_anonymous true 
# 用戶/密碼文件，默認(rèn)格式：username:password 
#password_file 
# PSK格式密碼文件，默認(rèn)格式：identity:key 
#psk_file 
# pattern write sensor/%u/data 
# ACL權(quán)限配置，常用語法如下： 
# 用戶限制：user  
# 話題限制：topic [read|write]  
# 正則限制：pattern write sensor/%u/data 
#acl_file 
# ================================================================= 
# Bridges 
# ================================================================= 
# 允許服務(wù)之間使用“橋接”模式（可用于分布式部署） 
#connection  
#address [:] 
#topic  [[[out | in | both] qos-level] local-prefix remote-prefix] 
# 設(shè)置橋接的客戶端ID 
#clientid 
# 橋接斷開時(shí)，是否清除遠(yuǎn)程服務(wù)器中的消息 
#cleansession false 
# 是否發(fā)布橋接的狀態(tài)信息 
#notifications true 
# 設(shè)置橋接模式下，消息將會(huì)發(fā)布到的話題地址 
# $SYS/broker/connection//state #notification_topic 
# 設(shè)置橋接的keepalive數(shù)值 
#keepalive_interval 60 
# 橋接模式，目前有三種：automatic、lazy、once 
#start_type automatic 
# 橋接模式automatic的超時(shí)時(shí)間 
#restart_timeout 30 
# 橋接模式lazy的超時(shí)時(shí)間 
#idle_timeout 60 
# 橋接客戶端的用戶名 
#username 
# 橋接客戶端的密碼 
#password 
# bridge_cafile：橋接客戶端的CA證書文件 
# bridge_capath：橋接客戶端的CA證書目錄 
# bridge_certfile：橋接客戶端的PEM證書文件 
# bridge_keyfile：橋接客戶端的PEM密鑰文件 
#bridge_cafile 
#bridge_capath 
#bridge_certfile 
#bridge_keyfile

參考文獻(xiàn)：https://www.imooc.com/article/19459

brew services start/stop mosquitto

systemctl start/stop/restart mosquitto

與消息隊(duì)列相比，主題非常輕量級(jí)。 客戶端不需要在發(fā)布或訂閱之前創(chuàng)建所需的主題，因?yàn)榇斫邮苊總€(gè)有效主題時(shí)不需要進(jìn)行任何預(yù)初始化。

主題使用/來分層次，以下是幾個(gè)主題的示例：

sensors/COMPUTER_NAME/temperature/HARDDRIVE_NAME

更重要的是，MQTT提供了通配符

+號(hào)僅僅匹配一個(gè)主題層次。例如，finance/stock/+匹配finance/stock/ibm與finance/stock/xyz，但是不匹配finance/stock/ibm/closingprice。因?yàn)閱螌哟瓮ㄅ浞麅H僅匹配一個(gè)層次，finance/+不匹配finance。
單層次通配符可用于主題樹內(nèi)任何層次，并與多層次通配符一起使用。必須用于在頂層分隔符之后，除了當(dāng)自己指定時(shí)。因此，+和finance/+ 都是有效的，但是finance+無效。單層通配符可用于主題樹最后或者在主題樹內(nèi)，例如，finance/+與finance/+/ibm都是有效的。

示例如下：

* a/b/c/d
* +/b/c/d
* a/+/c/d
* a/+/+/d
* +/+/+/+

#號(hào)可以匹配主題內(nèi)任何層次

單層次通配符可用于主題樹內(nèi)任何層次，并與多層次通配符一起使用。必須用于在頂層分隔符之后，除了當(dāng)自己指定時(shí)。因此，+和finance/+ 都是有效的，但是finance+無效。單層通配符可用于主題樹最后或者在主題樹內(nèi)，例如，finance/+與finance/+/ibm都是有效的。

示例如下：

* a/b/c/d
* #
* a/#
* a/b/#
* a/b/c/#
* +/b/c/#

目前，我手上還沒有一個(gè)趁手的采樣設(shè)備，所以只能模擬生成數(shù)據(jù)，但是，預(yù)計(jì)下周，我就可以解決這個(gè)問題了

關(guān)于如何生成數(shù)據(jù)以及使用TPL Dataflow數(shù)據(jù)采集和處理，請(qǐng)參考我的另一篇博客：

像Labview一樣，使用C#構(gòu)建測(cè)量數(shù)據(jù)流式處理框架

這里重點(diǎn)介紹MQTTnet的使用

MQTTnet是一個(gè)高性能的MQTT基礎(chǔ)連接.NET庫。提供了MQTT服務(wù)端和客戶端支持。

MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker). The implementation is based on the documentation from http://mqtt.org/.

特性如下：

Async support

TLS 1.2 support for client and server (but not UWP servers)

Extensible communication channels (i.e. In-Memory, TCP, TCP+TLS, WS)

Lightweight (only the low level implementation of MQTT, no overhead)

Performance optimized (processing ~70.000 messages / second)*

Interfaces included for mocking and testing

Access to internal trace messages

Unit tested (~90 tests)

目前支持的版本如下：

.NET Standard 1.3+

.NET Core 1.1+

.NET Core App 1.1+

.NET Framework 4.5.2+ (x86, x64, AnyCPU)

Mono 5.2+

Universal Windows Platform (UWP) 10.0.10240+ (x86, x64, ARM, AnyCPU, Windows 10 IoT Core)

Xamarin.Android 7.5+

Xamarin.iOS 10.14+

重點(diǎn)是：支持異步

連接MQTT

/// 

/// 初始化初始化MQTT
/// 
private void InitialMqtt()
{
   this._mqttClient = new MqttFactory().CreateMqttClient();
   this._mqttClient.ConnectAsync(new MqttClientOptionsBuilder()
       .WithClientId(Guid.NewGuid().ToString("N"))
       .WithTcpServer("*****",1883)
       .WithCredentials("admin", "admin")
       .WithCleanSession()
       .Build());
}

可以看到，我們使用的ConnectAsync()方法，是異步連接。

發(fā)布消息

首先要構(gòu)建消息

var message = new MqttApplicationMessageBuilder()
                    .WithTopic("measure/force")
                    .WithPayload(t.ToString(CultureInfo.InvariantCulture))
                    .WithExactlyOnceQoS()
                    .WithRetainFlag()
                    .Build();

然后異步發(fā)送

this._mqttClient.PublishAsync(message);

詳細(xì)代碼請(qǐng)參考

客戶端： https://github.com/spartajet/IotWpfClient

服務(wù)端主要是提供用戶管理，參考源代碼即可，涉及到CORS跨域問題，請(qǐng)參考我的另一篇博客：

Spring boot 和Vue開發(fā)中CORS跨域問題

mqttjs是支持MQTT協(xié)議的客戶端javascript庫，注意只是客戶端，并且通信方式是websockt，所以要在mosquitto服務(wù)器開啟websocket支持。

MQTT.js is a client library for the MQTT protocol, written in JavaScript for node.js and the browser.

npm install mqtt

mqtt.connect()
mqtt.Client()
mqtt.Client#publish()
mqtt.Client#subscribe()
mqtt.Client#unsubscribe()
mqtt.Client#end()
mqtt.Client#removeOutgoingMessage()
mqtt.Client#reconnect()
mqtt.Client#handleMessage()
mqtt.Client#connected
mqtt.Client#reconnecting
mqtt.Client#getLastMessageId()
mqtt.Store()
mqtt.Store#put()
mqtt.Store#del()
mqtt.Store#createStream()
mqtt.Store#close()

連接服務(wù)器：

const client = mqtt.connect("ws://ip:1884", {
          clientid: "fdafdafas",
          username: "admin",
          password: "admin"
        })

設(shè)置連接后的事件，要訂閱相關(guān)主題的消息

client.on("connect", function() {
          client.subscribe("measure/force", function(err) {
            if (!err) {
               client.publish("measure/force", "Hello mqtt")
            }
          })
        })

消息推送通知事件

client.on("message", function (topic, message) {
  // message is Buffer
  console.log(message.toString())
  client.end()
})

其他內(nèi)容請(qǐng)參考項(xiàng)目源碼：

前端網(wǎng)頁：https://github.com/spartajet/iot-demo-web

目前的demo只是完成了mqtt的使用基礎(chǔ)范例，沒有其他功能

對(duì)于設(shè)備管理，用戶權(quán)限等功能，打算用hsweb大神的物聯(lián)網(wǎng)框架hsweb-iot-cloud，也不排除自己開發(fā)的可能，看我的時(shí)間和項(xiàng)目需求

mqtt的壓力測(cè)試還沒有測(cè)試，但是從目前的情況來看（我的MQTT服務(wù)器用的華為云），我客戶端每秒生成100個(gè)數(shù)據(jù)，網(wǎng)頁端顯示基本沒什么延時(shí)，但并不代表實(shí)時(shí)性很好