MQTT（Message Queuing Telemetry Transport）

即時通訊協(xié)議，

開發(fā)商 IBM

MQTT(消息隊列遙測傳輸)是ISO?標(biāo)準(zhǔn)(ISO/IEC PRF 20922)下基于發(fā)布/訂閱范式的消息協(xié)議。它工作在 TCP/IP協(xié)議族上，是為硬件性能低下的遠(yuǎn)程設(shè)備以及網(wǎng)絡(luò)狀況糟糕的情況下而設(shè)計的發(fā)布/訂閱型消息協(xié)議，為此，它需要一個消息中間件?。

MQTT是一個基于客戶端-服務(wù)器的消息發(fā)布/訂閱傳輸協(xié)議。MQTT協(xié)議是輕量、簡單、開放和易于實現(xiàn)的，這些特點使它適用范圍非常廣泛。在很多情況下，包括受限的環(huán)境中，如：機(jī)器與機(jī)器（M2M）通信和物聯(lián)網(wǎng)（IoT）。其在，通過衛(wèi)星鏈路通信傳感器、偶爾撥號的醫(yī)療設(shè)備、智能家居、及一些小型化設(shè)備中已廣泛使用。

新建Winform應(yīng)用程序，將默認(rèn)的Form1重命名為FormMqttServer

管理Nuget包，輸入關(guān)鍵字Mqtt

點擊安裝

安裝完成后示例如圖

我們使用MQTTnet.dll，程序集 MQTTnet, Version=4.3.6.1152。應(yīng)用程序.net framework版本需要4.6.1或者以上

窗體?FormMqttServer 設(shè)計如下：

窗體 FormMqttServer設(shè)計器代碼如下：

文件 FormMqttServer.Designer.cs?

