国产一级a片免费看高清,亚洲熟女中文字幕在线视频,黄三级高清在线播放,免费黄色视频在线看

1 什么是 MQTT ？

MQTT（Message Queuing Telemetry Transport，消息隊列遙測傳輸）是 IBM 開發(fā)的一個即時通訊協(xié)議，有可能成為物聯(lián)網(wǎng)的重要組成部分。MQTT 是基于二進(jìn)制消息的發(fā)布/訂閱編程模式的消息協(xié)議，如今已經(jīng)成為 OASIS 規(guī)范，由于規(guī)范很簡單，非常適合需要低功耗和網(wǎng)絡(luò)帶寬有限的 IoT 場景。MQTT官網(wǎng)

2 MQTTnet

MQTTnet 是一個基于 MQTT 通信的高性能 .NET 開源庫，它同時支持 MQTT 服務(wù)器端和客戶端。而且作者也保持更新，目前支持新版的.NET core，這也是選擇 MQTTnet 的原因。 MQTTnet 在 Github 并不是下載最多的 .NET 的 MQTT 開源庫，其他的還 MqttDotNet、nMQTT、M2MQTT 等

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/.

3 創(chuàng)建項目并導(dǎo)入類庫

這里我們使用 Visual Studio 2017 創(chuàng)建一個空解決方案，并在其中添加兩個項目，即一個服務(wù)端和一個客戶端，服務(wù)端項目模板選擇最新的 .NET Core 控制臺應(yīng)用，客戶端項目選擇傳統(tǒng)的 WinForm 窗體應(yīng)用程序。.NET Core 項目模板如下圖所示：

在解決方案在右鍵單擊-選擇“管理解決方案的 NuGet 程序包”-在“瀏覽”選項卡下面搜索 MQTTnet，為服務(wù)端項目和客戶端項目都安裝上 MQTTnet 庫，當(dāng)前最新穩(wěn)定版為 2.4.0。項目結(jié)構(gòu)如下圖所示：

4 服務(wù)端

MQTT 服務(wù)端主要用于與多個客戶端保持連接，并處理客戶端的發(fā)布和訂閱等邏輯。一般很少直接從服務(wù)端發(fā)送消息給客戶端（可以使用 mqttServer.Publish(appMsg); 直接發(fā)送消息），多數(shù)情況下服務(wù)端都是轉(zhuǎn)發(fā)主題匹配的客戶端消息，在系統(tǒng)中起到一個中介的作用。

4.1 創(chuàng)建服務(wù)端并啟動

創(chuàng)建服務(wù)端最簡單的方式是采用 MqttServerFactory 對象的 CreateMqttServer 方法來實現(xiàn)，該方法需要一個 MqttServerOptions 參數(shù)。

var options = new MqttServerOptions();var mqttServer = new MqttServerFactory().CreateMqttServer(options);
1
2

通過上述方式創(chuàng)建了一個 IMqttServer 對象后，調(diào)用其 StartAsync 方法即可啟動 MQTT 服務(wù)。值得注意的是：之前版本采用的是 Start 方法，作者也是緊跟 C# 語言新特性，能使用異步的地方也都改為異步方式。

await mqttServer.StartAsync();
1

4.2 驗證客戶端

在 MqttServerOptions 選項中，你可以使用 ConnectionValidator 來對客戶端連接進(jìn)行驗證。比如客戶端ID標(biāo)識 ClientId，用戶名 Username 和密碼 Password 等。

var options = new MqttServerOptions{    ConnectionValidator = c =>    {        if (c.ClientId.Length < 10)        {            return MqttConnectReturnCode.ConnectionRefusedIdentifierRejected;        }        if (c.Username != "xxx" || c.Password != "xxx")        {            return MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword;        }        return MqttConnectReturnCode.ConnectionAccepted;    }};
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

4.3 相關(guān)事件

服務(wù)端支持 ClientConnected、ClientDisconnected 和 ApplicationMessageReceived 事件，分別用來檢查客戶端連接、客戶端斷開以及接收客戶端發(fā)來的消息。

其中 ClientConnected 和 ClientDisconnected 事件的事件參數(shù)一個客戶端連接對象 ConnectedMqttClient，通過該對象可以獲取客戶端ID標(biāo)識 ClientId 和 MQTT 版本 ProtocolVersion。

