摘要：本文是其中的一個(gè)解決方案。地址客戶(hù)端服務(wù)端前端網(wǎng)頁(yè)介紹，消息隊(duì)列遙測(cè)傳輸是開(kāi)發(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

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

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

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

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

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

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

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

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

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

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

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

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

MQTT（Message Queuing Telemetry Transport，消息隊(duì)列遙測(cè)傳輸）是IBM開(kāi)發(fā)的一個(gè)即時(shí)通訊協(xié)議，有可能成為物聯(lián)網(wǎng)的重要組成部分。該協(xié)議支持所有平臺(tái)，幾乎可以把所有聯(lián)網(wǎng)物品和外部連接起來(lái)，被用來(lái)當(dāng)做傳感器和制動(dòng)器（比如通過(guò)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ā)布完全依賴(lài)底層 TCP/IP 網(wǎng)絡(luò)。會(huì)發(fā)生消息丟失或重復(fù)。這一級(jí)別可用于如下情況，環(huán)境傳感器數(shù)據(jù)，丟失一次讀記錄無(wú)所謂，因?yàn)椴痪煤筮€會(huì)有第二次發(fā)送。
“至少一次”，確保消息到達(dá)，但消息重復(fù)可能會(huì)發(fā)生。
“只有一次”，確保消息到達(dá)一次。這一級(jí)別可用于如下情況，在計(jì)費(fèi)系統(tǒng)中，消息重復(fù)或丟失會(huì)導(dǎo)致不正確的結(jié)果。
5、小型傳輸，開(kāi)銷(xiāo)很?。ü潭ㄩL(zhǎng)度的頭部是 2 字節(jié)），協(xié)議交換最小化，以降低網(wǎng)絡(luò)流量；
6、使用 Last Will 和 Testament 特性通知有關(guān)各方客戶(hù)端異常中斷的機(jī)制；

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

本文使用開(kāi)源的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沒(méi)有mosquitto包。要安裝它，我們將首先安裝一個(gè)名為Extra Packages for Enterprise Linux或EPEL的額外軟件存儲(chǔ)庫(kù)。

sudo yum -y install epel-release

然后安裝

sudo yum -y install mosquitto

