using MQTTnet; using MQTTnet.Diagnostics.Logger; using MQTTnet.Packets; using MQTTnet.Protocol; using System; using System.Text; using System.Threading; using System.Threading.Tasks; using XiaoZhiSharp.MQTT.Common; namespace XiaoZhiSharp.MQTT { /// /// 连接成功事件 /// /// /// public delegate void ConnectedHandler(object source, MqttClientConnectedEventArgs arg); /// /// 断开连接成功事件 /// /// /// public delegate void DisconnectedHandler(object source, MqttClientDisconnectedEventArgs arg); /// /// 数据接收事件 /// /// /// public delegate void ReceivedHandler(object source, MqttApplicationMessageReceivedEventArgs arg); /// /// MQTT通讯相关的工具类 /// public class MqttNetClient { #region 变量 private MqttClientFactory mqttFactory = null; /// /// 客户端 /// private IMqttClient _MqttClient = null; private string mqttServerUrl = "127.0.0.1"; private int port = 8085; /// /// 记录日志、输出、保存等操作 /// private Action _Callback = null; #endregion 变量 public bool IsConnected { get { return _MqttClient != null ? _MqttClient.IsConnected : false; } } public MqttNetClient(string mqttServerUrl, int port) { var logger = new MqttNetEventLogger(); MqttNetConsoleLogger.ForwardToConsole(logger); mqttFactory = new MqttClientFactory(logger); this.mqttServerUrl = mqttServerUrl; this.port = port; } /// /// 创建MQTTClient并运行 /// /// MQTTClient连接配置 /// 信息处理逻辑 /// public async Task CreateMQTTClientAndStart(MqttClientOptionsBuilder mqttClientOptionsBuilder, Action callback) { ResultMqtt resultData_MQTT = new ResultMqtt(); _Callback = callback; try { MqttClientOptions options = mqttClientOptionsBuilder.Build(); _MqttClient = mqttFactory.CreateMqttClient(); RegisterEvents(_MqttClient); await _MqttClient.ConnectAsync(options); // 连接 if (_MqttClient.IsConnected) { resultData_MQTT = new ResultMqtt() { ResultCode = 1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了开启MQTTClient_成功!" }; } else { resultData_MQTT = new ResultMqtt() { ResultCode = -1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了开启MQTTClient_失败!" }; } } catch (Exception ex) { resultData_MQTT = new ResultMqtt() { 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) { ResultMqtt resultData_MQTT = new ResultMqtt(); _Callback = callback; try { MqttClientOptionsBuilder mqttClientOptionsBuilder = new MqttClientOptionsBuilder(); if (!string.IsNullOrEmpty(userName)) { mqttClientOptionsBuilder.WithCredentials(userName, userPassword);// 设置鉴权参数 } mqttClientOptionsBuilder.WithClientId(Guid.NewGuid().ToString("N"));// 设置客户端序列号 mqttClientOptionsBuilder.WithCleanSession(true); mqttClientOptionsBuilder.WithKeepAlivePeriod(TimeSpan.FromSeconds(120.0));//默认15秒 mqttClientOptionsBuilder.WithTimeout(TimeSpan.FromSeconds(2));//默认100秒 var isWebSocket = false; if (isWebSocket) { mqttClientOptionsBuilder.WithWebSocketServer(a => a.WithUri(mqttServerUrl));// 设置MQTT服务器地址,(带port时,或者启动Server时,不能在emqx里显示信息) } else { mqttClientOptionsBuilder.WithTcpServer(mqttServerUrl, port);// 设置MQTT服务器地址,(带port时,或者启动Server时,不能在emqx里显示信息) } MqttClientOptions options = mqttClientOptionsBuilder.Build(); _MqttClient = mqttFactory.CreateMqttClient(); RegisterEvents(_MqttClient); await _MqttClient.ConnectAsync(options); // 连接 if (_MqttClient.IsConnected) { resultData_MQTT = new ResultMqtt() { ResultCode = 1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了开启MQTTClient_成功!" }; } else { resultData_MQTT = new ResultMqtt() { ResultCode = -1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了开启MQTTClient_失败!" }; } } catch (Exception ex) { resultData_MQTT = new ResultMqtt() { ResultCode = -1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了开启MQTTClient_失败!错误信息:" + ex.Message }; } _Callback?.Invoke(resultData_MQTT); return resultData_MQTT; } private void RegisterEvents(IMqttClient mqttClient) { mqttClient.ConnectedAsync += ConnectedHandle; // 服务器连接事件 mqttClient.DisconnectedAsync += DisconnectedHandle; // 服务器断开事件(可以写入重连事件) mqttClient.ApplicationMessageReceivedAsync += ApplicationMessageReceivedHandle; // 发送消息事件 } /// /// 关闭MQTTClient /// public async Task DisconnectAsync() { ResultMqtt resultData_MQTT = new ResultMqtt(); try { if (_MqttClient != null && _MqttClient.IsConnected) { await _MqttClient.DisconnectAsync(); _MqttClient.Dispose(); _MqttClient = null; resultData_MQTT = new ResultMqtt() { ResultCode = 1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了关闭MQTTClient_成功!" }; } else { resultData_MQTT = new ResultMqtt() { ResultCode = -1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了关闭MQTTClient_失败!MQTTClient未开启连接!" }; } } catch (Exception ex) { resultData_MQTT = new ResultMqtt() { ResultCode = -1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了关闭MQTTClient_失败!错误信息:" + ex.Message }; } _Callback?.Invoke(resultData_MQTT); } /// /// 重连 /// /// public async Task ReconnectAsync() { ResultMqtt resultData_MQTT = new ResultMqtt(); try { if (_MqttClient != null) { await _MqttClient.ReconnectAsync(); resultData_MQTT = new ResultMqtt() { ResultCode = 1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了MQTTClient重连_成功!" }; } else { resultData_MQTT = new ResultMqtt() { ResultCode = -1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了MQTTClient重连_失败!未设置MQTTClient连接!" }; } } catch (Exception ex) { resultData_MQTT = new ResultMqtt() { ResultCode = -1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了MQTTClient重连_失败!错误信息:" + ex.Message }; } _Callback?.Invoke(resultData_MQTT); } /// /// 订阅 /// /// 主题 public async void SubscribeAsync(string topic) { /* Console.WriteLine("### CONNECTED WITH SERVER ###"); await _MqttClient.SubscribeAsync("#"); Console.WriteLine("### SUBSCRIBED ###"); */ ResultMqtt resultData_MQTT = new ResultMqtt(); try { /* var mqttClientSubscribeOptionsBuilder = mqttFactory.CreateSubscribeOptionsBuilder(); mqttClientSubscribeOptionsBuilder = mqttClientSubscribeOptionsBuilder.WithTopicFilter( f => { f.WithTopic(topic); }); MqttClientSubscribeOptions mqttClientSubscribeOptions = mqttClientSubscribeOptionsBuilder.Build(); await _MqttClient.SubscribeAsync(mqttClientSubscribeOptions, CancellationToken.None); */ MqttTopicFilter topicFilter = new MqttTopicFilterBuilder() .WithTopic(topic) .WithAtLeastOnceQoS() .Build(); await _MqttClient.SubscribeAsync(topicFilter, CancellationToken.None); resultData_MQTT = new ResultMqtt() { ResultCode = 1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + $">>>MQTTClient执行了订阅'{topic}'_成功!" }; } catch (Exception ex) { resultData_MQTT = new ResultMqtt() { ResultCode = -1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + $">>>MQTTClient执行了订阅'{topic}'_失败!错误信息:" + ex.Message }; } _Callback?.Invoke(resultData_MQTT); } /// /// 退订阅 /// /// 主题 public async void UnsubscribeAsync(string topic) { ResultMqtt resultData_MQTT = new ResultMqtt(); try { await _MqttClient.UnsubscribeAsync(topic, CancellationToken.None); resultData_MQTT = new ResultMqtt() { ResultCode = 1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + $">>>MQTTClient执行了退订'{topic}'_成功!" }; } catch (Exception ex) { resultData_MQTT = new ResultMqtt() { ResultCode = -1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + $">>>MQTTClient执行退订'{topic}'_失败!错误信息:" + ex.Message }; } _Callback?.Invoke(resultData_MQTT); } /// /// 发布消息( 必须在成功连接以后才生效 ) /// /// 主题 /// 信息 /// 是否保留 /// public async Task PublishAsync(string topic, string msg, bool retained) { ResultMqtt resultData_MQTT = new ResultMqtt(); try { /* https://zhuanlan.zhihu.com/p/671565499 * MQTT 定义了三个 服务质量QoS(Quality of Service)等级,分别为:QoS 0(默认),最多交付一次、QoS 1,至少交付一次、QoS 2,只交付一次。 QoS 0 最多交付一次: QoS 0 是最低的QoS等级。QoS 0消息即发即弃,不需要等待确认,不需要存储和重传, 因此对于接收方来说,永远都不需要担心收到重复的消息。 缺点是当我们使用QoS 0传递消息时,消息的可靠性完全依赖于底层的TCP协议。而TCP只能保证在连接稳定不关闭的情况下消息的可靠到达, 一旦出现连接关闭、重置,仍有可能丢失当前处于网络链路或操作系统底层缓冲区中的消息,这也是QoS 0消息最主要的丢失场景。 QoS 1 至少交付一次: 可以保证消息到达,所以适合传输一些较为重要的数据,比如下达关键指令、更新重要的有实时性要求的状态等。 缺点是可能会导致消息重复,所以当我们选择使用QoS 1时,还需要能够处理消息的重复,或者能够允许消息的重复。 QoS 2 恰好交付一次: 既可以保证消息到达,也可以保证消息不会重复,但传输成本最高。 如果我们不愿意自行实现去重方案,并且能够接受QoS 2带来的额外开销,那么QoS 2将是一个合适的选择。 通常我们会在金融、航空等行业场景下会更多地见到QoS 2的使用。 */ /* var applicationMessage = new MqttApplicationMessageBuilder() .WithTopic("A/B/C") .WithPayload("Hello World") .WithRetainFlag(retained) .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce) .Build(); */ MqttApplicationMessageBuilder mqttAppMsgBuilder = new MqttApplicationMessageBuilder(); mqttAppMsgBuilder.WithTopic(topic); // 主题 mqttAppMsgBuilder.WithPayload(msg); // 信息 mqttAppMsgBuilder.WithRetainFlag(retained); // 保留 mqttAppMsgBuilder.WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce); MqttApplicationMessage messageObj = mqttAppMsgBuilder.Build(); if (_MqttClient.IsConnected) { await _MqttClient.PublishAsync(messageObj, CancellationToken.None); resultData_MQTT = new ResultMqtt() { ResultCode = 1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + $">>>执行了发布信息_成功!主题:'{topic}',信息:'{msg}',是否保留:'{retained}'" }; } else { // 未连接 resultData_MQTT = new ResultMqtt() { ResultCode = -1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了发布信息_失败!MQTTClient未开启连接!" }; } } catch (Exception ex) { resultData_MQTT = new ResultMqtt() { ResultCode = -1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了发布信息_失败!错误信息:" + ex.Message }; } _Callback?.Invoke(resultData_MQTT); } #region 事件 /// /// 服务器连接事件 /// private Task ConnectedHandle(MqttClientConnectedEventArgs arg) { _Callback?.Invoke(new ResultMqtt() { ResultCode = 1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>已连接到MQTT服务器!" }); OnConnected(arg); return Task.CompletedTask;//CompletedTask.Instance; } /// /// 服务器断开事件(可以写入重连事件) /// private Task DisconnectedHandle(MqttClientDisconnectedEventArgs arg) { /* Console.WriteLine("### DISCONNECTED FROM SERVER ###"); await Task.Delay(TimeSpan.FromSeconds(5)); try { await _MqttClient.ConnectAsync(clientOptions);//重新连接 } catch { Console.WriteLine("### RECONNECTING FAILED ###"); } */ _Callback?.Invoke(new ResultMqtt() { ResultCode = 1, ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + $">>>已断开与MQTT服务器连接!" }); OnDisconnected(arg); return Task.CompletedTask;//CompletedTask.Instance; } /// /// 接收消息事件 /// private Task ApplicationMessageReceivedHandle(MqttApplicationMessageReceivedEventArgs arg) { Console.WriteLine("### RECEIVED APPLICATION MESSAGE ###"); Console.WriteLine($"+ Topic = {arg.ApplicationMessage.Topic}"); Console.WriteLine($"+ Payload = {Encoding.UTF8.GetString(arg.ApplicationMessage.Payload)}"); Console.WriteLine($"+ QoS = {arg.ApplicationMessage.QualityOfServiceLevel}"); Console.WriteLine($"+ Retain = {arg.ApplicationMessage.Retain}"); Console.WriteLine(); _Callback?.Invoke(new ResultMqtt() { 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) }); OnReceived(arg); return Task.CompletedTask;//CompletedTask.Instance; } #endregion 事件 #region Session Handles public event ConnectedHandler Connected; public event DisconnectedHandler Disconnected; public event ReceivedHandler Received; protected virtual void OnConnected(MqttClientConnectedEventArgs arg) { if (this.Connected != null) { this.Connected(this, arg); } } protected virtual void OnDisconnected(MqttClientDisconnectedEventArgs arg) { if (this.Disconnected != null) { this.Disconnected(this, arg); } } protected virtual void OnReceived(MqttApplicationMessageReceivedEventArgs arg) { if (this.Received != null) { this.Received(this, arg); } } #endregion Session Handles } }