ApplicationMessageReceived 的事件參數(shù)包含了客戶端ID標(biāo)識 ClientId 和 MQTT 應(yīng)用消息 MqttApplicationMessage 對象，通過該對象可以獲取主題 Topic、QoS QualityOfServiceLevel 和消息內(nèi)容 Payload 等信息。

5 客戶端

MQTT 與 HTTP 不同，后者是基于請求/響應(yīng)方式的，服務(wù)器端無法直接發(fā)送數(shù)據(jù)給客戶端。而 MQTT 是基于發(fā)布/訂閱模式的,所有的客戶端均與服務(wù)端保持連接狀態(tài)。

那么客戶端之間是如何通信的呢？

具體邏輯是：某些客戶端向服務(wù)端訂閱它感興趣（主題）的消息，另一些客戶端向服務(wù)端發(fā)布（主題）消息，服務(wù)端將訂閱和發(fā)布的主題進(jìn)行匹配，并將消息轉(zhuǎn)發(fā)給匹配通過的客戶端。

5.1 創(chuàng)建客戶端并連接

使用 MQTTnet 創(chuàng)建 MQTT 也非常簡單，只需要使用 MqttClientFactory 對象的 CreateMqttClient 方法即可。

var mqttClient = new MqttClientFactory().CreateMqttClient();
1

創(chuàng)建客戶端對象后，調(diào)用其異步方法 ConnectAsync 來連接到服務(wù)端。

await mqttClient.ConnectAsync(options);
1

調(diào)用該方法時需要傳遞一個 MqttClientTcpOptions 對象（之前的版本是在創(chuàng)建對象時使用該選項），該選項包含了客戶端ID標(biāo)識 ClientId、服務(wù)端地址（可以使用IP地址或域名）Server、端口號 Port、用戶名 UserName、密碼 Password 等信息。

var options = new MqttClientTcpOptions{    Server = "127.0.0.1",    ClientId = "c001",    UserName = "u001",    Password = "p001",    CleanSession = true};
1
2
3
4
5
6
7
8

5.2 相關(guān)事件

客戶端支持 Connected、Disconnected 和 ApplicationMessageReceived 事件，用來處理客戶端與服務(wù)端連接、客戶端從服務(wù)端斷開以及客戶端收到消息的事情。

5.2 訂閱消息

客戶端連接到服務(wù)端之后，可以使用 SubscribeAsync 異步方法訂閱消息，該方法可以傳入一個可枚舉或可變參數(shù)的主題過濾器 TopicFilter 參數(shù)，主題過濾器包含主題名和 QoS 等級。

mqttClient.SubscribeAsync(new List<TopicFilter> {    new TopicFilter("家/客廳/空調(diào)/#", MqttQualityOfServiceLevel.AtMostOnce)});
1
2
3

5.3 發(fā)布消息

發(fā)布消息前需要先構(gòu)建一個消息對象 MqttApplicationMessage，最直接的方法是使用其實構(gòu)造函數(shù)，傳入主題、內(nèi)容、Qos 等參數(shù)。

var appMsg = new MqttApplicationMessage("家/客廳/空調(diào)/開關(guān)", Encoding.UTF8.GetBytes("消息內(nèi)容"), MqttQualityOfServiceLevel.AtMostOnce, false);
1

得到 MqttApplicationMessage 消息對象后，通過客戶端對象調(diào)用其 PublishAsync 異步方法進(jìn)行消息發(fā)布。

mqttClient.PublishAsync(appMsg);
1

6 跟蹤消息

MQTTnet 提供了一個靜態(tài)類 MqttNetTrace 來對消息進(jìn)行跟蹤，該類可用于服務(wù)端和客戶端。MqttNetTrace 的事件 TraceMessagePublished 用于跟蹤服務(wù)端和客戶端應(yīng)用的日志消息，比如啟動、停止、心跳、消息訂閱和發(fā)布等。事件參數(shù) MqttNetTraceMessagePublishedEventArgs 包含了線程ID ThreadId、來源 Source、日志級別 Level、日志消息 Message、異常信息 Exception 等。

