chore: 初始化平芝126KV总装线 WebApi
This commit is contained in:
yexingqiang
215
Server/MqttServer.cs
Normal file
215
Server/MqttServer.cs
Normal file
@@ -0,0 +1,215 @@
|
||||
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;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user