摘要：本文是其中的一個解決方案。地址客戶端服務端前端網頁介紹，消息隊列遙測傳輸是開發的一個即時通訊協議，有可能成為物聯網的重要組成部分。必須用于在頂層分隔符之后，除了當自己指定時。

最近，本實驗室大量上馬云測量，云監控方面的項目，大概是屬于物聯網應用的一個分支。老板也有將舊有儀器改造的想法，所以要實現儀器設備的云控制。本文是其中的一個解決方案。

消息隊列：MQTT，服務器使用centos，安裝mosquitto

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

服務端采用spring-cloud微服務框架

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

客戶端的測量采集數據程序使用TPL Dataflow

這個解決方案加入如圖(請原諒我的懶惰)：

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

客戶端跟隨儀器，原則上一臺儀器一個控制程序，當然也可以有多個。客戶端實現了對儀器所有硬件設備的控制，對所有數據的采集。客戶端可以有界面，也可以沒有界面，一般來說，我們是需要一個界面的，客戶端可以獨立完成測量任務。

客戶端集成了網絡通信功能，可以完全替代用戶對客戶端的操作使用。具體架構如下（請原諒我的懶惰）：

其中，客戶端界面和網絡接口是等效的，可以多帶帶控制，也可以共同控制。

服務端基于spring-cloud微服務框架，主要提供服務發現，用戶管理，權限管理，設備管理，MQTT節點管理等管理功能

前端網頁是用戶通過網絡操作儀器設備的交互接口。采用日前流行的Vue框架，由于是后臺管理模式，就使用Element-admin-ui這個框架。

客戶端： https://github.com/spartajet/IotWpfClient
服務端：https://github.com/spartajet/iot-demo-server
前端網頁：https://github.com/spartajet/iot-demo-web

MQTT（Message Queuing Telemetry Transport，消息隊列遙測傳輸）是IBM開發的一個即時通訊協議，有可能成為物聯網的重要組成部分。該協議支持所有平臺，幾乎可以把所有聯網物品和外部連接起來，被用來當做傳感器和制動器（比如通過Twitter讓房屋聯網）的通信協議。

MQTT協議是為大量計算能力有限，且工作在低帶寬、不可靠的網絡的遠程傳感器和控制設備通訊而設計的協議，它具有以下主要的幾項特性：
1、使用發布/訂閱消息模式，提供一對多的消息發布，解除應用程序耦合；
2、對負載內容屏蔽的消息傳輸；
3、使用 TCP/IP 提供網絡連接；
4、有三種消息發布服務質量：
“至多一次”，消息發布完全依賴底層 TCP/IP 網絡。會發生消息丟失或重復。這一級別可用于如下情況，環境傳感器數據，丟失一次讀記錄無所謂，因為不久后還會有第二次發送。
“至少一次”，確保消息到達，但消息重復可能會發生。
“只有一次”，確保消息到達一次。這一級別可用于如下情況，在計費系統中，消息重復或丟失會導致不正確的結果。
5、小型傳輸，開銷很小（固定長度的頭部是 2 字節），協議交換最小化，以降低網絡流量；
6、使用 Last Will 和 Testament 特性通知有關各方客戶端異常中斷的機制；

詳細的MQTT協議內容請參考：https://mcxiaoke.gitbooks.io/mqtt-cn/content/mqtt/01-Introduction.html

本文使用開源的MQTT 服務器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包。要安裝它，我們將首先安裝一個名為Extra Packages for Enterprise Linux或EPEL的額外軟件存儲庫。

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工具設置密碼,使用方法如下：

?  ~ 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指令生成密碼

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

參考文獻：https://www.imooc.com/article/19459

brew services start/stop mosquitto

systemctl start/stop/restart mosquitto

與消息隊列相比，主題非常輕量級。 客戶端不需要在發布或訂閱之前創建所需的主題，因為代理接受每個有效主題時不需要進行任何預初始化。

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

sensors/COMPUTER_NAME/temperature/HARDDRIVE_NAME

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

+號僅僅匹配一個主題層次。例如，finance/stock/+匹配finance/stock/ibm與finance/stock/xyz，但是不匹配finance/stock/ibm/closingprice。因為單層次通配符僅僅匹配一個層次，finance/+不匹配finance。
單層次通配符可用于主題樹內任何層次，并與多層次通配符一起使用。必須用于在頂層分隔符之后，除了當自己指定時。因此，+和finance/+ 都是有效的，但是finance+無效。單層通配符可用于主題樹最后或者在主題樹內，例如，finance/+與finance/+/ibm都是有效的。

示例如下：

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

#號可以匹配主題內任何層次

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

示例如下：

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

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

關于如何生成數據以及使用TPL Dataflow數據采集和處理，請參考我的另一篇博客：

像Labview一樣，使用C#構建測量數據流式處理框架

這里重點介紹MQTTnet的使用

MQTTnet是一個高性能的MQTT基礎連接.NET庫。提供了MQTT服務端和客戶端支持。

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+

重點是：支持異步

連接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()方法，是異步連接。

發布消息

首先要構建消息

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

然后異步發送

this._mqttClient.PublishAsync(message);

詳細代碼請參考

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

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

Spring boot 和Vue開發中CORS跨域問題

mqttjs是支持MQTT協議的客戶端javascript庫，注意只是客戶端，并且通信方式是websockt，所以要在mosquitto服務器開啟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()

連接服務器：

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

設置連接后的事件，要訂閱相關主題的消息

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()
})

其他內容請參考項目源碼：

前端網頁：https://github.com/spartajet/iot-demo-web

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

對于設備管理，用戶權限等功能，打算用hsweb大神的物聯網框架hsweb-iot-cloud，也不排除自己開發的可能，看我的時間和項目需求

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