添加项目文件。
This commit is contained in:
Administrator
@@ -0,0 +1,376 @@
|
||||
using Newtonsoft.Json.Linq;
|
||||
using PLCCommunication;
|
||||
using PLCCommunication.LogNet;
|
||||
using PLCCommunication.MQTT;
|
||||
using System;
|
||||
using System.Text;
|
||||
using System.Threading.Tasks;
|
||||
using System.Xml.Linq;
|
||||
|
||||
namespace WindowsService1
|
||||
{
|
||||
public class MqttClientWrapper
|
||||
{
|
||||
private MqttClient mqttClient;
|
||||
private bool mqttIsConnected;
|
||||
private const int MaxReconnectAttempts = 5;
|
||||
private const int ReconnectDelay = 5000; // 5 seconds
|
||||
|
||||
private MqttConnectionOptions options;
|
||||
public event Action<string, string, string> OnMessageReceived;
|
||||
public long ReceiveCount { get; private set; }
|
||||
|
||||
public PLCCommunication.LogNet.ILogNet LogNet = new LogNetDateTime("D:\\APILog\\Logs\\MQTTLogs", GenerateMode.ByEveryDay, 20);
|
||||
|
||||
public enum MessageFormat
|
||||
{
|
||||
Binary,
|
||||
Text,
|
||||
Xml,
|
||||
Json
|
||||
}
|
||||
|
||||
|
||||
public MqttClientWrapper(string ipAddress, int port, string clientId, MqttCredential credentials)
|
||||
{
|
||||
InitializeMqttOptions(ipAddress, port, clientId, credentials);
|
||||
}
|
||||
|
||||
private void InitializeMqttOptions(string ipAddress, int port, string clientId, MqttCredential credentials)
|
||||
{
|
||||
options = new MqttConnectionOptions
|
||||
{
|
||||
IpAddress = ipAddress,
|
||||
Port = port,
|
||||
ClientId = clientId,
|
||||
Credentials = credentials,
|
||||
UseRSAProvider = false,
|
||||
CleanSession = false,
|
||||
UseSSL = false,
|
||||
SSLSecure = false,
|
||||
CertificateFile = "",
|
||||
KeepAlivePeriod = TimeSpan.FromSeconds(30),
|
||||
KeepAliveSendInterval = TimeSpan.FromSeconds(10),
|
||||
ConnectTimeout = 5,
|
||||
WillMessage = null
|
||||
};
|
||||
}
|
||||
|
||||
public async Task InitializeAndConnectAsync()
|
||||
{
|
||||
try
|
||||
{
|
||||
mqttClient = new MqttClient(options);
|
||||
Logger.WriteMqtt("MQTT 初始化成功");
|
||||
|
||||
await ConnectMqttAsync();
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
Logger.WriteError($"MQTT 初始化失败: {ex.Message}");
|
||||
}
|
||||
}
|
||||
|
||||
private async Task ConnectMqttAsync()
|
||||
{
|
||||
try
|
||||
{
|
||||
JYResult connect = await mqttClient.ConnectServerAsync();
|
||||
if (connect.IsSuccess)
|
||||
{
|
||||
mqttIsConnected = true;
|
||||
Logger.WriteMqtt("MQTT 连接服务器成功");
|
||||
}
|
||||
else
|
||||
{
|
||||
Logger.WriteError("MQTT 无法连接到服务器");
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
Logger.WriteError($"MQTT 无法连接到服务器: {ex.Message}");
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
public async Task SendMessageAsync(string source, string publishTopic, string type, JObject messageObj)
|
||||
{
|
||||
if (!mqttClient.IsConnected)
|
||||
{
|
||||
mqttIsConnected = false;
|
||||
|
||||
Logger.WriteMqtt("客户端未连接到服务器。正在尝试重新连接...");
|
||||
|
||||
await ReconnectWithRetry();
|
||||
if (!mqttClient.IsConnected)
|
||||
{
|
||||
throw new InvalidOperationException("无法与 MQTT 代理建立连接.");
|
||||
}
|
||||
}
|
||||
try
|
||||
{
|
||||
if (messageObj != null)
|
||||
{
|
||||
string json = messageObj.ToString();
|
||||
//string json = JsonConvert.SerializeObject(messageObj);
|
||||
JYResult result = await mqttClient.PublishMessageAsync(new MqttApplicationMessage
|
||||
{
|
||||
Topic = publishTopic,
|
||||
QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce,
|
||||
Payload = Encoding.UTF8.GetBytes(json),
|
||||
Retain = false,
|
||||
});
|
||||
if (result.IsSuccess)
|
||||
{
|
||||
if (Service1.SwitchLog)
|
||||
{
|
||||
Logger.WriteMqtt($"线程 [{source}] 发送 [{type}] 消息到主题 [{publishTopic}]: {json}");
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
Logger.WriteMqtt($"线程 [{source}] 发送 [{type}] 消息到主题 [{publishTopic}] 失败: {result.Message}");
|
||||
|
||||
}
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
Logger.WriteError($"发送消息错误: {ex.Message}");
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
public async Task SendMessageAsync(string source, string publishTopic, string type, string messageObj)
|
||||
{
|
||||
if (!mqttClient.IsConnected)
|
||||
{
|
||||
mqttIsConnected = false;
|
||||
|
||||
Logger.WriteMqtt("客户端未连接到服务器。正在尝试重新连接...");
|
||||
|
||||
|
||||
await ReconnectWithRetry();
|
||||
if (!mqttClient.IsConnected)
|
||||
{
|
||||
throw new InvalidOperationException("无法与 MQTT 代理建立连接.");
|
||||
}
|
||||
}
|
||||
try
|
||||
{
|
||||
if (messageObj != null)
|
||||
{
|
||||
JYResult result = await mqttClient.PublishMessageAsync(new MqttApplicationMessage
|
||||
{
|
||||
Topic = publishTopic,
|
||||
QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce,
|
||||
Payload = Encoding.UTF8.GetBytes(messageObj),
|
||||
Retain = false,
|
||||
});
|
||||
|
||||
if (result.IsSuccess)
|
||||
{
|
||||
if (Service1.SwitchLog)
|
||||
{
|
||||
Logger.WriteMqtt($"线程 [{source}] 发送 [{type}] 消息到主题 [{publishTopic}]: {messageObj}");
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
Logger.WriteMqtt($"线程 [{source}] 发送 [{type}] 消息到主题 [{publishTopic}] 失败: {result.Message}");
|
||||
|
||||
}
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
Logger.WriteError($"发送消息错误: {ex.Message}");
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
public void SubscribeMessage(string topic, MessageFormat format = MessageFormat.Text, bool truncateLongMessage = false)
|
||||
{
|
||||
if (!mqttIsConnected)
|
||||
{
|
||||
ConnectMqttAsync().Wait();
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
JYResult operateResult = mqttClient.SubscribeMessage(new string[] { topic });
|
||||
if (!operateResult.IsSuccess)
|
||||
{
|
||||
Logger.WriteMqtt($"订阅消息失败: {operateResult.Message}");
|
||||
return;
|
||||
}
|
||||
|
||||
SubscribeTopic subscribeTopic = mqttClient.GetSubscribeTopic(topic);
|
||||
if (subscribeTopic != null)
|
||||
{
|
||||
subscribeTopic.OnMqttMessageReceived += (sender, args) =>
|
||||
{
|
||||
HandleReceivedMessage(args.Topic, args.Payload, format, truncateLongMessage);
|
||||
};
|
||||
Logger.WriteMqtt($"成功订阅主题: {topic}");
|
||||
}
|
||||
else
|
||||
{
|
||||
Logger.WriteMqtt($"无法获取订阅主题 {topic}");
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
Logger.WriteError($"订阅主题时出错 {topic}: {ex.Message}");
|
||||
}
|
||||
}
|
||||
|
||||
private void HandleReceivedMessage(string topic, byte[] payload, MessageFormat format, bool truncateLongMessage)
|
||||
{
|
||||
ReceiveCount++;
|
||||
string text = string.Empty;
|
||||
|
||||
try
|
||||
{
|
||||
switch (format)
|
||||
{
|
||||
case MessageFormat.Binary:
|
||||
text = BitConverter.ToString(payload).Replace("-", " ");
|
||||
break;
|
||||
case MessageFormat.Text:
|
||||
text = Encoding.UTF8.GetString(payload);
|
||||
break;
|
||||
case MessageFormat.Xml:
|
||||
try
|
||||
{
|
||||
text = XElement.Parse(Encoding.UTF8.GetString(payload)).ToString();
|
||||
}
|
||||
catch
|
||||
{
|
||||
text = Encoding.UTF8.GetString(payload);
|
||||
}
|
||||
break;
|
||||
case MessageFormat.Json:
|
||||
try
|
||||
{
|
||||
text = JObject.Parse(Encoding.UTF8.GetString(payload)).ToString();
|
||||
}
|
||||
catch
|
||||
{
|
||||
text = Encoding.UTF8.GetString(payload);
|
||||
}
|
||||
break;
|
||||
}
|
||||
//if (Service1.SwitchLog)
|
||||
//{
|
||||
// LogNet.WriteDebug($"收到主题 {topic}: {text}");
|
||||
//}
|
||||
//if (truncateLongMessage && text.Length > 200)
|
||||
//{
|
||||
// text = text.Substring(0, 200) + "...";
|
||||
//}
|
||||
|
||||
string timestamp = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff");
|
||||
OnMessageReceived?.Invoke(timestamp, topic, text);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
Logger.WriteError($"处理收到的消息时出错: {ex.Message}");
|
||||
}
|
||||
}
|
||||
|
||||
public void UnsubscribeMessage(string topic)
|
||||
{
|
||||
try
|
||||
{
|
||||
JYResult operateResult = mqttClient.UnSubscribeMessage(new string[] { topic });
|
||||
if (operateResult.IsSuccess)
|
||||
{
|
||||
Logger.WriteMqtt($"成功取消订阅主题: {topic}");
|
||||
}
|
||||
else
|
||||
{
|
||||
Logger.WriteMqtt($"取消订阅主题失败 {topic}: {operateResult.Message}");
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
Logger.WriteError($"取消订阅主题时出错 {topic}: {ex.Message}");
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 重连
|
||||
/// </summary>
|
||||
/// <returns></returns>
|
||||
private async Task ReconnectWithRetry()
|
||||
{
|
||||
int attempts = 0;
|
||||
while (!mqttClient.IsConnected && attempts < MaxReconnectAttempts)
|
||||
{
|
||||
try
|
||||
{
|
||||
await mqttClient.ConnectServerAsync();
|
||||
Logger.WriteMqtt("重新连接成功.");
|
||||
|
||||
return;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
attempts++;
|
||||
Logger.WriteMqtt($"重新连接尝试 {attempts} 失败的: {ex.Message}");
|
||||
await Task.Delay(ReconnectDelay);
|
||||
}
|
||||
}
|
||||
if (!mqttClient.IsConnected)
|
||||
{
|
||||
Logger.WriteMqtt("尝试次数达到上限后仍无法重新连接.");
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
private async Task ReconnectAsync()
|
||||
{
|
||||
Logger.WriteMqtt("正在尝试重新连接...");
|
||||
mqttIsConnected = false;
|
||||
int retryCount = 0;
|
||||
const int maxRetries = 5;
|
||||
const int retryDelayMs = 5000;
|
||||
|
||||
while (!mqttIsConnected && retryCount < maxRetries)
|
||||
{
|
||||
try
|
||||
{
|
||||
await ConnectMqttAsync();
|
||||
if (mqttIsConnected)
|
||||
{
|
||||
Logger.WriteMqtt("重新连接成功");
|
||||
return;
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
Logger.WriteMqtt($"重新连接尝试 {retryCount + 1} 失败的: {ex.Message}");
|
||||
}
|
||||
|
||||
retryCount++;
|
||||
await Task.Delay(retryDelayMs);
|
||||
}
|
||||
|
||||
if (!mqttIsConnected)
|
||||
{
|
||||
Logger.WriteMqtt("多次尝试后仍无法重新连接");
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
public async Task Disconnect()
|
||||
{
|
||||
await mqttClient.ConnectCloseAsync();
|
||||
mqttIsConnected = false;
|
||||
Logger.WriteMqtt("与 MQTT服务端断开连接！！！");
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user