MQTT类 - C# - .NET Core 一个常用的MQTT的操作类using System; using System.Text; using System.Threading; using System.Threading.Tasks; using System.Security.Cryptography.X509Certificates; using MQTTnet; using MQTTnet.Client; using MQTTnet.Client.Options; using MQTTnet.Client.Ssl; public class MqttClientService { private IMqttClient _client; private MqttClientOptions _options; private readonly string _broker = "your-mqtt-broker"; // MQTT 服务器地址 private readonly int _port = 1883; // MQTT 端口（1883: TCP, 8883: TLS） private readonly string _clientId = "CSharpClient"; // MQTT 客户端ID private readonly string _username = "your-username"; // MQTT 账号 private readonly string _password = "your-password"; // MQTT 密码 private readonly int _reconnectDelay = 5000; // 断线重连间隔（毫秒） public event Action<string, string> OnMessageReceived; // 事件回调（主题，消息） public event Action<bool> OnConnectionChanged; // 连接状态回调（true=已连接，false=断开） public bool IsConnected { get; private set; } = false; // 连接状态 public MqttClientService() { InitializeMqttClient(); } private void InitializeMqttClient() { var factory = new MqttFactory(); _client = factory.CreateMqttClient(); _client.UseConnectedHandler(async e => { Console.WriteLine("MQTT 已连接到服务器"); IsConnected = true; OnConnectionChanged?.Invoke(true); // 触发连接成功事件 }); _client.UseDisconnectedHandler(async e => { Console.WriteLine("MQTT 连接丢失，尝试重连..."); IsConnected = false; OnConnectionChanged?.Invoke(false); // 触发连接断开事件 await Task.Delay(_reconnectDelay); await ConnectAsync(); }); _client.UseApplicationMessageReceivedHandler(e => { string topic = e.ApplicationMessage.Topic; string payload = Encoding.UTF8.GetString(e.ApplicationMessage.Payload); OnMessageReceived?.Invoke(topic, payload); // 触发消息接收事件 }); _options = new MqttClientOptionsBuilder() .WithTcpServer(_broker, _port) .WithClientId(_clientId) .WithCredentials(_username, _password) .WithCleanSession() .Build(); } /// <summary> /// 普通连接（非 SSL/TLS） /// </summary> public async Task ConnectAsync() { if (!_client.IsConnected) { try { await _client.ConnectAsync(_options, CancellationToken.None); } catch (Exception ex) { Console.WriteLine($"MQTT 连接失败: {ex.Message}"); await Task.Delay(_reconnectDelay); await ConnectAsync(); } } } public async Task SubscribeAsync(string topic) { if (_client.IsConnected) { await _client.SubscribeAsync(new MqttTopicFilterBuilder() .WithTopic(topic) .Build()); Console.WriteLine($"已订阅主题: {topic}"); } else { Console.WriteLine("MQTT 未连接，无法订阅主题"); } } public async Task PublishAsync(string topic, string message) { if (_client.IsConnected) { var msg = new MqttApplicationMessageBuilder() .WithTopic(topic) .WithPayload(message) .WithQualityOfServiceLevel(MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce) .Build(); await _client.PublishAsync(msg); Console.WriteLine($"已发布消息: 主题={topic}, 内容={message}"); } else { Console.WriteLine("MQTT 未连接，无法发布消息"); } } public async Task DisconnectAsync() { if (_client.IsConnected) { await _client.DisconnectAsync(); IsConnected = false; OnConnectionChanged?.Invoke(false); // 触发连接断开事件 Console.WriteLine("MQTT 已断开连接"); } } } 使用示例using System; using System.Threading.Tasks; class Program { static async Task Main(string[] args) { var mqttService = new MqttClientService(); // 订阅消息接收事件 mqttService.OnMessageReceived += HandleMqttMessage; // 订阅连接状态变化事件 mqttService.OnConnectionChanged += HandleConnectionStatus; // **普通连接** await mqttService.ConnectAsync(); await mqttService.SubscribeAsync("test/topic"); // 定时检查连接状态 Task.Run(async () => { while (true) { Console.WriteLine($"当前 MQTT 连接状态: {(mqttService.IsConnected ? "已连接" : "未连接")}"); await Task.Delay(5000); } }); // 发布消息 await mqttService.PublishAsync("test/topic", "Hello MQTT Secure!"); Console.ReadLine(); await mqttService.DisconnectAsync(); } // **单独的回调方法 - 处理收到的消息** static void HandleMqttMessage(string topic, string message) { Console.WriteLine($"收到消息: 主题={topic}, 内容={message}"); } // **单独的回调方法 - 处理连接状态变化** static void HandleConnectionStatus(bool isConnected) { Console.WriteLine($"MQTT 连接状态变化: {(isConnected ? "已连接" : "断开连接")}"); } }