MqttNetTrace.TraceMessagePublished += MqttNetTrace_TraceMessagePublished;private static void MqttNetTrace_TraceMessagePublished(object sender, MqttNetTraceMessagePublishedEventArgs e){    Console.WriteLine($">> 線程ID：{e.ThreadId} 來源：{e.Source} 跟蹤級別：{e.Level} 消息: {e.Message}");    if (e.Exception != null)    {        Console.WriteLine(e.Exception);    }}
1
2
3
4
5
6
7
8
9
10
11

同時 MqttNetTrace 類還提供了4個不同消息等級的靜態(tài)方法，Verbose、Information、Warning 和 Error，用于給出不同級別的日志消息，該消息將會在 TraceMessagePublished 事件中輸出，你可以使用 e.Level 進(jìn)行過慮。

7 運(yùn)行效果

以下分別是服務(wù)端、客戶端1和客戶端2的運(yùn)行效果，其中客戶端1和客戶端2只是同一個項目運(yùn)行了兩個實例?？蛻舳?用于訂閱傳感器的“溫度”數(shù)據(jù)，并模擬上位機(jī)（如 APP 等）發(fā)送開關(guān)控制命令；客戶端2訂閱上位機(jī)傳來的“開關(guān)”控制命令，并模擬溫度傳感器上報溫度數(shù)據(jù)。

7.1 服務(wù)端

7.2 客戶端1

7.2 客戶端2

8 Demo代碼

8.1 服務(wù)端代碼

using MQTTnet;using MQTTnet.Core.Adapter;using MQTTnet.Core.Diagnostics;using MQTTnet.Core.Protocol;using MQTTnet.Core.Server;using System;using System.Text;using System.Threading;namespace MqttServerTest{    class Program    {        private static MqttServer mqttServer = null;        static void Main(string[] args)        {            MqttNetTrace.TraceMessagePublished += MqttNetTrace_TraceMessagePublished;            new Thread(StartMqttServer).Start();            while (true)            {                var inputString = Console.ReadLine().ToLower().Trim();                if (inputString == "exit")                {                    mqttServer?.StopAsync();                    Console.WriteLine("MQTT服務(wù)已停止！");                    break;                }                else if (inputString == "clients")                {                    foreach (var item in mqttServer.GetConnectedClients())                    {                        Console.WriteLine($"客戶端標(biāo)識：{item.ClientId}，協(xié)議版本：{item.ProtocolVersion}");                    }                }                else                {                    Console.WriteLine($"命令[{inputString}]無效！");                }            }        }        private static void StartMqttServer()        {            if (mqttServer == null)            {                try                {                    var options = new MqttServerOptions                    {                        ConnectionValidator = p =>                        {                            if (p.ClientId == "c001")                            {                                if (p.Username != "u001" || p.Password != "p001")                                {                                    return MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword;                                }                            }                            return MqttConnectReturnCode.ConnectionAccepted;                        }                    };                    mqttServer = new MqttServerFactory().CreateMqttServer(options) as MqttServer;                    mqttServer.ApplicationMessageReceived += MqttServer_ApplicationMessageReceived;                    mqttServer.ClientConnected += MqttServer_ClientConnected;                    mqttServer.ClientDisconnected += MqttServer_ClientDisconnected;                }                catch (Exception ex)                {                    Console.WriteLine(ex.Message);                    return;                }            }            mqttServer.StartAsync();            Console.WriteLine("MQTT服務(wù)啟動成功！");        }        private static void MqttServer_ClientConnected(object sender, MqttClientConnectedEventArgs e)        {            Console.WriteLine($"客戶端[{e.Client.ClientId}]已連接，協(xié)議版本：{e.Client.ProtocolVersion}");        }        private static void MqttServer_ClientDisconnected(object sender, MqttClientDisconnectedEventArgs e)        {            Console.WriteLine($"客戶端[{e.Client.ClientId}]已斷開連接！");        }        private static void MqttServer_ApplicationMessageReceived(object sender, MqttApplicationMessageReceivedEventArgs e)        {            Console.WriteLine($"客戶端[{e.ClientId}]>> 主題：{e.ApplicationMessage.Topic} 負(fù)荷：{Encoding.UTF8.GetString(e.ApplicationMessage.Payload)} Qos：{e.ApplicationMessage.QualityOfServiceLevel} 保留：{e.ApplicationMessage.Retain}");        }        private static void MqttNetTrace_TraceMessagePublished(object sender, MqttNetTraceMessagePublishedEventArgs e)        {            /*Console.WriteLine($">> 線程ID：{e.ThreadId} 來源：{e.Source} 跟蹤級別：{e.Level} 消息: {e.Message}");            if (e.Exception != null)            {                Console.WriteLine(e.Exception);            }*/        }    }}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108

