当前位置：首页 > news >正文

基于C#的MQTT通信实战：从EMQX搭建到发布订阅全解析

news 来源：原创 2025/5/18 14:38:16

MQTT(Message Queueing Telemetry Transport) 消息队列遥测传输，在物联网领域应用的很广泛，它是基于Publish/Subscribe模式，具有简单易用，支持QoS，传输效率高的特点。

它被设计用于低带宽，不稳定或高延迟的网络环境，因此非常适合于设备之间的数据通信。

EMQX提供了MQTT的服务器，并且可以在后台网页查看面板，还支持中文显示。

下载链接：Directory listing for EMQX: / | EMQ

由于5.0之后的版本不再支持Windows所以使用的是4.0版本的包，在下载完压缩包后，不用安装，进入cmd导航到安装的bin目录下（注意：路径中不能包含中文），执行命令：emqx start，看见没有报错就说明启动成功了。

之后在浏览器里输入：http://127.0.0.1:18083 进入面板。

在WebSocket菜单里可以模拟发布/订阅的操作，接下来我们将使用C#完成这一系列的操作。

1、连接主机

首先新建一个WPF项目，然后在Nuget中下载MQTTnet。

    // 连接主机MqttFactory factory = new MqttFactory();_client = factory.CreateMqttClient();var options = new MqttClientOptionsBuilder().WithTcpServer(this.ipAddress.Text, Convert.ToInt32(this.port.Text)).WithClientId(this.clientId.Text).Build();var result = await _client.ConnectAsync(options, CancellationToken.None);if (result.ResultCode == MqttClientConnectResultCode.Success){this.log.Text = DateTime.Now.ToString() + "    连接成功" + Environment.NewLine + this.log.Text;}else{this.log.Text = DateTime.Now.ToString() + $"    连接失败，{result.ReasonString}" + Environment.NewLine + this.log.Text;return;}

上述使用的是TCP的方式进行连接，需要主机地址，端口号，客户编号（一个用于区分用户的字符串）。

2、订阅消息

订阅消息分为两块，一个是消息的回显，一个是订阅消息。

    // 订阅消息var option = new MqttClientSubscribeOptions();MqttQualityOfServiceLevel level;switch (this.subscribeQos.SelectedIndex){case 0:level = MQTTnet.Protocol.MqttQualityOfServiceLevel.AtMostOnce;break;case 1:level = MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce;break;case 2:level = MQTTnet.Protocol.MqttQualityOfServiceLevel.ExactlyOnce;break;default:throw new Exception("请选择服务质量");}option.TopicFilters = new List<MqttTopicFilter>(){new MqttTopicFilter(){Topic = this.subscribeTopic.Text,QualityOfServiceLevel = level}};this._client.SubscribeAsync(option, CancellationToken.None);// 将订阅的消息回显到日志区this._client.ApplicationMessageReceivedAsync += e =>{var task = Task.Factory.StartNew(() => {try{var array = e.ApplicationMessage.PayloadSegment.Array;if (array == null){return;}var str = Encoding.UTF8.GetString(array);// 跨线程更新UIApplication.Current.Dispatcher.Invoke(() => {this.log.Text = DateTime.Now.ToString() + "    收到消息：" + str + Environment.NewLine + this.log.Text;});}catch (Exception ex){this.log.Text = DateTime.Now.ToString() + $"    {ex.Message}" + Environment.NewLine + this.log.Text;}});return task;};this.log.Text = DateTime.Now.ToString() + "    订阅成功" + Environment.NewLine + this.log.Text;

订阅消息只需要两个参数：主题Topic和服务质量QoC，主题是用来区分不同频段的消息，避免出现冲突，如果想接收到所有的消息可以这么写：topicXXX/#，#就代表不限制范围，如果打算只接受固定区域的消息，则需要将#改成某个字符串。

服务质量QoC是用来控制可用性的，0是最低等级，最多只发送一次，1是中级，至少发一次，但有可能出现重复接收的情况，2是最高级，只发一次，不会多也不会少。

将消息回显需要注册ApplicationMessageReceivedAsync事件，传入的参数是回显对象，返回值是一个Task类型，是在Task中获取回显的值并完成控件的更新操作。

3、发布消息

发布消息的参数比订阅多两个：消息内容Payload，持久会话（在恢复连接后保留之前的订阅和消息传递状态）

    var msg = new MqttApplicationMessage();msg.Topic = this.topic.Text;msg.PayloadSegment = Encoding.UTF8.GetBytes(this.msg.Text);msg.Retain = isSave.IsChecked??false;MqttQualityOfServiceLevel level;switch (this.publishQos.SelectedIndex){case 0:level = MQTTnet.Protocol.MqttQualityOfServiceLevel.AtMostOnce;break;case 1:level = MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce;break;case 2:level = MQTTnet.Protocol.MqttQualityOfServiceLevel.ExactlyOnce;break;default:throw new Exception("请选择服务质量");}msg.QualityOfServiceLevel = level;var resultPublish = await _client.PublishAsync(msg, CancellationToken.None);if (resultPublish.IsSuccess == true){this.log.Text = DateTime.Now.ToString() + "    发送成功" + Environment.NewLine + this.log.Text;}else{this.log.Text = DateTime.Now.ToString() + "    发送失败" + Environment.NewLine + this.log.Text;}