C# 基于.NET Framework框架WPF应用程序-MQTTNet库实现MQTT消息订阅发布

MQTT简述

mqtt官网

MQTTNet简述

MQTTnet
MQTTnet 是一个强大的开源 MQTT 客户端库，适用于 C# 平台。它提供了丰富的功能和灵活的 API，可以轻松地构建基于 MQTT 协议的应用程序。

创建项目（基于.NET Framework框架）

安装MQTTNet库

• NuGet包管理器 搜索MQTTNet
  

项目源码

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Windows;
using System.Windows.Controls;
using System.Windows.Data;
using System.Windows.Documents;
using System.Windows.Input;
using System.Windows.Media;
using System.Windows.Media.Imaging;
using System.Windows.Navigation;
using System.Windows.Shapes;


using MQTTnet;
using MQTTnet.Client;
using MQTTnet.Server;
using MQTTnet.Adapter;
using MQTTnet.Protocol;
using MQTTnet.Formatter;
using MQTTnet.Extensions;

namespace MQTTNetDemo
{
    /// <summary>
    /// MainWindow.xaml 的交互逻辑
    /// </summary>
    public partial class MainWindow : Window
    {
        public string STR_MQTT_HOST = "mqtt borker host"; // MQTT borker address
        public string STR_MQTT_USERNAME = "your username";
        public string STR_MQTT_PASSWORD = "your password";
        public string STR_MQTT_CLINETID = "mqtt clientid";
        public string STR_MQTT_SUBTOPIC = "topic_device"; // 订阅Topic
        public string STR_MQTT_PUBTOPIC = "topic_server"; // 发布Topic

        private MqttFactory mqttFactory = new MqttFactory();    // 创建MQTT工厂实例
        private IMqttClient mqttClient = null;  // 定义MqttClient接口

        public MainWindow()
        {
            InitializeComponent();
        }

        /// <summary>
        /// 连接MQTT服务
        /// </summary>
        /// <returns></returns>
        private async Task UserMqttConnect()
        { 
            mqttClient = mqttFactory.CreateMqttClient();
            var mqttClientOptions = new MqttClientOptionsBuilder()
                .WithTcpServer(STR_MQTT_HOST, 1883)
                .WithCredentials(STR_MQTT_USERNAME, STR_MQTT_PASSWORD)
                .WithClientId(STR_MQTT_CLINETID)
                .WithCleanSession(true)
                .Build();
            var response = await mqttClient.ConnectAsync(mqttClientOptions);
            Console.WriteLine("The MQTT Client is connected.");
            Console.WriteLine(response.ToString());
        }

        /// <summary>
        /// 断开MQTT服务
        /// </summary>
        /// <returns></returns>
        private async Task UserMqttDisconnect()
        {
            await mqttClient.DisconnectAsync();
            mqttClient.Dispose();
            Console.WriteLine("The MQTT Client is Disconnected.");            
        }

        /// <summary>
        /// MQTT发布消息
        /// </summary>
        /// <param name="payload">消息内容</param>
        /// <returns></returns>
        public async Task UserMqttPublishMessage(string payload)
        {
            if (mqttClient != null && mqttClient.IsConnected == true)
            {
                var publishPacket = new MqttApplicationMessageBuilder()
                    .WithTopic(STR_MQTT_PUBTOPIC)
                    .WithPayload(payload)
                    .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.ExactlyOnce)
                    .Build();

                //MqttQualityOfServiceLevel.AtMostOnce;
                //MqttQualityOfServiceLevel.AtLeastOnce
                //MqttQualityOfServiceLevel.ExactlyOnce;

                await mqttClient.PublishAsync(publishPacket);
            }
            else
            {
                MessageBox.Show("MQTT Client is Disconnected!");
            }
        }

        /// <summary>
        /// MQTT订阅消息
        /// </summary>
        /// <returns></returns>
        private async Task UserMqttSubscribeMessage()
        {
            await mqttClient.SubscribeAsync(new MqttTopicFilterBuilder()
                .WithTopic(STR_MQTT_SUBTOPIC)
                .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce)
                .Build());

            mqttClient.ApplicationMessageReceivedAsync += delegate (MqttApplicationMessageReceivedEventArgs args)
            {
                // TODO: 解析订阅消息payload内容
                var payload = Encoding.UTF8.GetString(args.ApplicationMessage.PayloadSegment.Array);
                Console.WriteLine("Recv Message:" + payload);
                Console.WriteLine("Recv ClientID:" + args.ClientId);
                Console.WriteLine("Recv Topic:" + args.ApplicationMessage.Topic);
                Console.WriteLine("Recv Retain:" + args.ApplicationMessage.Retain); // 指示消息是否为保留消息
                Console.WriteLine("Recv QOS:" + args.ApplicationMessage.QualityOfServiceLevel);

                // Now respond to the broker with a reason code other than success.
                args.ReasonCode = MqttApplicationMessageReceivedReasonCode.ImplementationSpecificError;
                args.ResponseReasonString = "That did not work!";

                // User properties require MQTT V5!
                args.ResponseUserProperties.Add(new MQTTnet.Packets.MqttUserProperty("My", "Data"));


                // Now the broker will resend the message again;
                return Task.CompletedTask;
            };

            Console.WriteLine("The MQTT is subscribe ...\n");
        }

        private async void btnMqttConnect_Click(object sender, RoutedEventArgs e)
        {
            await UserMqttConnect();
        }

        private async void btnMqttDisconnect_Click(object sender, RoutedEventArgs e)
        {
            await UserMqttDisconnect();
        }

        private async void btnMqttSubscribe_Click(object sender, RoutedEventArgs e)
        {
            await UserMqttSubscribeMessage();
        }

        private async void btnMqttPublish_Click(object sender, RoutedEventArgs e)
        {
            if ("" != textBoxMqttPublishMessage.Text)
            {
                string payload = textBoxMqttPublishMessage.Text.ToString();
                await UserMqttPublishMessage(payload);
            }
            else
            {
                MessageBox.Show("发布消息内容为空！");
            }            
        }
    }
}

运行效果