前面说到 MQTT技术介绍

现在物联网的概念很火，就工业智能互联来说，水平方向上如何把流水线式的孤岛式机台联动起来，我们比较有经验，但是垂直方向上，如何做数采，或者说如何高效灵活的做数采，需要补课的东西还有很多。MQTT是IBM很早以前就提出来的协议，但很可惜一直没有接触过，新公司的项目上引用了MQTTnet的开源库

官网指路：https://mqtt.org/

MQTTnet

MQTTnet是基于MQTT通信的高性能.NET库，它提供了一个MQTT客户端和一个MQTT服务器（代理）。支持.net core，支持MQTT 3.X和5.0版本。

https://github.com/chkr1011/MQTTnet
MQTTnet的Git路径。

本Demo设计为一个WPF程序。基于MQTTnet，实现了一个MQTT Server或者说Broker的创建，同时在窗体上提供了MQTT Client的创建功能，MQTT Client跟Server连接之后，通过点击按钮，实现主题订阅、发布的基础功能。

并可以通过MQTTX工具调试。

实例化MqttServer以及委托实现MqttServer的各个事件

mqttServer = new MqttFactory().CreateMqttServer() as MqttServer;
mqttServer.StartedHandler = new MqttServerStartedHandlerDelegate(OnMqttServerStarted);
mqttServer.StoppedHandler = new MqttServerStoppedHandlerDelegate(OnMqttServerStopped);
下面是完成的实现代码：
public MqttServer mqttServer = null;
public async void StartMqttServer()
if (mqttServer == null)
var config = ReadConfiguration();
var optionsBuilder = new MqttServerOptionsBuilder()
.WithDefaultEndpoint().WithDefaultEndpointPort(int.Parse(config["Port"].ToString())).WithConnectionValidator(
var currentUser = config["Users"][0]["UserName"].ToString();
var currentPWD = config["Users"][0]["Password"].ToString();
if (currentUser == null || currentPWD == null)
c.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword;
return;
if (c.Username != currentUser)
c.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword;
return;
if (c.Password != currentPWD)
c.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword;
return;
c.ReasonCode = MqttConnectReasonCode.Success;
}).WithSubscriptionInterceptor(
c.AcceptSubscription = true;
}).WithApplicationMessageInterceptor(
c.AcceptPublish = true;
mqttServer = new MqttFactory().CreateMqttServer() as MqttServer;
mqttServer.StartedHandler = new MqttServerStartedHandlerDelegate(OnMqttServerStarted);
mqttServer.StoppedHandler = new MqttServerStoppedHandlerDelegate(OnMqttServerStopped);
mqttServer.ClientConnectedHandler = new MqttServerClientConnectedHandlerDelegate(OnMqttServerClientConnected);
mqttServer.ClientDisconnectedHandler = new MqttServerClientDisconnectedHandlerDelegate(OnMqttServerClientDisconnected);
mqttServer.ClientSubscribedTopicHandler = new MqttServerClientSubscribedHandlerDelegate(OnMqttServerClientSubscribedTopic);
mqttServer.ClientUnsubscribedTopicHandler = new MqttServerClientUnsubscribedTopicHandlerDelegate(OnMqttServerClientUnsubscribedTopic);
mqttServer.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(OnMqttServer_ApplicationMessageReceived);
await mqttServer.StartAsync(optionsBuilder.Build());
lbxMonitor.BeginInvoke(_updateMonitorAction,
Logger.TraceLog(Logger.Level.Info, "MQTT Server is started."));
catch (Exception ex)
lbxMonitor.BeginInvoke(_updateMonitorAction,
Logger.TraceLog(Logger.Level.Fatal, $"MQTT Server start fail.>{ex.Message}"));
public async void StopMqttServer()
if (mqttServer == null) return;
await mqttServer?.StopAsync();
mqttServer = null;
lbxMonitor.BeginInvoke(_updateMonitorAction,
Logger.TraceLog(Logger.Level.Info, "MQTT Server is stopped."));
catch (Exception ex)
lbxMonitor.BeginInvoke(_updateMonitorAction,
Logger.TraceLog(Logger.Level.Fatal, $"MQTT Server stop fail.>{ex.Message}"));
MQTT Server/Broker发布消息
从MQTT的设计来看，服务端是代理的角色，订阅者和发布者是客户端，所以通常来说，消息的订阅与发布应当都是客户端干的事。但是，服务端自然也是可以参与一下发布者的角色的。我还没想到这样用的实际场景，先功能实现一下——
很粗暴，先实例化一个MqttApplicationMessage对象，然后作为传参调用MqttServer.PublishAsync进行消息发布。
public async void ServerPublishMqttTopic(string topic, string payload)
var message = new MqttApplicationMessage()
Topic = topic,
Payload = Encoding.UTF8.GetBytes(payload)
await mqttServer.PublishAsync(message);
lbxMonitor.BeginInvoke(_updateMonitorAction,
Logger.TraceLog(Logger.Level.Info, string.Format("MQTT Broker发布主题[{0}]成功！", topic)));

当然啦，MqttApplicationMessage还有很多属性值可以传，QoS(MqttQualityOfServiceLevel)、Retain等等，这里只是为了先实现功能，就只传了最简单的Topic和Payload。
MQTT Client创建

整体的实现思路跟Server端如出一辙，声明一个MqttClientOptions，赋值各种连接Server端需要的参数，最后作为MqttClient.ConnectAsync的传参，连接Server。
这里我在MqttClientOptions里面尝试了一下WillMessage，其实就是一个MqttApplicationMessage对象，WillMessage作为遗言机制，用于Client跟Server端挂点时的广播通知。
再之后就是实现MqttClient的各个事件，用来客制Connected，Disconnected，MessageReceived的各种逻辑，跟Server端实现没有什么区别，不再赘述。


private MqttClient mqttClient = null;
private async Task ClientStart()
var tcpServer = txtIPAddr.Text;
var tcpPort = int.Parse(txtPort.Text.Trim());
var mqttUser = txtUserName.Text.Trim();
var mqttPassword = txtPWD.Text.Trim();
var mqttFactory = new MqttFactory();
var options = new MqttClientOptions
ClientId = txtClientID.Text.Trim(),
ProtocolVersion = MQTTnet.Formatter.MqttProtocolVersion.V311,
ChannelOptions = new MqttClientTcpOptions
Server = tcpServer,
Port = tcpPort
WillDelayInterval = 10,
WillMessage = new MqttApplicationMessage()
Topic = $"LastWill/{txtClientID.Text.Trim()}",
Payload= Encoding.UTF8.GetBytes("I Lost the connection!"),
QualityOfServiceLevel = MqttQualityOfServiceLevel.ExactlyOnce
if (options.ChannelOptions == null)
throw new InvalidOperationException();
if (!string.IsNullOrEmpty(mqttUser))
options.Credentials = new MqttClientCredentials
Username = mqttUser,
Password = Encoding.UTF8.GetBytes(mqttPassword)
options.CleanSession = true;
options.KeepAlivePeriod = TimeSpan.FromSeconds(5);
mqttClient = mqttFactory.CreateMqttClient() as MqttClient;
mqttClient.ConnectedHandler = new MqttClientConnectedHandlerDelegate(OnMqttClientConnected);
mqttClient.DisconnectedHandler = new MqttClientDisconnectedHandlerDelegate(OnMqttClientDisConnected);
mqttClient.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(OnSubscriberMessageReceived);
await mqttClient.ConnectAsync(options);
lbxMonitor.BeginInvoke(_updateMonitorAction,
Logger.TraceLog(Logger.Level.Info, $"客户端[{options.ClientId}]尝试连接..."));
catch (Exception ex)
lbxMonitor.BeginInvoke(_updateMonitorAction,
Logger.TraceLog(Logger.Level.Fatal, $"客户端尝试连接出错.>{ex.Message}"));
private async Task ClientStop()
if (mqttClient == null) return;
await mqttClient.DisconnectAsync();
mqttClient = null;
catch (Exception ex)
lbxMonitor.BeginInvoke(_updateMonitorAction,
Logger.TraceLog(Logger.Level.Fatal, $"客户端尝试断开Server出错.>{ex.Message}"));
MQTT Client发布消息
这里的实现逻辑跟写法和Server端的发布别无二致，我在这里的MqttApplicationMessage补上了QoS和Retain的设置，由Form页面的控件传参。
这里补一句关于Retain的用法：Retain意为保留，设为True表示这条消息发布的时候如果没有订阅者，则该消息保留在Server端，直到被人订阅时立刻发布出去并删除，设为False时则没有这样的效果。
public async void ClientPublishMqttTopic(string topic, string payload)
var message = new MqttApplicationMessage()
Topic = topic,
Payload = Encoding.UTF8.GetBytes(payload),
QualityOfServiceLevel = (MqttQualityOfServiceLevel)cmbQos.SelectedIndex,
Retain = bool.Parse(cmbRetain.SelectedItem.ToString())
await mqttClient.PublishAsync(message);
lbxMonitor.BeginInvoke(_updateMonitorAction,
Logger.TraceLog(Logger.Level.Info, string.Format("客户端[{0}]发布主题[{1}]成功！", mqttClient.Options.ClientId, topic)));
catch (Exception ex)
lbxMonitor.BeginInvoke(_updateMonitorAction,
Logger.TraceLog(Logger.Level.Fatal, string.Format("客户端[{0}]发布主题[{1}]异常！>{2}", mqttClient.Options.ClientId, topic,ex.Message)));

MQTT Clien订阅消息
呼叫MqttClient.SubscribeAsync，传入消息主题即可。
public async void ClientSubscribeTopic(string topic)
await mqttClient.SubscribeAsync(topic);
lbxMonitor.BeginInvoke(_updateMonitorAction,
Logger.TraceLog(Logger.Level.Info, string.Format("客户端[{0}]订阅主题[{1}]成功！", mqttClient.Options.ClientId, topic)));

MQTT Clien取消订阅消息
呼叫MqttClient.UnsubscribeAsync，取消消息订阅。
public async void ClientUnSubscribeTopic(string topic)
await mqttClient.UnsubscribeAsync(topic);
lbxMonitor.BeginInvoke(_updateMonitorAction,
Logger.TraceLog(Logger.Level.Info, string.Format("客户端[{0}]取消主题[{1}]成功！", mqttClient.Options.ClientId, topic)));
以上代码是参考
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 MahApps.Metro.Controls;
using System.ComponentModel;
using MQTTnet.Server;
using MQTTnet.Adapter;
using MQTTnet.Protocol;
using MQTTnet;
using System.Collections.ObjectModel;
namespace MqttDemo.WPFServer
    /// <summary>
    /// Interaction logic for MainWindow.xaml
    /// </summary>
    public partial class MainWindow : MetroWindow
        private MainWindowModel _model;
        private IMqttServer server;
        public MainWindow()
            InitializeComponent();
            _model = new MainWindowModel();
            this.DataContext = _model;
        #region 启动按钮事件
        private async void btnStart_Click(object sender, RoutedEventArgs e)
            var optionBuilder = new MqttServerOptionsBuilder().WithDefaultEndpointBoundIPAddress(System.Net.IPAddress.Parse(_model.HostIP)).WithDefaultEndpointPort(_model.HostPort).WithDefaultCommunicationTimeout(TimeSpan.FromMilliseconds(_model.Timeout)).WithConnectionValidator(t =>
                if (t.Username!=_model.UserName||t.Password!=_model.Password)
                    t.ReturnCode = MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword;
                t.ReturnCode = MqttConnectReturnCode.ConnectionAccepted;
            var option = optionBuilder.Build();
            server = new MqttFactory().CreateMqttServer();
            server.ApplicationMessageReceived += Server_ApplicationMessageReceived;//绑定消息接收事件
            server.ClientConnected += Server_ClientConnected;//绑定客户端连接事件
            server.ClientDisconnected += Server_ClientDisconnected;//绑定客户端断开事件
            server.ClientSubscribedTopic += Server_ClientSubscribedTopic;//绑定客户端订阅主题事件
            server.ClientUnsubscribedTopic += Server_ClientUnsubscribedTopic;//绑定客户端退订主题事件
            server.Started += Server_Started;//绑定服务端启动事件
            server.Stopped += Server_Stopped;//绑定服务端停止事件
            await server.StartAsync(option);
        #endregion
        #region 停止按钮事件
        private async void btnStop_Click(object sender, RoutedEventArgs e)
            if (server != null)
                await server.StopAsync();
        #endregion
        #region 服务端停止事件
        private void Server_Stopped(object sender, EventArgs e)
            WriteToStatus("服务端已停止！");
        #endregion
        #region 服务端启动事件
        private void Server_Started(object sender, EventArgs e)
            WriteToStatus("服务端已启动！");
        #endregion
        #region 客户端退订主题事件
        private void Server_ClientUnsubscribedTopic(object sender, MqttClientUnsubscribedTopicEventArgs e)
            this.Dispatcher.Invoke(() =>
                if (_model.AllTopics.Any(t => t.Topic == e.TopicFilter))
                    TopicModel model = _model.AllTopics.First(t => t.Topic == e.TopicFilter);
                    _model.AllTopics.Remove(model);
                    model.Clients.Remove(e.ClientId);
                    model.Count--;
                    if (model.Count > 0)
                        _model.AllTopics.Add(model);
            WriteToStatus("客户端" + e.ClientId + "退订主题" + e.TopicFilter);
        #endregion
        #region 客户端订阅主题事件
        private void Server_ClientSubscribedTopic(object sender, MqttClientSubscribedTopicEventArgs e)
            this.Dispatcher.Invoke(() =>
                if (_model.AllTopics.Any(t => t.Topic == e.TopicFilter.Topic))
                    TopicModel model = _model.AllTopics.First(t => t.Topic == e.TopicFilter.Topic);
                    _model.AllTopics.Remove(model);
                    model.Clients.Add(e.ClientId);
                    model.Count++;
                    _model.AllTopics.Add(model);
                    TopicModel model = new TopicModel(e.TopicFilter.Topic, e.TopicFilter.QualityOfServiceLevel)
                        Clients = new List<string> { e.ClientId },
                        Count = 1
                    _model.AllTopics.Add(model);
            WriteToStatus("客户端" + e.ClientId + "订阅主题" + e.TopicFilter.Topic);
        #endregion
        #region 客户端断开事件
        private void Server_ClientDisconnected(object sender, MqttClientDisconnectedEventArgs e)
            this.Dispatcher.Invoke(() =>
                _model.AllClients.Remove(e.ClientId);
                var query = _model.AllTopics.Where(t => t.Clients.Contains(e.ClientId));
                if (query.Any())
                    var tmp = query.ToList();
                    foreach (var model in tmp)
                        _model.AllTopics.Remove(model);
                        model.Clients.Remove(e.ClientId);
                        model.Count--;
                        _model.AllTopics.Add(model);
            WriteToStatus("客户端" + e.ClientId + "断开");
        #endregion
        #region 客户端连接事件
        private void Server_ClientConnected(object sender, MqttClientConnectedEventArgs e)
            this.Dispatcher.Invoke(() =>
                _model.AllClients.Add(e.ClientId);
            WriteToStatus("客户端" + e.ClientId + "连接");
        #endregion
        #region 收到消息事件
        private void Server_ApplicationMessageReceived(object sender, MqttApplicationMessageReceivedEventArgs e)
            if (e.ApplicationMessage.Topic == "/environ/temp")
                string str = System.Text.Encoding.UTF8.GetString(e.ApplicationMessage.Payload);
                double tmp;
                bool isdouble = double.TryParse(str, out tmp);
                if (isdouble)
                    string result = "";
                    if (tmp > 40)
                        result = "温度过高！";
                    else if (tmp < 10)
                        result = "温度过低！";
                        result = "温度正常！";
                    MqttApplicationMessage message = new MqttApplicationMessage()
                        Topic = e.ApplicationMessage.Topic,
                        Payload = Encoding.UTF8.GetBytes(result),
                        QualityOfServiceLevel = e.ApplicationMessage.QualityOfServiceLevel,
                        Retain = e.ApplicationMessage.Retain
                    server.PublishAsync(message);
            WriteToStatus("收到消息" + e.ApplicationMessage.ConvertPayloadToString() + ",来自客户端" + e.ClientId + ",主题为" + e.ApplicationMessage.Topic);
        #endregion
        #region 打开配置页面
        private void btnConfig_Click(object sender, RoutedEventArgs e)
            this.flyConfig.IsOpen = !this.flyConfig.IsOpen;
        #endregion
        #region 增加主题按钮事件
        private void btnAddTopic_Click(object sender, RoutedEventArgs e)
            if (!string.IsNullOrWhiteSpace(_model.AddTopic)&&server!=null)
                TopicModel topic = new TopicModel(_model.AddTopic, MqttQualityOfServiceLevel.AtLeastOnce);
                foreach (string clientId in _model.AllClients)
                    server.SubscribeAsync(clientId, new List<TopicFilter> { topic });
                _model.AllTopics.Add(topic);
        #endregion
        #region 清理内容
        private void menuClear_Click(object sender, RoutedEventArgs e)
            txtRich.Document.Blocks.Clear();
        #endregion
        #region 状态输出
        /// <summary>
        /// 状态输出
        /// </summary>
        /// <param name="message"></param>
        public void WriteToStatus(string message)
            if (!(txtRich.CheckAccess()))
                this.Dispatcher.Invoke(() =>
                    WriteToStatus(message)
                return;
            string strTime = "[" + System.DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss") + "] ";
            txtRich.AppendText(strTime + message + "\r");
        #endregion
    public class MainWindowModel : INotifyPropertyChanged
        public MainWindowModel()
            hostIP = "127.0.0.1";//绑定的IP地址
            hostPort = 61613;//绑定的端口号12345
            timeout = 3000;//连接超时时间
            username = "admin";//用户名
            password = "password";//密码
            allTopics = new ObservableCollection<TopicModel>();//主题
            allClients = new ObservableCollection<string>();//客户端
            addTopic = "";
        private ObservableCollection<TopicModel> allTopics;
        public ObservableCollection<TopicModel> AllTopics
            get { return allTopics; }
                if (allTopics != value)
                    allTopics = value;
                    this.OnPropertyChanged("AllTopics");
        private ObservableCollection<string> allClients;
        public ObservableCollection<string> AllClients
            get { return allClients; }
                if (allClients != value)
                    allClients = value;
                    this.OnPropertyChanged("AllClients");
        private string hostIP;
        public string HostIP
            get { return hostIP; }
                if (hostIP != value)
                    hostIP = value;
                    this.OnPropertyChanged("HostIP");
        private int hostPort;
        public int HostPort
            get { return hostPort; }
                if (hostPort != value)
                    hostPort = value;
                    this.OnPropertyChanged("HostPort");
        private int timeout;
        public int Timeout
            get { return timeout; }
                if (timeout != value)
                    timeout = value;
                    this.OnPropertyChanged("Timeout");
        private string username;
        public string UserName
            get { return username; }
                if (username != value)
                    username = value;
                    this.OnPropertyChanged("UserName");
        private string password;
        public string Password
            get { return password; }
                if (password != value)
                    password = value;
                    this.OnPropertyChanged("Password");
        private string addTopic;
        public string AddTopic
            get { return addTopic; }
                if (addTopic != value)
                    addTopic = value;
                    this.OnPropertyChanged("AddTopic");
        public event PropertyChangedEventHandler PropertyChanged;
        protected virtual void OnPropertyChanged(string propertyName = null)
            var handler = PropertyChanged;
            if (handler != null) handler(this, new PropertyChangedEventArgs(propertyName));
    public class TopicModel : TopicFilter,INotifyPropertyChanged
        public TopicModel(string topic, MqttQualityOfServiceLevel qualityOfServiceLevel) : base(topic, qualityOfServiceLevel)
            clients = new List<string>();
            count = 0;
        private int count;
        /// <summary>
        /// 订阅此主题的客户端数量
        /// </summary>
        public int Count
            get { return count; }
                if (count != value)
                    count = value;
                    this.OnPropertyChanged("Count");
        private List<string> clients;
        /// <summary>
        /// 订阅此主题的客户端
        /// </summary>
        public List<string> Clients
            get { return clients; }
                if (clients != value)
                    clients = value;
                    this.OnPropertyChanged("Clients");
        protected virtual void OnPropertyChanged(string propertyName = null)
            var handler = PropertyChanged;
            if (handler != null) handler(this, new PropertyChangedEventArgs(propertyName));
        public event PropertyChangedEventHandler PropertyChanged;
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 MahApps.Metro.Controls;
using MQTTnet;
using MQTTnet.Client;
using System.ComponentModel;
using MahApps.Metro.Controls.Dialogs;
namespace MqttDemo.MetroClient
    /// <summary>
    /// Interaction logic for MainWindow.xaml
    /// </summary>
    public partial class MainWindow : MetroWindow
        private MainWindowModel _model;
        private IMqttClient _client;
        public MainWindow()
            InitializeComponent();
            _model = new MainWindowModel()
                AllTopics = InitTopics(),
                SelectedTopics = new List<TopicFilter>(),
                ServerUri = "127.0.0.1",
                CurrentTopic = null,
                ServerPort=61613,
                ClientID = Guid.NewGuid().ToString("N")
            this.DataContext = _model;
        #region 订阅主题面板
        /// <summary>
        /// 打开订阅主题面板
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void btnSub_Click(object sender, RoutedEventArgs e)
            this.flySub.IsOpen = !this.flySub.IsOpen;
        #endregion
        #region 用户配置面板
        private void btnLogin_Click(object sender, RoutedEventArgs e)
            this.flyLogin.IsOpen = !this.flyLogin.IsOpen;
        #endregion
        #region 数据初始化
        /// <summary>
        /// 数据初始化
        /// </summary>
        /// <returns></returns>
        private List<TopicModel> InitTopics()
            List<TopicModel> topics = new List<TopicModel>();
            topics.Add(new TopicModel("/environ/temp", "环境-温度"));
            topics.Add(new TopicModel("/environ/hum", "环境-湿度"));
            //topics.Add(new TopicModel("/environ/pm25", "环境-PM2.5"));
            //topics.Add(new TopicModel("/environ/CO2", "环境-二氧化碳"));
            //topics.Add(new TopicModel("/energy/electric", "能耗-电"));
            //topics.Add(new TopicModel("/energy/water", "环境-水"));
            //topics.Add(new TopicModel("/energy/gas", "环境-电"));
            topics.Add(new TopicModel("/data/alarm", "数据-报警"));
            topics.Add(new TopicModel("/data/message", "数据-消息"));
            topics.Add(new TopicModel("/data/notify", "数据-通知"));
            return topics;
        /// <summary>
        /// 数据模型转换
        /// </summary>
        /// <param name="topics"></param>
        /// <returns></returns>
        private List<TopicFilter> ConvertTopics(List<TopicModel> topics)
            //MQTTnet.TopicFilter
            List<TopicFilter> filters = new List<TopicFilter>();
            foreach (TopicModel model in topics)
                TopicFilter filter = new TopicFilter(model.Topic,MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce);
                filters.Add(filter);
            return filters;
        #endregion
        #region 状态输出
        /// <summary>
        /// 状态输出
        /// </summary>
        /// <param name="message"></param>
        public void WriteToStatus(string message)
            if (!(txtRich.CheckAccess()))
                this.Dispatcher.Invoke(() =>
                    WriteToStatus(message)
                return;
            string strTime = "[" + System.DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss") + "] ";
            txtRich.AppendText(strTime + message + "\r");
            if (txtRich.ExtentHeight > 200)
                txtRich.Document.Blocks.Clear();
        #endregion
        #region 更改订阅的主题
        /// <summary>
        /// 保存订阅的主题
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void btnSave_Click(object sender, RoutedEventArgs e)
            List<TopicModel> topics = _model.AllTopics.Where(t => t.IsSelected == true).ToList();
            _model.SelectedTopics = ConvertTopics(topics);
            this.flySub.IsOpen = !this.flySub.IsOpen;
            SubscribeTopics(_model.SelectedTopics);
        #endregion
        #region 订阅主题
        private void SubscribeTopics(List<TopicFilter> filters)
            if (_client!=null)
                _client.SubscribeAsync(filters);
                string tmp = "";
                foreach (var filter in filters)
                    tmp += filter.Topic;
                    tmp += ",";
                if (tmp.Length>1)
                    tmp = tmp.Substring(0, tmp.Length - 1);
                WriteToStatus("成功订阅主题："+tmp);
                ShowDialog("提示", "请连接服务端后订阅主题！");
        #endregion
        #region 连接/断开服务端
        /// <summary>
        /// 连接服务端
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void btnStart_Click(object sender, RoutedEventArgs e)
            if (_model.ServerUri!=null&&_model.ServerPort>0)
                InitClient(_model.ClientID, _model.ServerUri, _model.ServerPort);
                ShowDialog("提示", "服务端地址或端口号不能为空！");
        /// <summary>
        /// 断开服务端
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void btnStop_Click(object sender, RoutedEventArgs e)
            if (_client != null)
                _client.DisconnectAsync();
        #endregion
        #region MQTT方法
        /// <summary>
        /// 初始化
        /// </summary>
        /// <param name="id"></param>
        /// <param name="url"></param>
        /// <param name="port"></param>
        private void InitClient(string id,string url = "127.0.0.1", int port = 1883)
            var options = new MqttClientOptions()
                ClientId = id
            options.ChannelOptions = new MqttClientTcpOptions()
                Server = url,
                Port = port
            options.Credentials = new MqttClientCredentials()
                Username=_model.UserName,
                Password=_model.Password
            options.CleanSession = true;
            options.KeepAlivePeriod = TimeSpan.FromSeconds(100);
            options.KeepAliveSendInterval = TimeSpan.FromSeconds(10000);
            if (_client != null)
                _client.DisconnectAsync();
                _client = null;
            _client = new MQTTnet.MqttFactory().CreateMqttClient();
            _client.ApplicationMessageReceived += _client_ApplicationMessageReceived;
            _client.Connected += _client_Connected;
            _client.Disconnected += _client_Disconnected;
            _client.ConnectAsync(options);
        /// <summary>
        /// 客户端与服务端断开连接
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void _client_Disconnected(object sender, MqttClientDisconnectedEventArgs e)
            _model.IsConnected = false;
            _model.IsDisConnected = true;
            WriteToStatus("与服务端断开连接！");
        /// <summary>
        /// 客户端与服务端建立连接
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void _client_Connected(object sender, MqttClientConnectedEventArgs e)
            _model.IsConnected = true;
            _model.IsDisConnected = false;
            WriteToStatus("与服务端建立连接");
        /// <summary>
        /// 客户端收到消息
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void _client_ApplicationMessageReceived(object sender, MQTTnet.MqttApplicationMessageReceivedEventArgs e)
            WriteToStatus("收到来自客户端" + e.ClientId + "，主题为" + e.ApplicationMessage.Topic + "的消息：" + Encoding.UTF8.GetString(e.ApplicationMessage.Payload));
        #endregion
        #region 发布消息
        /// <summary>
        /// 发布主题
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void btnPublish_Click(object sender, RoutedEventArgs e)
            if (_client!=null)
                if (this.comboTopics.SelectedIndex<0)
                    ShowDialog("提示", "请选择要发布的主题！");
                    return;
                if (string.IsNullOrEmpty(txtContent.Text))
                    ShowDialog("提示", "消息内容不能为空！");
                    return;
                string topic = comboTopics.SelectedValue as string;
                string content = txtContent.Text;
                MqttApplicationMessage msg = new MqttApplicationMessage
                    Topic=topic,
                    Payload=Encoding.UTF8.GetBytes(content),
                    QualityOfServiceLevel=MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce,
                    Retain=false
                _client.PublishAsync(msg);
                WriteToStatus("成功发布主题为" + topic+"的消息！");
                ShowDialog("提示", "请连接服务端后发布消息！");
                return;
        #endregion
        #region 提示框
        public void ShowDialog(string title, string content)
            _showDialog(title, content);
        /// <summary>
        /// 提示框
        /// </summary>
        /// <param name="title"></param>
        /// <param name="content"></param>
        private void _showDialog(string title, string content)
            var mySetting = new MetroDialogSettings()
                AffirmativeButtonText = "确定",
                //NegativeButtonText = "Go away!",
                //FirstAuxiliaryButtonText = "Cancel",
                ColorScheme = this.MetroDialogOptions.ColorScheme
            MessageDialogResult result = this.ShowModalMessageExternal(title, content, MessageDialogStyle.Affirmative, mySetting);
        #endregion
        private void btnSaveConfig_Click(object sender, RoutedEventArgs e)
    public class MainWindowModel: INotifyPropertyChanged
        private List<TopicModel> _allTopics;
        public List<TopicModel> AllTopics
            get { return _allTopics; }
                if (_allTopics!=value)
                    _allTopics = value;
                    OnPropertyChanged("AllTopics");
        private List<TopicFilter> _selectedTopics;
        public List<TopicFilter> SelectedTopics
            get { return _selectedTopics; }
                if (_selectedTopics!=value)
                    _selectedTopics = value;
                    OnPropertyChanged("SelectedTopics");
        private string _serverUri;
        public string ServerUri
            get { return _serverUri; }
                if (_serverUri!=value)
                    _serverUri = value;
                    OnPropertyChanged("ServerUri");
        private int _serverPort;
        public int ServerPort
            get { return _serverPort; }
                if (_serverPort!=value)
                    _serverPort = value;
                    OnPropertyChanged("ServerPort");
        private string _clientId;
        public string ClientID
            get { return _clientId; }
                if (_clientId!=value)
                    _clientId = value;
                    OnPropertyChanged("ClientID");
        private TopicFilter _currentTopic;
        public TopicFilter CurrentTopic
            get { return _currentTopic; }
                if (_currentTopic!=value)
                    _currentTopic = value;
                    OnPropertyChanged("CurrentTopic");
        private bool? _isConnected=false;
        public bool? IsConnected
            get { return _isConnected; }
                if (_isConnected!=value)
                    _isConnected = value;
                    OnPropertyChanged("IsConnected");
        private bool _isDisConnected=true;
        public bool IsDisConnected
            get { return _isDisConnected; }
                if (_isDisConnected != value)
                    _isDisConnected = value;
                    this.OnPropertyChanged("IsDisConnected");
        private string _userName="admin";
        public string UserName
            get { return _userName; }
                if (_userName != value)
                    _userName = value;
                    this.OnPropertyChanged("UserName");
        private string _password="password";
        public string Password
            get { return _password; }
                if (_password != value)
                    _password = value;
                    this.OnPropertyChanged("Password");
        public event PropertyChangedEventHandler PropertyChanged;
        protected virtual void OnPropertyChanged(string propertyName = null)
            var handler = PropertyChanged;
            if (handler != null) handler(this, new PropertyChangedEventArgs(propertyName));
    public class TopicModel: INotifyPropertyChanged
        public TopicModel()
        public TopicModel(string topic,string describe)
            _isSelected = false;
            _topic = topic;
            _describe = describe;
        private bool? _isSelected;
        public bool? IsSelected
            get { return _isSelected; }
                if (_isSelected!=value)
                    _isSelected = value;
                    OnPropertyChanged("IsSelected");
        private string _topic;
        public string Topic
            get { return _topic; }
                if (_topic!=value)
                    _topic = value;
                    OnPropertyChanged("Topic");
        private string _describe;
        public string Describe
            get { return _describe; }
                if (_describe!=value)
                    _describe = value;
                    OnPropertyChanged("Describe");
        public event PropertyChangedEventHandler PropertyChanged;
        protected virtual void OnPropertyChanged(string propertyName = null)
            var handler = PropertyChanged;
            if (handler != null) handler(this, new PropertyChangedEventArgs(propertyName));
https://github.com/ludewig/MqttDemo
鸣谢：
https://blog.csdn.net/xiakexingtx/article/details/108281359
https://blog.csdn.net/lordwish/article/details/84970800?spm=1001.2101.3001.6661.1&utm_medium=distribute.pc_relevant_t0.none-task-download-2%7Edefault%7ECTRLIST%7EPaid-1-13762029-blog-84970800.pc_relevant_paycolumn_v3&depth_1-utm_source=distribute.pc_relevant_t0.none-task-download-2%7Edefault%7ECTRLIST%7EPaid-1-13762029-blog-84970800.pc_relevant_paycolumn_v3&utm_relevant_index=1
    
本文来自博客园，作者：{春光牛牛,yak}，转载请注明原文链接：https://www.cnblogs.com/yakniu/p/16423899.html
欢迎各位大佬们评论指正
QQ讨论群：610129902