目录

基于开源库编写MQTT通讯

基于开源库编写MQTT通讯


1. MQTT是什么?

MQTT(Message Queuing Terlemetry Transport)消息队列遥测协议;是一种轻量级的发布/订阅消息传输协议,专为IOT、低带宽、高延迟的网络环境设计,具有高效、低耗、海量设备连接特性。

1.通讯原理

  • 发布(Publisher):发布消息到特定主题(Topic)
  • 订阅(Subscriber):订阅主题接收消息
  • 代理(Broker):消息路由管理,接收发布的消息,颁发给所有订阅者

示例:传感器(Publisher)发送消息到主题(Topic),代理(Broker)接收消息并检查该主题的所有客户端,订阅者(Subscriber)实时接收消息

2.消息结构

  • 主题(topic)+负载(payload)

3.必要条件

  • MQTT Broker: 消息代理服务器
  • 客户端库: 设备或应用需集成MQTT客户端
  • 端口: 默认非加密端口 1883,加密端口 8883(TLS/SSL)
  • 连接认证: 支持用户名/密码、客户端证书等认证方式
  • QoS服务质量/策略
QoS等级描述适用场景
0最多一次(无确认)实时性高,允许丢数据
1至少一次(需确认)数据需可靠但不重复
2精确一次(握手确认)关键数据,严格不重复

4. 关键特性

遗嘱消息(Last Will)

  • 设备异常断开时,Broker 自动发布预设消息(如“设备离线”),通知其他客户端。

保留消息(Retained)

  • Broker 保存主题的最新消息,新订阅者首次连接时立即获取。

