using System; using System.Collections.Generic; using System.Linq; using System.Text; using System.Threading.Tasks; using WebApi.Helpers; using Logger; using System.Reflection; using System.Threading; namespace WebApi { public class MqttServer { private static string mqttTopic; private static string mqttIp; private static int mqttPort; private static bool isConnected = false; private static bool isReconnecting = false; private static CancellationTokenSource reconnectCancellation; public static event Action ShowMsg; // 记录日志事件 public static async Task StartMqttServer() { try { reconnectCancellation = new CancellationTokenSource(); // 从配置文件读取MQTT配置 string mqttConfig = Tools.AppConfigManage.ReadConfig("mqttIp"); if (!string.IsNullOrEmpty(mqttConfig)) { var parts = mqttConfig.Split(','); if (parts.Length == 2) { mqttIp = parts[0]; mqttPort = Convert.ToInt32(parts[1]); } } mqttTopic = Tools.AppConfigManage.ReadConfig("mqttTopic"); // 注册MQTT事件处理 Mqtt.ConnectionStatusChanged += OnMqttConnectionStatusChanged; //Mqtt.MessageReceived += OnMqttMessageReceived; await ConnectToMqttServer(); } catch (Exception ex) { Log.FunError(ex, MethodBase.GetCurrentMethod().Name); // 如果首次连接失败,也启动重连 await StartReconnection(); } } private static async Task ConnectToMqttServer() { try { // 连接MQTT服务器 await Mqtt.RunMqttStart(mqttIp, mqttPort, "", ""); // 订阅主题 if (!string.IsNullOrEmpty(mqttTopic)) { await Mqtt.RunMqttSubscribe(mqttTopic); } } catch (Exception ex) { Log.FunError(ex, MethodBase.GetCurrentMethod().Name); throw; } } private static async Task StartReconnection() { if (isReconnecting) return; try { isReconnecting = true; int retryCount = 0; int maxRetryInterval = 10; // 最大重试间隔(秒) while (!isConnected && !reconnectCancellation.Token.IsCancellationRequested) { try { retryCount++; // 计算重试间隔(指数退避),但不超过最大间隔 double delaySeconds = Math.Min(Math.Pow(2, retryCount - 1), maxRetryInterval); ShowMsg?.Invoke($"MQTT准备重连,{delaySeconds}秒后重试..."); await Task.Delay(TimeSpan.FromSeconds(delaySeconds), reconnectCancellation.Token); await ConnectToMqttServer(); } catch (Exception ex) { Log.FunError(ex, "MQTT重连失败"); } } } finally { isReconnecting = false; } } private static void OnMqttConnectionStatusChanged(bool connected) { isConnected = connected; if (connected) { ShowMsg?.Invoke($"MQTT已连接到服务器 {mqttIp}:{mqttPort}"); } else { ShowMsg?.Invoke($"MQTT已断开连接"); // 启动重连 Task.Run(StartReconnection); } } private static void OnMqttMessageReceived(string clientId, string topic, string message) { try { ShowMsg?.Invoke($"收到MQTT消息: Topic={topic}, Message={message}"); // 这里可以添加消息处理逻辑 if (topic == mqttTopic) { // 处理特定主题的消息 } } catch (Exception ex) { Log.FunError(ex, MethodBase.GetCurrentMethod().Name); } } public static async Task StopMqttServer() { try { // 停止重连 if (reconnectCancellation != null) { reconnectCancellation.Cancel(); reconnectCancellation.Dispose(); reconnectCancellation = null; } // 取消订阅 if (!string.IsNullOrEmpty(mqttTopic)) { await Mqtt.RunMqttUnSubscribe(mqttTopic); } // 断开连接 await Mqtt.RunMqttStop(); // 取消事件注册 Mqtt.ConnectionStatusChanged -= OnMqttConnectionStatusChanged; //Mqtt.MessageReceived -= OnMqttMessageReceived; } catch (Exception ex) { Log.FunError(ex, MethodBase.GetCurrentMethod().Name); } } public static bool IsConnected() { return isConnected; } /// /// 发送MQTT消息 /// /// 要发送的消息内容到工位 /// public static async Task SendMqttMessageToOp(string opName, string type, string value) { try { if (!MqttServer.IsConnected()) { ShowMsg?.Invoke("MQTT未连接,无法发送消息"); return false; } string mqttTopic = Tools.AppConfigManage.ReadConfig("mqttTopic"); if (string.IsNullOrEmpty(mqttTopic)) { ShowMsg?.Invoke("MQTT主题未配置"); return false; } string sendMsg = $"MES|{type}|{opName}|{value}"; // MES/InstantMessaging/PZ126/OP010 // MES|AGVStatus|OP010|1 await Mqtt.RunMqttPublish(mqttTopic + opName, sendMsg); ShowMsg?.Invoke($"MQTTSend: Topic={mqttTopic}, Message={sendMsg}"); return true; } catch (Exception ex) { Log.FunError(ex, MethodBase.GetCurrentMethod().Name); return false; } } } }