/** *┌──────────────────────────────────────────────────────────────┐ *│ 描 述:MQTT通讯相关的工具类(MQTTnet 4.1.4.563) *│ 作 者:执笔小白 *│ 版 本:1.0 *│ 创建时间:2023-3-18 10:40:56 *└──────────────────────────────────────────────────────────────┘ *┌──────────────────────────────────────────────────────────────┐ *│ 命名空间: MqttnetServerWin *│ 类 名:MQTTHelper *└──────────────────────────────────────────────────────────────┘ */ using MQTTnet; using MQTTnet.Client; using MQTTnet.Packets; using MQTTnet.Server; using System; using System.Collections.Generic; using System.Linq; using System.Net; using System.Text; using System.Threading; using System.Threading.Tasks; namespace MqttnetServerWin { /// /// MQTT通讯相关的工具类 /// public class MQTTHelper { #region 变量 /// /// 记录日志、输出、保存等操作 /// private Action _Callback = null; #endregion 变量 #region Server /// /// MQTT服务 /// MqttServer _MqttServer = null; /// /// 创建MQTTServer并运行 /// public async Task CreateMQTTServerAndStart(MqttServerOptionsBuilder mqttServerOptionsBuilder, Action callback) { ResultData_MQTT resultData_MQTT = new ResultData_MQTT(); _Callback = callback; try { MqttServerOptions mqttServerOptions = mqttServerOptionsBuilder.Build(); _MqttServer = new MqttFactory().CreateMqttServer(mqttServerOptions); // 创建服务(配置) _MqttServer.StartedAsync += StartedHandle; // 服务器开启事件 _MqttServer.StoppedAsync += StoppedHandle; // 服务器关闭事件 _MqttServer.ClientConnectedAsync += ClientConnectedHandle; // 设置客户端连接成功后的处理程序 _MqttServer.ClientDisconnectedAsync += ClientDisconnectedHandle; // 设置客户端断开后的处理程序 _MqttServer.ClientSubscribedTopicAsync += ClientSubscribedTopicHandle; // 设置消息订阅通知 _MqttServer.ClientUnsubscribedTopicAsync += ClientUnsubscribedTopicHandle; // 设置消息退订通知 _MqttServer.ApplicationMessageNotConsumedAsync += ApplicationMessageNotConsumedHandle; // 设置消息处理程序 await _MqttServer.StartAsync(); // 开启服务 if (_MqttServer.IsStarted) { resultData_MQTT = new ResultData_MQTT() { ResultCode = 1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了开启MQTTServer_成功!" }; } else { resultData_MQTT = new ResultData_MQTT() { ResultCode = -1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了开启MQTTServer_失败!" }; } } catch (Exception ex) { resultData_MQTT = new ResultData_MQTT() { ResultCode = -1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了开启MQTTServer_失败!错误信息:" + ex.Message }; } _Callback?.Invoke(resultData_MQTT); return resultData_MQTT; } /// /// 简易创建MQTTServer并运行-不使用加密 /// /// IP /// 端口 /// 是否保持会话 /// 处理方法 /// public async Task CreateMQTTServerAndStart(string ip, int port, bool withPersistentSessions, Action callback) { ResultData_MQTT resultData_MQTT = new ResultData_MQTT(); _Callback = callback; try { MqttServerOptionsBuilder mqttServerOptionsBuilder = new MqttServerOptionsBuilder(); // MQTT服务器配置 mqttServerOptionsBuilder.WithDefaultEndpoint(); mqttServerOptionsBuilder.WithDefaultEndpointBoundIPAddress(IPAddress.Parse(ip)); // 设置Server的IP mqttServerOptionsBuilder.WithDefaultEndpointPort(port); // 设置Server的端口号 //mqttServerOptionsBuilder.WithEncryptedEndpointPort(port); // 使用加密的端点端口 mqttServerOptionsBuilder.WithPersistentSessions(withPersistentSessions); // 持续会话 mqttServerOptionsBuilder.WithConnectionBacklog(2000); // 最大连接数 //mqttServerOptionsBuilder.WithConnectionValidator(c => // 鉴权-方法失效 //{ // if (c.Username != uName || c.Password != uPwd) // { // c.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword; // } //}) MqttServerOptions mqttServerOptions = mqttServerOptionsBuilder.Build(); _MqttServer = new MqttFactory().CreateMqttServer(mqttServerOptions); // 创建服务(配置) _MqttServer.StartedAsync += StartedHandle; // 服务器开启事件 _MqttServer.StoppedAsync += StoppedHandle; // 服务器关闭事件 _MqttServer.ClientConnectedAsync += ClientConnectedHandle; // 设置客户端连接成功后的处理程序 _MqttServer.ClientDisconnectedAsync += ClientDisconnectedHandle; // 设置客户端断开后的处理程序 _MqttServer.ClientSubscribedTopicAsync += ClientSubscribedTopicHandle; // 设置消息订阅通知 _MqttServer.ClientUnsubscribedTopicAsync += ClientUnsubscribedTopicHandle; // 设置消息退订通知 _MqttServer.ValidatingConnectionAsync += ValidatingConnectionHandle; // 鉴权-未完 _MqttServer.ApplicationMessageNotConsumedAsync += ApplicationMessageNotConsumedHandle; // 设置消息处理程序 await _MqttServer.StartAsync(); // 开启服务 if (_MqttServer.IsStarted) { resultData_MQTT = new ResultData_MQTT() { ResultCode = 1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了开启MQTTServer_成功!" }; } else { resultData_MQTT = new ResultData_MQTT() { ResultCode = -1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了开启MQTTServer_失败!" }; } } catch (Exception ex) { resultData_MQTT = new ResultData_MQTT() { ResultCode = -1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了开启MQTTServer_失败!错误信息:" + ex.Message }; } _Callback?.Invoke(resultData_MQTT); return resultData_MQTT; } /// /// 关闭MQTTServer /// public async Task StopMQTTServer() { ResultData_MQTT resultData_MQTT = new ResultData_MQTT(); try { if (_MqttServer == null) { resultData_MQTT = new ResultData_MQTT() { ResultCode = -1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了关闭MQTTServer_出错!MQTTServer未在运行。" }; } else { foreach (var clientStatus in _MqttServer.GetClientsAsync().Result) { await clientStatus.DisconnectAsync(); } await _MqttServer.StopAsync(); _MqttServer = null; resultData_MQTT = new ResultData_MQTT() { ResultCode = 1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了关闭MQTTServer_成功!" }; } } catch (Exception ex) { resultData_MQTT = new ResultData_MQTT() { ResultCode = -1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了关闭MQTTServer_失败!错误信息:" + ex.Message }; } _Callback?.Invoke(resultData_MQTT); return resultData_MQTT; } /// /// 获取所有的客户端 /// public List GetClientsAsync() { return _MqttServer.GetClientsAsync().Result.ToList(); } /// /// 发送消息-未写 /// /// 主题 /// 消息 /// public Task SedMessage(string Topic, string msg) { try { //var clients = _MqttServer.GetClientsAsync().Result; //foreach (var client in clients) //{ //} } catch { } return Task.CompletedTask; } #region 处理事件 /// /// 开启Server的处理程序 /// private Task StartedHandle(EventArgs arg) { _Callback?.Invoke(new ResultData_MQTT() { ResultCode = 1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>MQTTServer已开启!" }); return Task.CompletedTask; } /// /// 关闭Server的处理程序 /// private Task StoppedHandle(EventArgs arg) { _Callback?.Invoke(new ResultData_MQTT() { ResultCode = 1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>MQTTServer已关闭!" }); return Task.CompletedTask; } /// /// 设置客户端连接成功后的处理程序 /// private Task ClientConnectedHandle(ClientConnectedEventArgs arg) { var clients = _MqttServer.GetClientsAsync().Result; _Callback?.Invoke(new ResultData_MQTT() { ResultCode = 1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + $">>>客户端'{arg.ClientId}'已成功连接!当前客户端连接数:{clients?.Count}个。" }); return Task.CompletedTask; } /// /// 设置客户端断开后的处理程序 /// private Task ClientDisconnectedHandle(ClientDisconnectedEventArgs arg) { var clients = _MqttServer.GetClientsAsync().Result; _Callback?.Invoke(new ResultData_MQTT() { ResultCode = 1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + $">>>客户端'{arg.ClientId}'已断开连接!当前客户端连接数:{clients?.Count}个。" }); return Task.CompletedTask; } /// /// 设置消息订阅通知 /// private Task ClientSubscribedTopicHandle(ClientSubscribedTopicEventArgs arg) { //if (!arg.Equals("admin")) //{ // var client = clients.Where(a => a.Id == arg.ClientId).FirstOrDefault(); // client?.DisconnectAsync(); // return Task.CompletedTask; //} _Callback?.Invoke(new ResultData_MQTT() { ResultCode = 1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + $">>>客户端'{arg.ClientId}'订阅了主题'{arg.TopicFilter.Topic}',主题服务质量:'{arg.TopicFilter.QualityOfServiceLevel}'!" }); return Task.CompletedTask; } /// /// 设置消息退订通知 /// private Task ClientUnsubscribedTopicHandle(ClientUnsubscribedTopicEventArgs arg) { _Callback?.Invoke(new ResultData_MQTT() { ResultCode = 1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + $">>>客户端{arg.ClientId}退订了主题{arg.TopicFilter}!" }); return Task.CompletedTask; } /// /// 鉴权-未写完 /// /// private Task ValidatingConnectionHandle(ValidatingConnectionEventArgs arg) // 鉴权 { if (arg.UserName != "Admin" || arg.Password != "Admin123") { } return Task.CompletedTask; } /// /// 设置消息处理程序 /// private Task ApplicationMessageNotConsumedHandle(ApplicationMessageNotConsumedEventArgs arg) { _Callback?.Invoke(new ResultData_MQTT() { ResultCode = -1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + $">>>客户端:'{arg.SenderId}'发布了消息:主题:'{arg.ApplicationMessage.Topic}'!内容:'{Encoding.UTF8.GetString(arg.ApplicationMessage.Payload)}';服务质量:{arg.ApplicationMessage.QualityOfServiceLevel};保留:{arg.ApplicationMessage.Retain}" }); return Task.CompletedTask; } #endregion 处理事件 #endregion Server #region Client /// /// 客户端 /// IMqttClient _MqttClient = null; /// /// 创建MQTTClient并运行 /// /// MQTTClient连接配置 /// 信息处理逻辑 /// public async Task CreateMQTTClientAndStart(MqttClientOptionsBuilder mqttClientOptionsBuilder, Action callback) { ResultData_MQTT resultData_MQTT = new ResultData_MQTT(); _Callback = callback; try { MqttClientOptions options = mqttClientOptionsBuilder.Build(); _MqttClient = new MqttFactory().CreateMqttClient(); _MqttClient.ConnectedAsync += ConnectedHandle; // 服务器连接事件 _MqttClient.DisconnectedAsync += DisconnectedHandle; // 服务器断开事件(可以写入重连事件) _MqttClient.ApplicationMessageReceivedAsync += ApplicationMessageReceivedHandle; // 接收消息事件 await _MqttClient.ConnectAsync(options); // 连接 if (_MqttClient.IsConnected) { resultData_MQTT = new ResultData_MQTT() { ResultCode = 1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了开启MQTTClient_成功!" }; } else { resultData_MQTT = new ResultData_MQTT() { ResultCode = -1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了开启MQTTClient_失败!" }; } } catch (Exception ex) { resultData_MQTT = new ResultData_MQTT() { ResultCode = -1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了开启MQTTClient_失败!错误信息:" + ex.Message }; } _Callback?.Invoke(resultData_MQTT); return resultData_MQTT; } /// /// 简易创建MQTTClient并运行 /// /// mqttServer的Url /// mqttServer的端口 /// 认证用用户名 /// 认证用密码 /// 信息处理逻辑 /// public async Task CreateMQTTClientAndStart(string mqttServerUrl, int port, string userName, string userPassword, Action callback) { ResultData_MQTT resultData_MQTT = new ResultData_MQTT(); _Callback = callback; try { MqttClientOptionsBuilder mqttClientOptionsBuilder = new MqttClientOptionsBuilder(); mqttClientOptionsBuilder.WithTcpServer(mqttServerUrl, port); // 设置MQTT服务器地址 if (!string.IsNullOrEmpty(userName)) { mqttClientOptionsBuilder.WithCredentials(userName, userPassword); // 设置鉴权参数 } mqttClientOptionsBuilder.WithClientId(Guid.NewGuid().ToString("N")); // 设置客户端序列号 MqttClientOptions options = mqttClientOptionsBuilder.Build(); _MqttClient = new MqttFactory().CreateMqttClient(); _MqttClient.ConnectedAsync += ConnectedHandle; // 服务器连接事件 _MqttClient.DisconnectedAsync += DisconnectedHandle; // 服务器断开事件(可以写入重连事件) _MqttClient.ApplicationMessageReceivedAsync += ApplicationMessageReceivedHandle; // 接收消息事件 await _MqttClient.ConnectAsync(options); // 连接 if (_MqttClient.IsConnected) { resultData_MQTT = new ResultData_MQTT() { ResultCode = 1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了开启MQTTClient_成功!" }; } else { resultData_MQTT = new ResultData_MQTT() { ResultCode = -1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了开启MQTTClient_失败!" }; } } catch (Exception ex) { resultData_MQTT = new ResultData_MQTT() { ResultCode = -1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了开启MQTTClient_失败!错误信息:" + ex.Message }; } _Callback?.Invoke(resultData_MQTT); return resultData_MQTT; } /// /// 关闭MQTTClient /// public async Task DisconnectAsync_Client() { ResultData_MQTT resultData_MQTT = new ResultData_MQTT(); try { if (_MqttClient != null && _MqttClient.IsConnected) { await _MqttClient.DisconnectAsync(); _MqttClient.Dispose(); _MqttClient = null; resultData_MQTT = new ResultData_MQTT() { ResultCode = 1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了关闭MQTTClient_成功!" }; } else { resultData_MQTT = new ResultData_MQTT() { ResultCode = -1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了关闭MQTTClient_失败!MQTTClient未开启连接!" }; } } catch (Exception ex) { resultData_MQTT = new ResultData_MQTT() { ResultCode = -1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了关闭MQTTClient_失败!错误信息:" + ex.Message }; } _Callback?.Invoke(resultData_MQTT); return resultData_MQTT; } /// /// 重连 /// /// public async Task ReconnectAsync_Client() { ResultData_MQTT resultData_MQTT = new ResultData_MQTT(); try { if (_MqttClient != null) { await _MqttClient.ReconnectAsync(); resultData_MQTT = new ResultData_MQTT() { ResultCode = 1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了MQTTClient重连_成功!" }; } else { resultData_MQTT = new ResultData_MQTT() { ResultCode = -1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了MQTTClient重连_失败!未设置MQTTClient连接!" }; } } catch (Exception ex) { resultData_MQTT = new ResultData_MQTT() { ResultCode = -1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了MQTTClient重连_失败!错误信息:" + ex.Message }; } _Callback?.Invoke(resultData_MQTT); return resultData_MQTT; } /// /// 订阅 /// /// 主题 public async Task SubscribeAsync_Client(string topic) { ResultData_MQTT resultData_MQTT = new ResultData_MQTT(); try { MqttTopicFilter topicFilter = new MqttTopicFilterBuilder().WithTopic(topic).Build(); await _MqttClient.SubscribeAsync(topicFilter, CancellationToken.None); resultData_MQTT = new ResultData_MQTT() { ResultCode = 1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + $">>>MQTTClient执行了订阅'{topic}'_成功!" }; } catch (Exception ex) { resultData_MQTT = new ResultData_MQTT() { ResultCode = -1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + $">>>MQTTClient执行了订阅'{topic}'_失败!错误信息:" + ex.Message }; } _Callback?.Invoke(resultData_MQTT); return resultData_MQTT; } /// /// 退订阅 /// /// 主题 public async Task UnsubscribeAsync_Client(string topic) { ResultData_MQTT resultData_MQTT = new ResultData_MQTT(); try { await MqttClientExtensions.UnsubscribeAsync(_MqttClient, topic, CancellationToken.None); resultData_MQTT = new ResultData_MQTT() { ResultCode = 1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + $">>>MQTTClient执行了退订'{topic}'_成功!" }; } catch (Exception ex) { resultData_MQTT = new ResultData_MQTT() { ResultCode = -1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + $">>>MQTTClient执行退订'{topic}'_失败!错误信息:" + ex.Message }; } _Callback?.Invoke(resultData_MQTT); return resultData_MQTT; } /// /// 发布消息( 必须在成功连接以后才生效 ) /// /// 主题 /// 信息 /// 是否保留 /// public async Task PublishAsync_Client(string topic, string msg, bool retained) { ResultData_MQTT resultData_MQTT = new ResultData_MQTT(); try { MqttApplicationMessageBuilder mqttApplicationMessageBuilder = new MqttApplicationMessageBuilder(); mqttApplicationMessageBuilder.WithTopic(topic); // 主题 mqttApplicationMessageBuilder.WithPayload(msg); // 信息 mqttApplicationMessageBuilder.WithRetainFlag(retained); // 保留 MqttApplicationMessage messageObj = mqttApplicationMessageBuilder.Build(); if (_MqttClient.IsConnected) { await _MqttClient.PublishAsync(messageObj, CancellationToken.None); resultData_MQTT = new ResultData_MQTT() { ResultCode = 1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + $">>>执行了发布信息_成功!主题:'{topic}',信息:'{msg}',是否保留:'{retained}'" }; } else { // 未连接 resultData_MQTT = new ResultData_MQTT() { ResultCode = -1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了发布信息_失败!MQTTClient未开启连接!" }; } } catch (Exception ex) { resultData_MQTT = new ResultData_MQTT() { ResultCode = -1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了发布信息_失败!错误信息:" + ex.Message }; } _Callback?.Invoke(resultData_MQTT); return resultData_MQTT; } #region 事件 /// /// 服务器连接事件 /// private Task ConnectedHandle(MqttClientConnectedEventArgs arg) { _Callback?.Invoke(new ResultData_MQTT() { ResultCode = 1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>已连接到MQTT服务器!" }); return Task.CompletedTask; } /// /// 服务器断开事件(可以写入重连事件) /// private Task DisconnectedHandle(MqttClientDisconnectedEventArgs arg) { _Callback?.Invoke(new ResultData_MQTT() { ResultCode = 1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + $">>>已断开与MQTT服务器连接!" }); return Task.CompletedTask; } /// /// 接收消息事件 /// private Task ApplicationMessageReceivedHandle(MqttApplicationMessageReceivedEventArgs arg) { _Callback?.Invoke(new ResultData_MQTT() { ResultCode = 1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + $">>>MQTTClient'{arg.ClientId}'内容:'{Encoding.UTF8.GetString(arg.ApplicationMessage.Payload)}';主题:'{arg.ApplicationMessage.Topic}',消息等级Qos:[{arg.ApplicationMessage.QualityOfServiceLevel}],是否保留:[{arg.ApplicationMessage.Retain}]", ResultObject1 = arg.ApplicationMessage.Topic, ResultObject2 = Encoding.UTF8.GetString(arg.ApplicationMessage.Payload) }); return Task.CompletedTask; } #endregion 事件 #endregion Client } /// /// 信息载体 /// public class ResultData_MQTT { /// /// 结果Code /// 正常1,其他为异常;0不作为回复结果 /// public int ResultCode { get; set; } = 0; /// /// 结果信息 /// public string ResultMsg { get; set; } = string.Empty; /// /// 扩展1 /// public object ResultObject1 { get; set; } = string.Empty; /// /// 扩展2 /// public object ResultObject2 { get; set; } = string.Empty; } }