主题层级(Topic Hierarchy)

  • 支持多级通配符(+ 单层,# 多层)
  • 例如: home/floor1/temperature; home/+/status(匹配所有楼层状态)

5.应用场景

  • 移动设备远程监控(AGV状态上报)
  • 跨厂区数据汇聚(通过云平台中转)

2. 开发交互UI

创建WindowsFormAPP项目,NuGet安装MQTTnet开源库(项目-属性-框架;需与程序包的依赖项一致<否则安装错误>: PM> NuGet\Install-Package MQTTnet -Version 2.8.2

控件

  • label、TextBox、ComboBox、Button
  • ListView(Dock=停靠<视图=Details;小Imagelist=Imagelist1;编辑列=Infoname、Content>)

容器

  • SplitContainer(Orienting=垂直;SplitterWith=1;BorderStyle=边框)

组件

  • Imagelist(添加图像)
  • timer(Enabled=true、Interval=1000)

状态

  • StatusStrip(系统时间、连接数量、版本说明)

Server端

  • 窗体设置(Text=标题、Font=字体、StartPosition=位置、FormBoardStyle=边框)
  • 给定服务IP,固定端口号,设置开始、停止服务、快捷打开客户端按钮
  • 设置日志消息显示窗口;设置状态栏

https://i-blog.csdnimg.cn/direct/30e45fded0014896a0d5b848718f9307.png

Client端

  • 技巧:复制FrmServer(修改窗口、Designer代码)
  • 容器:在SP1的Panel2添加SplitContainer2(Listview放在SP2的Panel1中);
  • 设置SP1的FixedPanel的Panel1不动;设置SP2的FixedPanel的Panel2不动
  • 客户端可发布主题消息
  • 设置连接、订阅、取消订阅、发布主题按钮;
  • 主题信息、给定QoS策略

https://i-blog.csdnimg.cn/direct/2da9f21b058345d5a42a88d45dc89a9a.png https://i-blog.csdnimg.cn/direct/f572eae085084a54ada1725c5eeced77.png


3. 服务器核心代码

初始化 -public FrmServer(){…}

  • 获取IP集合 (Dns.GetHostAddresses)
  • 绑定控件 (cmb_iplist.DataSource、.SelectedIndex)

创建服务器对象(IMqttServer) (->服务启动按钮点击事件)

  • 创建服务器配置 _ var optionsBuilder = new MqttServerOptionsBuilder()

    – 验证用户密码_ .WithConnectionValidator(c =>...

  • 实例化服务对象 _ mqttServer = new MqttFactory().CreateMqttServer();

  • 创建MQTT事件方法_ mqttServer.ClientConnected += MqttServer_ClientConnected;

    – 方法日志显示_ this.AddLog(0, "MQTT客户端已连接" + "ClietID:" + e.ClientId.Length);

  • 启动服务_ mqttServer.StartAsync(optionsBuilder.Build());

日志对象

  • 创建委托 _ public delegate void AddlogDelegate(int index, string info);
  • 委托方法_ private void AddlogMothod(int index, string info){...}
  • 委托对象_ private AddlogDelegate AddLog;
  • 对象绑定方法 _ this.AddLog = this.AddlogMothod;
  • 对象应用eg:_ this.AddLog(0, "MQTT服务端已停止");

状态栏

  • 系统时间_ this.tss_time.Text = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss");
  • 连接个数(连接/断开事件更新)_ this.tss_connnum.Text = mqttServer.GetClientSessionsStatusAsync().Result.Count.ToString();

其他

  • 停止服务_ mqttServer.StopAsync();
  • 清空日志_ this.lst_info.Items.Clear();
  • 打开客户端(注意“重新生成解决方案”)_ new FrmClient().Show();
  • 当前时间_ DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss");
  • 关闭窗体停止服务

4. 客户端核心代码

初始化

  • 获取IP集合 (Dns.GetHostAddresses)
  • 绑定控件 (cmb_iplist.DataSource、.SelectedIndex)

快捷打开客户端,ServerIP、Port自动填写

  • 创建带参数的初始化构造方法 _ public FrmClient(string ip, string port)

创建客户端对象(IMqttClient) (->客户端连接按钮点击事件)

  • 创建客户端配置_ var option = new MqttClientOptions() { ClientId = Guid.NewGuid().ToString("D") };
  • 创建通道配置_ option.ChannelOptions = new MqttClientTcpOptions()
  • 是否启用账户_ if (this.chk_isuserpwd.Checked)
  • 创建会话配置_ option.CleanSession = true;
  • 创建客户端对象_ mqttClient = new MqttFactory().CreateMqttClient();
  • 连接服务器_ mqttClient.ConnectAsync(option);
  • 绑定事件方法(添加日志)_ mqttClient.Connected += mqttClient_Connected;

日志显示

  • 创建委托方法 _与服务端一致
  • 创建委托对象_与服务端一致
  • 初始化对象绑定方法_与服务端一致
  • 对象引用eg_与服务端一致

其他

  • 断开连接_ mqttClient.DisconnectAsync();

5. 消息订阅与发布

客户端可以正确连接后,只有订阅与发布消息,才算真正进行数据通讯;消息的订阅和发布均在客户端进行,服务端只需提供一个服务供客户连接(桥梁的作用);小节解释消息订阅发布的核心代码。

QoS服务策略 (下拉框获取)

  • QoS服务策略(枚举类型)_ this.cmb_qos_pub.DataSource = Enum.GetNames(typeof(MqttQualityOfServiceLevel));

消息订阅

  • 订阅主题_ mqttClient.SubscribeAsync(new List<TopicFilter>(){...});

消息取消订阅

  • 取消订阅_ mqttClient.UnsubscribeAsync(this.txt_topic_sub.Text);

消息发布

  • 创建消息对象_ var msg = new MqttApplicationMessage(){...}
  • 发布消息_ mqttClient.PublishAsync(msg);

6. 通讯测试

https://i-blog.csdnimg.cn/direct/0f104609b3e94012a8784c60d3cbe1f1.png


7. MQTT与PLC通讯

将msg对象中的Payload更改为PLC的寄存器即可

自动发布

  • 添加定时器timer1_设置频率,事件(消息内容,自动发布)
  • 连接成功时启动定时器_ this.timer1.Enabled = true;

消息内容

  • 添加引用(西门子通讯库)
  • 创建PLC对象
  • 读取PLC变量
  • 添加到msg中的Payload中_ Payload = Encoding.UTF8.GetBytes(plcmsg),

https://i-blog.csdnimg.cn/direct/6a6a8940a7e84658ad04cbcca27c184d.png


最后. 核心总结

1. 开发实现

服务端:

  • 功能:IP/端口配置、启动/停止服务、连接监控
  • 初始化 MqttServer 对象
  • 处理连接/断开事件(日志记录、状态更新)

客户端:

  • 功能:连接/断开、订阅/取消主题、消息发布
  • 配置 MqttClient(IP、端口、认证)
  • 实现订阅 (SubscribeAsync) /发布 (PublishAsync)

2. 进阶应用

PLC 集成:

  • 自动发布:定时器读取 PLC 数据并推送
  • 数据格式:Payload 封装寄存器值(如 Siemens PLC 数据)

异常测试:

  • 基础测试:服务端启停、客户端连接/断开
  • 消息流验证:订阅发布功能、QoS 策略生效
  • 异常测试:网络断开重连、遗嘱消息触发

附:关键代码片段

服务端启动:

var options = new MqttServerOptionsBuilder().WithDefaultPort(1883).Build();
mqttServer = new MqttFactory().CreateMqttServer();
mqttServer.StartAsync(options);

客户端发布消息:

var msg = new MqttApplicationMessage {
    Topic = "sensor/temp",
    Payload = Encoding.UTF8.GetBytes("25℃"),
    QoS = MqttQualityOfServiceLevel.AtLeastOnce
};
client.PublishAsync(msg);

PLC 数据读取:

var plcValue = SiemensPLC.Read("DB1.DBD0"); // 读取浮点数
var payload = $"{{\"temperature\": {plcValue}}}";


  1. 创建TopicFilter对象_ new TopicFilter(this.txt_topic_sub.Text, (MqttQualityOfServiceLevel)Enum.Parse(typeof(MqttQualityOfServiceLevel),this.cmb_qos_sub.Text))
  2. 消息对象赋值_ Topic = this.txt_topic_pub.Text, QualityOfServiceLevel =(MqttQualityOfServiceLevel)Enum.Parse(typeof(MqttQualityOfServiceLevel), this.cmb_qos_pub.Text), Payload = Encoding.UTF8.GetBytes(this.txt_msg_pub.Text), Retain = false,