直接brew開(kāi)路：

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中把用的密碼寫(xiě)下來(lái)，如下：

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í)間較長(zhǎng)， 
#越高的值將產(chǎn)生相反的效果。 
#設(shè)置值為0意味著未引用的消息將以盡可能快的速度處理。 
store_clean_interval 10
#pid_file 
# 以什么用戶(hù)啟動(dòng) mosquitto，此配置在 windows 下無(wú)效，以非 root 運(yùn)行無(wú)效 
#user mosquitto 
#當(dāng)前每個(gè)客戶(hù)端正在傳輸?shù)腝o1和2消息的最大數(shù)量。 
#這包括通過(guò)握手信息，以及那些正在重試的信息。 
#默認(rèn)為20。設(shè)置為0表示無(wú)上限。 
#設(shè)置為1將保證QoS 1 和2的消息按順序傳遞 max_inflight_messages 20 
#當(dāng)前正在進(jìn)行的隊(duì)列中Qos 1和2條消息的最大數(shù)量。默認(rèn)為100。 
#設(shè)置到0表示沒(méi)有上限(不推薦)。同樣可參見(jiàn)queue_qos0_messages max_queued_messages 100 
#設(shè)置為true，當(dāng)一個(gè)持久客戶(hù)端被斷開(kāi)連接時(shí)，以Qos為0將消息放到隊(duì)列中。 
#這些消息受max_queued_messages限制 queue_qos0_messages false 
#此選項(xiàng)設(shè)置被代理允許發(fā)布的消息的大小。 
#超過(guò)這個(gè)尺寸的消息將不會(huì)被代理接受。 
#默認(rèn)值為0，這意味著所有有效的MQTT消息都被接受。 
#MQTT的最大有效大小為268435455字節(jié) 
message_size_limit 0 
# 用于設(shè)置客戶(hù)端長(zhǎng)連接的過(guò)期時(shí)間，默認(rèn)永不過(guò)期，必須以h d w m y為單位 
#分別代表 小時(shí)，天，星期，月，念 
#persistent_client_expiration 
# 如果客戶(hù)端訂閱了多個(gè)重疊的訂閱，例如foo/
#和foo/+/baz，然后MQTT期望當(dāng)代理接 
#收到一個(gè)與兩個(gè)訂閱相匹配的主題的消息時(shí)，例如foo/bar/baz，那么客戶(hù)端應(yīng)該只接 
#收一次消息。為了滿足這一要求，mosquitto不斷跟蹤發(fā)送給客戶(hù)的消息。允許重復(fù)的 
#消息選項(xiàng)允許禁用此行為，如果您有大量的客戶(hù)端訂閱相同的主題集合，并且非常關(guān)注 
#最小化內(nèi)存使用的情況，那么這個(gè)選項(xiàng)可能是有用的。如果你事先知道你的客戶(hù)端永不 
#會(huì)有重疊的訂閱，那么你的客戶(hù)必須能夠正確處理重復(fù)的信息，即使在Qo = 2的時(shí)候， 
#你的客戶(hù)端也必須能夠正確地處理重復(fù)的信息 
#allow_duplicate_messages false 
# ================================================================= 
# Default listener 
# ================================================================= 
# 服務(wù)綁定的IP地址 
#bind_address 
# 服務(wù)綁定的端口號(hào) 
#port 1883 
# 允許的最大連接數(shù)，-1表示沒(méi)有限制 
#max_connections -1 
# cafile：CA證書(shū)文件 
# capath：CA證書(shū)目錄 
# certfile：PEM證書(shū)文件 
# keyfile：PEM密鑰文件 
#cafile 
#capath 
#certfile 
#keyfile 
# 必須提供證書(shū)以保證數(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)保存功能的開(kāi)關(guān) 
#autosave_on_changes false 
# 持久化功能的開(kāi)關(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 
# 是否記錄客戶(hù)端連接信息 
#connection_messages true 
# 是否記錄日志時(shí)間 
#log_timestamp true 
# ================================================================= 
# Security 
# ================================================================= 
# 客戶(hù)端ID的前綴限制，可用于保證安全性 
#clientid_prefixes 
# 允許匿名用戶(hù) 
#allow_anonymous true 
# 用戶(hù)/密碼文件，默認(rèn)格式：username:password 
#password_file 
# PSK格式密碼文件，默認(rèn)格式：identity:key 
#psk_file 
# pattern write sensor/%u/data 
# ACL權(quán)限配置，常用語(yǔ)法如下： 
# 用戶(hù)限制：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è)置橋接的客戶(hù)端ID 
#clientid 
# 橋接斷開(kāi)時(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 
# 橋接客戶(hù)端的用戶(hù)名 
#username 
# 橋接客戶(hù)端的密碼 
#password 
# bridge_cafile：橋接客戶(hù)端的CA證書(shū)文件 
# bridge_capath：橋接客戶(hù)端的CA證書(shū)目錄 
# bridge_certfile：橋接客戶(hù)端的PEM證書(shū)文件 
# bridge_keyfile：橋接客戶(hù)端的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í)。 客戶(hù)端不需要在發(fā)布或訂閱之前創(chuàng)建所需的主題，因?yàn)榇斫邮苊總€(gè)有效主題時(shí)不需要進(jìn)行任何預(yù)初始化。

主題使用/來(lái)分層次，以下是幾個(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。
單層次通配符可用于主題樹(shù)內(nèi)任何層次，并與多層次通配符一起使用。必須用于在頂層分隔符之后，除了當(dāng)自己指定時(shí)。因此，+和finance/+ 都是有效的，但是finance+無(wú)效。單層通配符可用于主題樹(shù)最后或者在主題樹(shù)內(nèi)，例如，finance/+與finance/+/ibm都是有效的。

示例如下：

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

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

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

示例如下：

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

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

關(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庫(kù)。提供了MQTT服務(wù)端和客戶(hù)端支持。

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)參考

客戶(hù)端： https://github.com/spartajet/IotWpfClient

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

Spring boot 和Vue開(kāi)發(fā)中CORS跨域問(wèn)題

mqttjs是支持MQTT協(xié)議的客戶(hù)端javascript庫(kù)，注意只是客戶(hù)端，并且通信方式是websockt，所以要在mosquitto服務(wù)器開(kāi)啟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)頁(yè)：https://github.com/spartajet/iot-demo-web

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

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

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