介绍完基础理论部分,下面在Windows平台上搭建一个简单的MQTT应用,进行简单的应用,整体架构如下图所示;
消息模型:
运用MQTT协议,设备可以很方便地连接到物联网云服务,管理设备并处理数据,最后应用到各种业务场景,如下图所示
前面介绍过,MQTT可以运行在几乎所有的平台,windows,linux什么的都可以,各种语言都有实现MQTT的组件,.net也好,Java也好,都有封装好的mqtt服务器和客户端组件或插件来实现,本系列是在.net平台下实现mqtt的服务器和客户端。
常见的MQTT服务器包括Eclipse Mosquitto、EMQ X、HiveMQ、RabbitMQ、MQTTNET等,本系列文章都是基于.net平台的mqtt服务通信,开发环境vs2022,.net framework4.8。
?
?
?
?
?
注意这里面用到了多线程及任务task,委托action,异步async及await的技术,在mqtt中必须使用这些技术,否则界面会卡死。
using MQTTnet.Client.Receiving;
using MQTTnet.Server;
using MQTTnet;
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.Protocol;
namespace MQTTNETServerForms
{
public partial class Form1 : Form
{
private IMqttServer server;//mqtt服务器对象
List<TopicItem> Topics = new List<TopicItem>();
public Form1()
{
InitializeComponent();
}
private void Form1_Load(object sender, EventArgs e)
{
//创建服务器对象
server = new MqttFactory().CreateMqttServer();
server.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(new Action<MqttApplicationMessageReceivedEventArgs>(Server_ApplicationMessageReceived));//绑定消息接收事件
server.ClientConnectedHandler = new MqttServerClientConnectedHandlerDelegate(new Action<MqttServerClientConnectedEventArgs>(Server_ClientConnected));//绑定客户端连接事件
server.ClientDisconnectedHandler = new MqttServerClientDisconnectedHandlerDelegate(new Action<MqttServerClientDisconnectedEventArgs>(Server_ClientDisconnected));//绑定客户端断开事件
server.ClientSubscribedTopicHandler = new MqttServerClientSubscribedHandlerDelegate(new Action<MqttServerClientSubscribedTopicEventArgs>(Server_ClientSubscribedTopic));//绑定客户端订阅主题事件
server.ClientUnsubscribedTopicHandler = new MqttServerClientUnsubscribedTopicHandlerDelegate(new Action<MqttServerClientUnsubscribedTopicEventArgs>(Server_ClientUnsubscribedTopic));//绑定客户端退订主题事件
server.StartedHandler = new MqttServerStartedHandlerDelegate(new Action<EventArgs>(Server_Started));//绑定服务端启动事件
server.StoppedHandler = new MqttServerStoppedHandlerDelegate(new Action<EventArgs>(Server_Stopped));//绑定服务端停止事件
}
/// <summary>
/// 绑定消息接收事件
/// </summary>
/// <param name="e"></param>
private void Server_ApplicationMessageReceived(MqttApplicationMessageReceivedEventArgs e)
{
string msg = e.ApplicationMessage.ConvertPayloadToString();
WriteLog(">>> 收到消息:" + msg + ",QoS =" + e.ApplicationMessage.QualityOfServiceLevel + ",客户端=" + e.ClientId + ",主题:" + e.ApplicationMessage.Topic);
}
/// <summary>
/// 绑定客户端连接事件
/// </summary>
/// <param name="e"></param>
private void Server_ClientConnected(MqttServerClientConnectedEventArgs e)
{
Task.Run(new Action(() =>
{
lbClients.BeginInvoke(new Action(() =>
{
lbClients.Items.Add(e.ClientId);
}));
}));
WriteLog(">>> 客户端" + e.ClientId + "连接");
}
/// <summary>
/// 绑定客户端断开事件
/// </summary>
/// <param name="e"></param>
private void Server_ClientDisconnected(MqttServerClientDisconnectedEventArgs e)
{
Task.Run(new Action(() =>
{
lbClients.BeginInvoke(new Action(() =>
{
lbClients.Items.Remove(e.ClientId);
}));
}));
WriteLog(">>> 客户端" + e.ClientId + "断开");
}
/// <summary>
/// 绑定客户端订阅主题事件
/// </summary>
/// <param name="e"></param>
private void Server_ClientSubscribedTopic(MqttServerClientSubscribedTopicEventArgs e)
{
Task.Run(new Action(() =>
{
var topic = Topics.FirstOrDefault(t => t.Topic == e.TopicFilter.Topic);
if (topic == null)
{
topic = new TopicItem { Topic = e.TopicFilter.Topic, Count = 0 };
Topics.Add(topic);
}
if (!topic.Clients.Exists(c => c == e.ClientId))
{
topic.Clients.Add(e.ClientId);
topic.Count++;
}
lvTopic.Invoke(new Action(() =>
{
this.lvTopic.Items.Clear();
}));
foreach (var item in this.Topics)
{
lvTopic.Invoke(new Action(() =>
{
this.lvTopic.Items.Add($"{item.Topic}:{item.Count}");
}));
}
}));
WriteLog(">>> 客户端" + e.ClientId + "订阅主题" + e.TopicFilter.Topic);
}
/// <summary>
/// 绑定客户端退订主题事件
/// </summary>
/// <param name="e"></param>
private void Server_ClientUnsubscribedTopic(MqttServerClientUnsubscribedTopicEventArgs e)
{
Task.Run(new Action(() =>
{
var topic = Topics.FirstOrDefault(t => t.Topic == e.TopicFilter);
if (topic != null)
{
topic.Count--;
topic.Clients.Remove(e.ClientId);
}
this.lvTopic.Items.Clear();
foreach (var item in this.Topics)
{
this.lvTopic.Items.Add($"{item.Topic}:{item.Count}");
}
}));
WriteLog(">>> 客户端" + e.ClientId + "退订主题" + e.TopicFilter);
}
/// <summary>
/// 绑定服务端启动事件
/// </summary>
/// <param name="e"></param>
private void Server_Started(EventArgs e)
{
WriteLog(">>> 服务端已启动!");
}
/// <summary>
/// 绑定服务端停止事件
/// </summary>
/// <param name="e"></param>
private void Server_Stopped(EventArgs e)
{
WriteLog(">>> 服务端已停止!");
}
/// <summary>
/// 显示日志
/// </summary>
/// <param name="message"></param>
public void WriteLog(string message)
{
if (txtMsg.InvokeRequired)
{
txtMsg.Invoke(new Action(() =>
{
txtMsg.Text = "";
txtMsg.Text = (message + "\r");
}));
}
else
{
txtMsg.Text = "";
txtMsg.Text = (message + "\r");
}
}
/// <summary>
/// 启动
/// </summary>
/// <param name="sender"></param>
/// <param name="e"></param>
[Obsolete]
private async void btnStart_Click(object sender, EventArgs e)
{
var optionBuilder = new MqttServerOptionsBuilder()
.WithDefaultEndpointBoundIPAddress(System.Net.IPAddress.Parse(this.txtIP.Text))
.WithDefaultEndpointPort(int.Parse(this.txtPort.Text))
.WithDefaultCommunicationTimeout(TimeSpan.FromMilliseconds(5000))
.WithConnectionValidator(t =>
{
string un = "", pwd = "";
un = this.txtUname.Text;
pwd = this.txtUpwd.Text;
if (t.Username != un || t.Password != pwd)
{
t.ReturnCode = MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword;
}
else
{
t.ReturnCode = MqttConnectReturnCode.ConnectionAccepted;
}
});
var option = optionBuilder.Build();
//启动
await server.StartAsync(option);
WriteLog(">>> 服务器启动成功");
}
/// <summary>
/// 停止
/// </summary>
/// <param name="sender"></param>
/// <param name="e"></param>
private void btnStop_Click(object sender, EventArgs e)
{
if (server != null)
{
server.StopAsync();
}
}
}
}
?注意这个端口,帐号,密码可以自己决定,Ip地址也是。
?注意这里面的代码,服务器上必须注册绑定实现的几个事件:消息接收事件?,客户端连接事件,客户端断开事件,客户端订阅主题事件,客户端退订主题事件,服务端启动事件,服务端停止事件
server.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(new Action<MqttApplicationMessageReceivedEventArgs>(Server_ApplicationMessageReceived));//绑定消息接收事件
server.ClientConnectedHandler = new MqttServerClientConnectedHandlerDelegate(new Action<MqttServerClientConnectedEventArgs>(Server_ClientConnected));//绑定客户端连接事件
server.ClientDisconnectedHandler = new MqttServerClientDisconnectedHandlerDelegate(new Action<MqttServerClientDisconnectedEventArgs>(Server_ClientDisconnected));//绑定客户端断开事件
server.ClientSubscribedTopicHandler = new MqttServerClientSubscribedHandlerDelegate(new Action<MqttServerClientSubscribedTopicEventArgs>(Server_ClientSubscribedTopic));//绑定客户端订阅主题事件
server.ClientUnsubscribedTopicHandler = new MqttServerClientUnsubscribedTopicHandlerDelegate(new Action<MqttServerClientUnsubscribedTopicEventArgs>(Server_ClientUnsubscribedTopic));//绑定客户端退订主题事件
server.StartedHandler = new MqttServerStartedHandlerDelegate(new Action<EventArgs>(Server_Started));//绑定服务端启动事件
server.StoppedHandler = new MqttServerStoppedHandlerDelegate(new Action<EventArgs>(Server_Stopped));//绑定服务端停止事件
?启动成功。
注意这里添加的与服务器不一样,别混错了
?
?布局使用的是常规的label,textbox,button?
?
using MQTTnet.Client.Options;
using MQTTnet.Client;
using MQTTnet.Extensions.ManagedClient;
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;
namespace MQTTNETClientForms
{
public partial class Form1 : Form
{
private IManagedMqttClient mqttClient;//客户端mqtt对象
public Form1()
{
InitializeComponent();
}
/// <summary>
/// 连接
/// </summary>
/// <param name="sender"></param>
/// <param name="e"></param>
private async void btnConn_Click(object sender, EventArgs e)
{
var mqttClientOptions = new MqttClientOptionsBuilder()
.WithClientId(this.txtId.Text)
.WithTcpServer(this.txtIP.Text, int.Parse(this.txtPort.Text))
.WithCredentials(this.txtName.Text, this.txtUpwd.Text);
var options = new ManagedMqttClientOptionsBuilder()
.WithAutoReconnectDelay(TimeSpan.FromSeconds(5))
.WithClientOptions(mqttClientOptions.Build())
.Build();
//开启
await mqttClient.StartAsync(options);
}
/// <summary>
/// 断开
/// </summary>
/// <param name="sender"></param>
/// <param name="e"></param>
private async void btnClose_Click(object sender, EventArgs e)
{
if (mqttClient != null)
{
if (mqttClient.IsStarted)
{
await mqttClient.StopAsync();
}
mqttClient.Dispose();
}
}
private void Form1_Load(object sender, EventArgs e)
{
var factory = new MqttFactory();
mqttClient = factory.CreateManagedMqttClient();//创建客户端对象
//绑定断开事件
mqttClient.UseDisconnectedHandler(async ee =>
{
WriteLog("与服务器之间的连接断开了,正在尝试重新连接");
// 等待 5s 时间
await Task.Delay(TimeSpan.FromSeconds(5));
try
{
mqttClient.UseConnectedHandler(tt =>
{
WriteLog(">>> 连接到服务成功");
});
}
catch (Exception ex)
{
WriteLog($"重新连接服务器失败:{ex}");
}
});
//绑定接收事件
mqttClient.UseApplicationMessageReceivedHandler(aa =>
{
try
{
string msg = aa.ApplicationMessage.ConvertPayloadToString();
WriteLog(">>> 消息:" + msg + ",QoS =" + aa.ApplicationMessage.QualityOfServiceLevel + ",客户端=" + aa.ClientId + ",主题:" + aa.ApplicationMessage.Topic);
}
catch (Exception ex)
{
WriteLog($"+ 消息 = " + ex.Message);
}
});
//绑定连接事件
mqttClient.UseConnectedHandler(ee =>
{
WriteLog(">>> 连接到服务成功");
});
}
/// <summary>
/// 显示日志
/// </summary>
/// <param name="message"></param>
private void WriteLog(string message)
{
if (txtMsg.InvokeRequired)
{
txtMsg.Invoke(new Action(() =>
{
txtMsg.Text = (message);
}));
}
else
{
txtMsg.Text = (message);
}
}
/// <summary>
/// 订阅
/// </summary>
/// <param name="sender"></param>
/// <param name="e"></param>
[Obsolete]
private async void btnSub_Click(object sender, EventArgs e)
{
if (string.IsNullOrWhiteSpace(this.txtTopic.Text))
{
WriteLog(">>> 请输入主题");
return;
}
//在 MQTT 中有三种 QoS 级别:
//At most once(0) 最多一次
//At least once(1) 至少一次
//Exactly once(2) 恰好一次
//await mqttClient.SubscribeAsync(new TopicFilterBuilder().WithTopic(this.tbTopic.Text).WithAtMostOnceQoS().Build());//最多一次, QoS 级别0
await mqttClient.SubscribeAsync(new TopicFilterBuilder().WithTopic(this.txtTopic.Text).WithAtLeastOnceQoS().Build());//恰好一次, QoS 级别1
WriteLog($">>> 成功订阅");
}
/// <summary>
/// 发布
/// </summary>
/// <param name="sender"></param>
/// <param name="e"></param>
private async void btnPub_Click(object sender, EventArgs e)
{
if (string.IsNullOrWhiteSpace(this.txtTopik.Text))
{
WriteLog(">>> 请输入主题");
return;
}
var result = await mqttClient.PublishAsync(
this.txtTopik.Text,
this.txtContent.Text,
MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce);//恰好一次, QoS 级别1
WriteLog($">>> 主题:{this.txtTopik.Text},消息:{this.txtContent.Text},结果: {result.ReasonCode}");
}
}
}
生成编译解决方案,成功后开始测试
找到生成的客户端debug目录下的.exe文件
连接成功,服务器看到客户端上线了
再运行一个客户端,连接服务器
?c1向cced主题发布一个消息,结果是c1,c2都收到了消息
同样,c2发布一个消息,c1,c2都收到了消息
c1关闭,服务器马上知道了
基于mqttnet的组件搭建的mqtt服务器和客户端通信成功,发布和订阅都ko,ko,ko。
讲解不易,分析不易,原创不易,整理不易,伙伴们动动你的金手指,你的支持是我最大的动力。
?