46-Pingzhi126-ZZ-WebApi/Server/MqttServer.cs

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<string> 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;
        }

        /// <summary>
        /// 发送MQTT消息
        /// </summary>
        /// <param name="message">要发送的消息内容到工位</param>
        /// <returns></returns>
        public static async Task<bool> 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;
            }
        }
    }
}