namespace CultureDemo
{partial class FormMqttServer{/// <summary>/// Required designer variable./// </summary>private System.ComponentModel.IContainer components = null;/// <summary>/// Clean up any resources being used./// </summary>/// <param name="disposing">true if managed resources should be disposed; otherwise, false.</param>protected override void Dispose(bool disposing){if (disposing && (components != null)){components.Dispose();}base.Dispose(disposing);}#region Windows Form Designer generated code/// <summary>/// Required method for Designer support - do not modify/// the contents of this method with the code editor./// </summary>private void InitializeComponent(){this.lvTopic = new System.Windows.Forms.ListBox();this.lbClients = new System.Windows.Forms.ListBox();this.rtxtMessage = new System.Windows.Forms.RichTextBox();this.label5 = new System.Windows.Forms.Label();this.label4 = new System.Windows.Forms.Label();this.btnStop = new System.Windows.Forms.Button();this.btnStart = new System.Windows.Forms.Button();this.txtPassword = new System.Windows.Forms.TextBox();this.label3 = new System.Windows.Forms.Label();this.txtUsername = new System.Windows.Forms.TextBox();this.label2 = new System.Windows.Forms.Label();this.txtPort = new System.Windows.Forms.TextBox();this.label1 = new System.Windows.Forms.Label();this.txtIP = new System.Windows.Forms.TextBox();this.lb = new System.Windows.Forms.Label();this.dgvTopic = new System.Windows.Forms.DataGridView();this.Column1 = new System.Windows.Forms.DataGridViewTextBoxColumn();this.Column2 = new System.Windows.Forms.DataGridViewTextBoxColumn();this.Column3 = new System.Windows.Forms.DataGridViewTextBoxColumn();this.Column4 = new System.Windows.Forms.DataGridViewTextBoxColumn();((System.ComponentModel.ISupportInitialize)(this.dgvTopic)).BeginInit();this.SuspendLayout();// // lvTopic// this.lvTopic.FormattingEnabled = true;this.lvTopic.ItemHeight = 12;this.lvTopic.Location = new System.Drawing.Point(686, 35);this.lvTopic.Name = "lvTopic";this.lvTopic.Size = new System.Drawing.Size(300, 136);this.lvTopic.TabIndex = 31;// // lbClients// this.lbClients.FormattingEnabled = true;this.lbClients.ItemHeight = 12;this.lbClients.Location = new System.Drawing.Point(343, 35);this.lbClients.Name = "lbClients";this.lbClients.Size = new System.Drawing.Size(300, 136);this.lbClients.TabIndex = 30;// // rtxtMessage// this.rtxtMessage.Location = new System.Drawing.Point(24, 188);this.rtxtMessage.Name = "rtxtMessage";this.rtxtMessage.Size = new System.Drawing.Size(467, 338);this.rtxtMessage.TabIndex = 29;this.rtxtMessage.Text = "";// // label5// this.label5.AutoSize = true;this.label5.Location = new System.Drawing.Point(684, 17);this.label5.Name = "label5";this.label5.Size = new System.Drawing.Size(53, 12);this.label5.TabIndex = 28;this.label5.Text = "主題列表";// // label4// this.label4.AutoSize = true;this.label4.Location = new System.Drawing.Point(341, 17);this.label4.Name = "label4";this.label4.Size = new System.Drawing.Size(65, 12);this.label4.TabIndex = 27;this.label4.Text = "連接客戶端";// // btnStop// this.btnStop.Location = new System.Drawing.Point(129, 148);this.btnStop.Name = "btnStop";this.btnStop.Size = new System.Drawing.Size(75, 23);this.btnStop.TabIndex = 26;this.btnStop.Text = "停止";this.btnStop.UseVisualStyleBackColor = true;this.btnStop.Click += new System.EventHandler(this.btnStop_Click);// // btnStart// this.btnStart.Location = new System.Drawing.Point(32, 148);this.btnStart.Name = "btnStart";this.btnStart.Size = new System.Drawing.Size(75, 23);this.btnStart.TabIndex = 25;this.btnStart.Text = "啟動";this.btnStart.UseVisualStyleBackColor = true;this.btnStart.Click += new System.EventHandler(this.btnStart_Click);// // txtPassword// this.txtPassword.Location = new System.Drawing.Point(91, 121);this.txtPassword.Name = "txtPassword";this.txtPassword.Size = new System.Drawing.Size(211, 21);this.txtPassword.TabIndex = 24;this.txtPassword.Text = "1";// // label3// this.label3.AutoSize = true;this.label3.Location = new System.Drawing.Point(44, 124);this.label3.Name = "label3";this.label3.Size = new System.Drawing.Size(29, 12);this.label3.TabIndex = 23;this.label3.Text = "密碼";// // txtUsername// this.txtUsername.Location = new System.Drawing.Point(91, 94);this.txtUsername.Name = "txtUsername";this.txtUsername.Size = new System.Drawing.Size(211, 21);this.txtUsername.TabIndex = 22;this.txtUsername.Text = "a";// // label2// this.label2.AutoSize = true;this.label2.Location = new System.Drawing.Point(44, 97);this.label2.Name = "label2";this.label2.Size = new System.Drawing.Size(29, 12);this.label2.TabIndex = 21;this.label2.Text = "賬號";// // txtPort// this.txtPort.Location = new System.Drawing.Point(91, 58);this.txtPort.Name = "txtPort";this.txtPort.Size = new System.Drawing.Size(211, 21);this.txtPort.TabIndex = 20;this.txtPort.Text = "12345";// // label1// this.label1.AutoSize = true;this.label1.Location = new System.Drawing.Point(44, 61);this.label1.Name = "label1";this.label1.Size = new System.Drawing.Size(29, 12);this.label1.TabIndex = 19;this.label1.Text = "端口";// // txtIP// this.txtIP.Location = new System.Drawing.Point(91, 16);this.txtIP.Name = "txtIP";this.txtIP.Size = new System.Drawing.Size(211, 21);this.txtIP.TabIndex = 18;this.txtIP.Text = "127.0.0.1";// // lb// this.lb.AutoSize = true;this.lb.Location = new System.Drawing.Point(32, 19);this.lb.Name = "lb";this.lb.Size = new System.Drawing.Size(41, 12);this.lb.TabIndex = 17;this.lb.Text = "IP地址";// // dgvTopic// this.dgvTopic.AllowUserToAddRows = false;this.dgvTopic.AllowUserToDeleteRows = false;this.dgvTopic.ColumnHeadersHeightSizeMode = System.Windows.Forms.DataGridViewColumnHeadersHeightSizeMode.AutoSize;this.dgvTopic.Columns.AddRange(new System.Windows.Forms.DataGridViewColumn[] {this.Column1,this.Column2,this.Column3,this.Column4});this.dgvTopic.Location = new System.Drawing.Point(522, 188);this.dgvTopic.Name = "dgvTopic";this.dgvTopic.ReadOnly = true;this.dgvTopic.RowTemplate.Height = 23;this.dgvTopic.Size = new System.Drawing.Size(464, 338);this.dgvTopic.TabIndex = 32;// // Column1// this.Column1.HeaderText = "Topic";this.Column1.Name = "Column1";this.Column1.ReadOnly = true;// // Column2// this.Column2.HeaderText = "ClientId";this.Column2.Name = "Column2";this.Column2.ReadOnly = true;// // Column3// this.Column3.HeaderText = "Qos";this.Column3.Name = "Column3";this.Column3.ReadOnly = true;// // Column4// this.Column4.HeaderText = "SessionItems";this.Column4.Name = "Column4";this.Column4.ReadOnly = true;// // FormMqttServer// this.AutoScaleDimensions = new System.Drawing.SizeF(6F, 12F);this.AutoScaleMode = System.Windows.Forms.AutoScaleMode.Font;this.ClientSize = new System.Drawing.Size(1013, 538);this.Controls.Add(this.dgvTopic);this.Controls.Add(this.lvTopic);this.Controls.Add(this.lbClients);this.Controls.Add(this.rtxtMessage);this.Controls.Add(this.label5);this.Controls.Add(this.label4);this.Controls.Add(this.btnStop);this.Controls.Add(this.btnStart);this.Controls.Add(this.txtPassword);this.Controls.Add(this.label3);this.Controls.Add(this.txtUsername);this.Controls.Add(this.label2);this.Controls.Add(this.txtPort);this.Controls.Add(this.label1);this.Controls.Add(this.txtIP);this.Controls.Add(this.lb);this.Name = "FormMqttServer";this.Text = "FormMqttServer";this.Load += new System.EventHandler(this.FormMqttServer_Load);((System.ComponentModel.ISupportInitialize)(this.dgvTopic)).EndInit();this.ResumeLayout(false);this.PerformLayout();}#endregionprivate System.Windows.Forms.ListBox lvTopic;private System.Windows.Forms.ListBox lbClients;private System.Windows.Forms.RichTextBox rtxtMessage;private System.Windows.Forms.Label label5;private System.Windows.Forms.Label label4;private System.Windows.Forms.Button btnStop;private System.Windows.Forms.Button btnStart;private System.Windows.Forms.TextBox txtPassword;private System.Windows.Forms.Label label3;private System.Windows.Forms.TextBox txtUsername;private System.Windows.Forms.Label label2;private System.Windows.Forms.TextBox txtPort;private System.Windows.Forms.Label label1;private System.Windows.Forms.TextBox txtIP;private System.Windows.Forms.Label lb;private System.Windows.Forms.DataGridView dgvTopic;private System.Windows.Forms.DataGridViewTextBoxColumn Column1;private System.Windows.Forms.DataGridViewTextBoxColumn Column2;private System.Windows.Forms.DataGridViewTextBoxColumn Column3;private System.Windows.Forms.DataGridViewTextBoxColumn Column4;}
}

窗體FormMqttServer代碼如下【MQTT服務(wù)端相關(guān)代碼】

文件FormMqttServer.cs

using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Data;
using System.Drawing;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Windows.Forms;using MQTTnet.Server;
using MQTTnet;
using MQTTnet.Protocol;
using System.Collections;namespace CultureDemo
{public partial class FormMqttServer : Form{private MqttServer server;//mqtt服務(wù)器對象public FormMqttServer(){InitializeComponent();rtxtMessage.ReadOnly = true;/** 在MQTT協(xié)議中，一個MQTT數(shù)據(jù)包由：固定頭（Fixed header）、可變頭（Variable header）、消息體（payload）三部分構(gòu)成。
固定頭（Fixed header）：存在于所有MQTT數(shù)據(jù)包中，表示數(shù)據(jù)包類型及數(shù)據(jù)包的分組類標(biāo)識。
可變頭（Variable header）：存在于部分MQTT數(shù)據(jù)包中，數(shù)據(jù)包類型決定了可變頭是否存在及其具體內(nèi)容。
消息體（Payload）：存在于部分MQTT數(shù)據(jù)包中，表示客戶端收到的具體內(nèi)容。*/}/// <summary>/// 打印相關(guān)消息/// </summary>/// <param name="contents"></param>private void DisplayMessage(string contents) {if (!this.IsHandleCreated) {return;}this.BeginInvoke(new Action(() => {if (rtxtMessage.TextLength >= 40960) {rtxtMessage.Clear();}rtxtMessage.AppendText($"{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff")}->{contents}\n");rtxtMessage.ScrollToCaret();}));}/// <summary>/// 初始化Mqtt服務(wù)并啟動服務(wù)/// </summary>/// <param name="ip">IPV4地址</param>/// <param name="port">端口：0~65535之間</param>private Task StartMqttServer(string ip, int port){MqttServerOptions mqttServerOptions =new MqttServerOptionsBuilder().WithDefaultEndpoint().WithDefaultEndpointBoundIPAddress(System.Net.IPAddress.Parse(ip)).WithDefaultEndpointPort(port)   .WithDefaultCommunicationTimeout(TimeSpan.FromMilliseconds(5000)).Build();server = new MqttFactory().CreateMqttServer(mqttServerOptions); // create MQTT service objectserver.ValidatingConnectionAsync += Server_ValidatingConnectionAsync;//驗證用戶名和密碼server.ClientConnectedAsync += Server_ClientConnectedAsync;//綁定客戶端連接事件server.ClientDisconnectedAsync += Server_ClientDisconnectedAsync;//綁定客戶端斷開事件server.ClientSubscribedTopicAsync += Server_ClientSubscribedTopicAsync;//綁定客戶端訂閱主題事件server.ClientUnsubscribedTopicAsync += Server_ClientUnsubscribedTopicAsync;//綁定客戶端退訂主題事件server.InterceptingPublishAsync += Server_InterceptingPublishAsync;//消息接收事件server.ClientAcknowledgedPublishPacketAsync += Server_ClientAcknowledgedPublishPacketAsync;server.InterceptingClientEnqueueAsync += Server_InterceptingClientEnqueueAsync;server.ApplicationMessageNotConsumedAsync += Server_ApplicationMessageNotConsumedAsync;server.StartedAsync += Server_StartedAsync;//綁定服務(wù)端啟動事件server.StoppedAsync += Server_StoppedAsync;//綁定服務(wù)端停止事件return server.StartAsync();}private Task Server_ApplicationMessageNotConsumedAsync(ApplicationMessageNotConsumedEventArgs e){try{DisplayMessage($"【MessageNotConsumed】-SenderId:{e.SenderId}-Message:{e.ApplicationMessage.ConvertPayloadToString()}");}catch (Exception ex){DisplayMessage($"Server_ApplicationMessageNotConsumedAsync出現(xiàn)異常：{ex.Message}");}return Task.CompletedTask;}private Task Server_InterceptingClientEnqueueAsync(InterceptingClientApplicationMessageEnqueueEventArgs e){try{DisplayMessage($"【InterceptingClientEnqueue】-SenderId:{e.SenderClientId}-Message:{e.ApplicationMessage.ConvertPayloadToString()}");}catch (Exception ex){DisplayMessage($"Server_InterceptingClientEnqueueAsync出現(xiàn)異常：{ex.Message}");}return Task.CompletedTask;}private Task Server_ClientAcknowledgedPublishPacketAsync(ClientAcknowledgedPublishPacketEventArgs e){try{DisplayMessage($"【ClientAcknowledgedPublishPacket】-SenderId:{e.ClientId}-Message:{Encoding.UTF8.GetString(e.PublishPacket.PayloadSegment.ToArray())}");}catch (Exception ex){DisplayMessage($"Server_ClientAcknowledgedPublishPacketAsync出現(xiàn)異常：{ex.Message}");}return Task.CompletedTask;}/// <summary>/// 消息接收/// </summary>/// <param name="arg"></param>/// <returns></returns>private Task Server_InterceptingPublishAsync(InterceptingPublishEventArgs e){try{string client = e.ClientId;string topic = e.ApplicationMessage.Topic;string contents = e.ApplicationMessage.ConvertPayloadToString();//Encoding.UTF8.GetString(arg.ApplicationMessage.PayloadSegment.ToArray());DisplayMessage($"接收到消息：Client：【{client}】 Topic：【{topic}】 Message：【{contents}】");}catch (Exception ex){DisplayMessage($"Server_InterceptingPublishAsync出現(xiàn)異常：{ex.Message}");}return Task.CompletedTask;}private void FormMqttServer_Load(object sender, EventArgs e){rtxtMessage.Text = @"MQTT服務(wù)端關(guān)鍵事件說明：
ClientConnectedAsync-->綁定客戶端連接事件
ValidatingConnectionAsync-->驗證用戶名和密碼
ClientSubscribedTopicAsync-->綁定客戶端訂閱主題事件
InterceptingPublishAsync-->消息接收事件
";}private Task Server_StoppedAsync(EventArgs arg){return Task.Run(new Action(() =>{DisplayMessage($"服務(wù)端【{txtIP.Text}:{txtPort.Text}】已停止MQTT");}));}private Task Server_StartedAsync(EventArgs e){return Task.Run(new Action(() =>{DisplayMessage($"服務(wù)端【{txtIP.Text}:{txtPort.Text}】已啟用MQTT");}));}private Task Server_ClientUnsubscribedTopicAsync(ClientUnsubscribedTopicEventArgs e){return Task.Run(new Action(() =>{lvTopic.Invoke(new Action(() =>{string itemContents = $"【{e.ClientId}】訂閱主題【{e.TopicFilter}】";if (lvTopic.Items.Contains(itemContents)){//如果存在該項，就移除lvTopic.Items.Remove(itemContents);}IDictionary collection = e.SessionItems;string[] sessionItemContents = new string[collection.Count];int idx = 0;foreach (object key in collection.Keys){sessionItemContents[idx++] = $"[{key}:{collection[key]}]";}}));DisplayMessage($"客戶端【{e.ClientId}】退訂主題【{e.TopicFilter}】");}));}private Task Server_ClientSubscribedTopicAsync(ClientSubscribedTopicEventArgs e){return Task.Run(new Action(() =>{lvTopic.Invoke(new Action(() =>{string itemContents = $"【{e.ClientId}】訂閱主題【{e.TopicFilter.Topic}】";if (!lvTopic.Items.Contains(itemContents)){//如果不存在該項，就添加lvTopic.Items.Add(itemContents);}IDictionary collection = e.SessionItems;string[] sessionItemContents = new string[collection.Count];int idx = 0;foreach (object key in collection.Keys){sessionItemContents[idx++] = $"[{key}:{collection[key]}]";}dgvTopic.Rows.Add(e.TopicFilter.Topic, e.ClientId, e.TopicFilter.QualityOfServiceLevel, string.Join(",", sessionItemContents));}));DisplayMessage($"客戶端【{e.ClientId}】訂閱主題【{e.TopicFilter.Topic}】");}));}private Task Server_ClientDisconnectedAsync(ClientDisconnectedEventArgs e){return Task.Run(new Action(() =>{lbClients.BeginInvoke(new Action(() =>{lbClients.Items.Remove(e.ClientId);}));DisplayMessage($"客戶端已斷開.ClientId:【{e.ClientId}】,Endpoint:【{e.Endpoint}】.ReasonCode:【{e.ReasonCode}】,DisconnectType:【{e.DisconnectType}】");}));}/// <summary>/// 綁定客戶端連接事件/// </summary>/// <param name="arg"></param>/// <returns></returns>private Task Server_ClientConnectedAsync(ClientConnectedEventArgs e){return Task.Run(new Action(() =>{lbClients.BeginInvoke(new Action(() =>{lbClients.Items.Add(e.ClientId);}));DisplayMessage($"客戶端已連接.ClientId:【{e.ClientId}】,Endpoint:【{e.Endpoint}】");}));}private Task Server_ValidatingConnectionAsync(ValidatingConnectionEventArgs e){return Task.Run(new Action(() =>{string UserName = txtUsername.Text;string Password = txtPassword.Text;if (e.UserName == UserName && e.Password == Password){e.ReasonCode = MqttConnectReasonCode.Success;DisplayMessage($"客戶端已驗證成功.ClientId:【{e.ClientId}】,Endpoint:【{e.Endpoint}】");}else{e.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword;DisplayMessage($"客戶端驗證失敗.ClientId:【{e.ClientId}】,Endpoint:【{e.Endpoint}】");}}));}private async void btnStart_Click(object sender, EventArgs e){await StartMqttServer(txtIP.Text, int.Parse(txtPort.Text));btnStart.Enabled = false;}private async void btnStop_Click(object sender, EventArgs e){await server.StopAsync();btnStart.Enabled = true;}}
}

程序運(yùn)行如圖：

?連接、訂閱、發(fā)布消息如圖