一、什么是MQTT
MQTT协议由于其用极少的代码和有限的带宽,为连接远程设备提供实时可靠的消息服务,具有开销低、占用带宽低、即时通讯等优点,
使其在物联网、小型设备、移动应用等方面有较广泛的应用,在工业物联网中,MQTT也有广泛的应用。
主要有以下特点:
- 使用发布/订阅消息模式,提供一对多的消息发布
- 使用TCP/IP提供网络连接
- 小型传输,开销很小(固定长度的头部是 2 字节),协议交换最小化,以降低网络流量,传输的内容最大为256MB。
- 使用 Last Will 和 Testament 特性通知有关各方客户端异常中断的机制。
二、MQTT服务器设计
MQTT 服务端主要用于与多个客户端保持连接,并处理客户端的发布和订阅等逻辑,
多数情况下服务端都是转发主题匹配的客户端消息,在系统中起到一个中介的作用。
下面就是在项目中引入mqtt服务端和客户端:
引用nugut包
![](/i/l/?n=23&i=blog/1165395/202307/1165395-20230724135408500-1188871578.png)
服务代码:
using System; using System.Collections.Generic; using System.Text; using MQTTnet; using MQTTnet.Client.Receiving; using MQTTnet.Protocol; using MQTTnet.Server; namespace LR.Utils.MQTT { public class MqttServer { private MQTTnet.Server.MqttServer mqttServer = null; public delegate void MqttServerStarted(EventArgs obj) ; public MqttServerStarted MqttServerStartedHandler; public delegate void MqttServerStopped(EventArgs obj); public MqttServerStopped MqttServerStoppedHandler; public delegate void MqttServerClientConnected(MqttServerClientConnectedEventArgs obj); public MqttServerClientConnected MqttServerClientConnectedHandler; public delegate void MqttServerClientDisConnected(MqttServerClientDisconnectedEventArgs obj); public MqttServerClientDisConnected MqttServerClientDisConnectedHandler; public async void StartMqttServer(int port, string username, string pwd) { try { if (mqttServer == null) { var optionsBuilder = new MqttServerOptionsBuilder() .WithDefaultEndpoint().WithDefaultEndpointPort(port).WithConnectionValidator( c => { if (username == null || pwd == null) { c.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword; return; } if (c.Username != username) { c.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword; return; } if (c.Password != pwd) { c.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword; return; } c.ReasonCode = MqttConnectReasonCode.Success; }).WithSubscriptionInterceptor( c => { c.AcceptSubscription = true; }).WithApplicationMessageInterceptor( c => { c.AcceptPublish = true; }); mqttServer = new MqttFactory().CreateMqttServer() as MQTTnet.Server.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()); } } catch (Exception eX) { throw eX; } } public async void StopMqttServer() { if (mqttServer == null) return; try { await mqttServer?.StopAsync(); mqttServer = null; } catch (Exception eX) { throw eX; } } #region 事件 private void OnMqttServerStarted(EventArgs obj) { if (MqttServerStartedHandler!=null) { MqttServerStartedHandler(obj); } } private void OnMqttServerStopped(EventArgs obj) { if (MqttServerStoppedHandler != null) { MqttServerStoppedHandler(obj); } } private void OnMqttServerClientConnected(MqttServerClientConnectedEventArgs obj) { if (MqttServerClientConnectedHandler != null) { MqttServerClientConnectedHandler(obj); } } private void OnMqttServerClientDisconnected(MqttServerClientDisconnectedEventArgs obj) { if (MqttServerClientDisConnectedHandler != null) { MqttServerClientDisConnectedHandler(obj); } } public delegate void MqttServerClientUnsubscribedTopic(MqttServerClientUnsubscribedTopicEventArgs obj); public MqttServerClientUnsubscribedTopic MqttServerClientUnsubscribedTopicHandler; private void OnMqttServerClientUnsubscribedTopic(MqttServerClientUnsubscribedTopicEventArgs obj) { if (MqttServerClientUnsubscribedTopicHandler != null) { MqttServerClientUnsubscribedTopicHandler(obj); } } public delegate void MqttServerClientSubscribedTopic(MqttServerClientSubscribedTopicEventArgs obj); public MqttServerClientSubscribedTopic MqttServerClientSubscribedTopicHandler; private void OnMqttServerClientSubscribedTopic(MqttServerClientSubscribedTopicEventArgs obj) { if (MqttServerClientSubscribedTopicHandler != null) { MqttServerClientSubscribedTopicHandler(obj); } } public delegate void MqttServer_ApplicationMessageReceived(MqttApplicationMessageReceivedEventArgs obj); public MqttServer_ApplicationMessageReceived MqttServer_ApplicationMessageReceivedHandler; private void OnMqttServer_ApplicationMessageReceived(MqttApplicationMessageReceivedEventArgs obj) { if (MqttServer_ApplicationMessageReceivedHandler != null) { MqttServer_ApplicationMessageReceivedHandler(obj); } } #endregion } }
客户端代码
using MQTTnet; using MQTTnet.Client; using MQTTnet.Client.Connecting; using MQTTnet.Client.Disconnecting; using MQTTnet.Client.Options; using MQTTnet.Client.Receiving; using MQTTnet.Protocol; using System; using System.Collections.Generic; using System.Text; using System.Threading.Tasks; namespace LR.Utils.MQTT { public class MqttClient: IDisposable { private MQTTnet.Client.MqttClient mqttClient = null; public async Task ClientStart(string ip,int port,string username,string password) { try { var tcpServer = ip.Trim(); var tcpPort = port; var mqttUser = username.Trim(); var mqttPassword = password.Trim(); var clientID = Guid.NewGuid().ToString(); var mqttFactory = new MqttFactory(); var options = new MqttClientOptions { ClientId = Guid.NewGuid().ToString(), ProtocolVersion = MQTTnet.Formatter.MqttProtocolVersion.V311, ChannelOptions = new MqttClientTcpOptions { Server = tcpServer, Port = tcpPort }, WillDelayInterval = 10, WillMessage = new MqttApplicationMessage() { Topic = $"LastWill/{clientID}", 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 MQTTnet.Client.MqttClient; mqttClient.ConnectedHandler = new MqttClientConnectedHandlerDelegate(OnMqttClientConnected); mqttClient.DisconnectedHandler = new MqttClientDisconnectedHandlerDelegate(OnMqttClientDisConnected); mqttClient.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(OnSubscriberMessageReceived); await mqttClient.ConnectAsync(options); } catch (Exception ex) { throw ex; } } public async Task WSClientStart(string ip, int port, string username, string password) { try { var tcpServer = ip.Trim(); var tcpPort = port; var mqttUser = username.Trim(); var mqttPassword = password.Trim(); var clientID = Guid.NewGuid().ToString(); var mqttFactory = new MqttFactory(); var options = mqttFactory.CreateClientOptionsBuilder() .WithWebSocketServer("ws://" + ip + ":" + port + "/ws") .WithCredentials(username, password) .WithClientId(clientID) .Build(); mqttClient = mqttFactory.CreateMqttClient() as MQTTnet.Client.MqttClient; mqttClient.ConnectedHandler = new MqttClientConnectedHandlerDelegate(OnMqttClientConnected); mqttClient.DisconnectedHandler = new MqttClientDisconnectedHandlerDelegate(OnMqttClientDisConnected); mqttClient.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(OnSubscriberMessageReceived); await mqttClient.ConnectAsync(options); } catch (Exception ex) { throw ex; } } #region 事件 public delegate void MqttClientConnected(MqttClientConnectedEventArgs obj); public MqttClientConnected MqttClientConnectedHandler; public delegate void MqttClientDisConnected(MqttClientDisconnectedEventArgs obj); public MqttClientDisConnected MqttClientDisConnectedHandler; public delegate void SubscriberMessageReceived(MqttApplicationMessageReceivedEventArgs obj); public SubscriberMessageReceived SubscriberMessageReceivedHandler; private void OnSubscriberMessageReceived(MqttApplicationMessageReceivedEventArgs obj) { if (SubscriberMessageReceivedHandler != null) { SubscriberMessageReceivedHandler(obj); } } private void OnMqttClientDisConnected(MqttClientDisconnectedEventArgs obj) { if (MqttClientDisConnectedHandler != null) { MqttClientDisConnectedHandler(obj); } } private void OnMqttClientConnected(MqttClientConnectedEventArgs obj) { if (MqttClientConnectedHandler != null) { MqttClientConnectedHandler(obj); } } #endregion private async Task ClientStop() { try { if (mqttClient == null) return; await mqttClient.DisconnectAsync(); mqttClient = null; } catch (Exception ex) { throw ex; } } public async void ClientPublishMqttTopic(string topic, string payload) { try { var message = new MqttApplicationMessage() { Topic = topic, Payload = Encoding.UTF8.GetBytes(payload), QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce, Retain =true }; await mqttClient.PublishAsync(message); } catch (Exception ex) { throw ex; } } public async void ClientSubscribeTopic(string topic) { await mqttClient.SubscribeAsync(topic); } public async void ClientUnSubscribeTopic(string topic) { await mqttClient.UnsubscribeAsync(topic); } public bool IsConnected() { if (mqttClient == null) throw new Exception("客户端未创建"); return mqttClient.IsConnected; } public void Dispose() { mqttClient.Dispose(); } } }
客户端使用示例:
private Dictionary<string,List<MqttSensor>> GetMqttSensorData(string ip,int port,string username,string pwd,List<string> topics) { //声明一个MqttClient客户端对象 MqttClient client = new MqttClient(); var allDicts = new Dictionary<string, List<MqttSensor>>(); try { //链接成功事件 client.MqttClientConnectedHandler += delegate { //Console.WriteLine("连接成功!"); foreach (var topic in topics) client.ClientSubscribeTopic(topic); }; //接收消息事件 client.SubscriberMessageReceivedHandler += delegate (MqttApplicationMessageReceivedEventArgs args) { string ctopic = args.ApplicationMessage.Topic; string msg = Encoding.UTF8.GetString(args.ApplicationMessage.Payload); var dicts = JsonConvert.DeserializeObject<Dictionary<string, MqttSensor>>(msg); var result = new List<MqttSensor>(); foreach (var key in dicts.Keys) { if (result.Where(x => x.ID == key).Count() == 0) { var d = dicts[key] as MqttSensor; d.ID = key; result.Add(d); } } if (!allDicts.ContainsKey(ctopic) && topics.Contains(ctopic)) { allDicts.Add(ctopic, result); //Console.WriteLine("-------------------获取数据:" + ctopic + "-----------------"); //Console.WriteLine(JsonConvert.SerializeObject(result)); client.ClientUnSubscribeTopic(ctopic); topics.Remove(ctopic); } }; //异步开启客户端 client.WSClientStart(ip, port, username, pwd).GetAwaiter().GetResult(); while (true) { Thread.Sleep(100); if (topics.Count() == 0) { break; } } } catch (Exception) { throw; } finally { client.Dispose(); } return allDicts; } private string GetValue(string value, string type) { string result = ""; if (type == "运行状态") { result = (value.ToLower() == "true" ? "运行正常" : "停止运行"); } else { decimal v = 0; if (decimal.TryParse(value, out v)) { result = v + ""; } else { result = "-"; } } return result; }
参考文献:
https://zhuanlan.zhihu.com/p/419561816
https://www.jianshu.com/p/73d9c6668dfc
https://www.cnblogs.com/MyWPF/p/15129549.html
标签:obj,string,c#,void,笔记,MQTT,var,new,public From: https://www.cnblogs.com/misakayoucn/p/17577059.html