8.2 客戶端代碼

using MQTTnet;using MQTTnet.Core;using MQTTnet.Core.Client;using MQTTnet.Core.Packets;using MQTTnet.Core.Protocol;using System;using System.Collections.Generic;using System.Text;using System.Threading.Tasks;using System.Windows.Forms;namespace MqttClientWin{    public partial class FmMqttClient : Form    {        private MqttClient mqttClient = null;        public FmMqttClient()        {            InitializeComponent();            Task.Run(async () => { await ConnectMqttServerAsync(); });        }        private async Task ConnectMqttServerAsync()        {            if (mqttClient == null)            {                mqttClient = new MqttClientFactory().CreateMqttClient() as MqttClient;                mqttClient.ApplicationMessageReceived += MqttClient_ApplicationMessageReceived;                mqttClient.Connected += MqttClient_Connected;                mqttClient.Disconnected += MqttClient_Disconnected;            }            try            {                var options = new MqttClientTcpOptions                {                    Server = "127.0.0.1",                    ClientId = Guid.NewGuid().ToString().Substring(0, 5),                    UserName = "u001",                    Password = "p001",                    CleanSession = true                };                await mqttClient.ConnectAsync(options);            }            catch (Exception ex)            {                Invoke((new Action(() =>                {                    txtReceiveMessage.AppendText($"連接到MQTT服務(wù)器失??！" + Environment.NewLine + ex.Message + Environment.NewLine);                })));            }        }        private void MqttClient_Connected(object sender, EventArgs e)        {            Invoke((new Action(() =>            {                txtReceiveMessage.AppendText("已連接到MQTT服務(wù)器！" + Environment.NewLine);            })));        }        private void MqttClient_Disconnected(object sender, EventArgs e)        {            Invoke((new Action(() =>            {                txtReceiveMessage.AppendText("已斷開MQTT連接！" + Environment.NewLine);            })));        }        private void MqttClient_ApplicationMessageReceived(object sender, MqttApplicationMessageReceivedEventArgs e)        {            Invoke((new Action(() =>            {                txtReceiveMessage.AppendText($">> {Encoding.UTF8.GetString(e.ApplicationMessage.Payload)}{Environment.NewLine}");            })));        }        private void BtnSubscribe_ClickAsync(object sender, EventArgs e)        {            string topic = txtSubTopic.Text.Trim();            if (string.IsNullOrEmpty(topic))            {                MessageBox.Show("訂閱主題不能為空！");                return;            }            if (!mqttClient.IsConnected)            {                MessageBox.Show("MQTT客戶端尚未連接！");                return;            }            mqttClient.SubscribeAsync(new List<TopicFilter> {                new TopicFilter(topic, MqttQualityOfServiceLevel.AtMostOnce)            });            txtReceiveMessage.AppendText($"已訂閱[{topic}]主題" + Environment.NewLine);            txtSubTopic.Enabled = false;            btnSubscribe.Enabled = false;        }        private void BtnPublish_Click(object sender, EventArgs e)        {            string topic = txtPubTopic.Text.Trim();            if (string.IsNullOrEmpty(topic))            {                MessageBox.Show("發(fā)布主題不能為空！");                return;            }            string inputString = txtSendMessage.Text.Trim();            var appMsg = new MqttApplicationMessage(topic, Encoding.UTF8.GetBytes(inputString), MqttQualityOfServiceLevel.AtMostOnce, false);            mqttClient.PublishAsync(appMsg);        }    }}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121

9 參考

• 快速搭建MQTT服務(wù)器(MQTTnet和Apache Apollo)
• MQTTnet
• “mqtt” - 可譯網(wǎng)
• MQTT Essentials