refactor(mqtt): 优化MQTT服务代码格式和数据处理逻辑
- 调整using语句的顺序以符合规范 - 格式化日志输出和代码缩进 - 在设备数据处理中添加时间间隔检查,每小时上传一次数据 - 优化取消订阅逻辑的代码结构
This commit is contained in:
@@ -1,16 +1,16 @@
|
|||||||
using Infrastructure.Extensions;
|
using System;
|
||||||
|
using System.Linq;
|
||||||
|
using System.Text;
|
||||||
|
using System.Text.Json;
|
||||||
|
using System.Threading;
|
||||||
|
using System.Threading.Tasks;
|
||||||
|
using Infrastructure.Extensions;
|
||||||
using Microsoft.Extensions.Configuration;
|
using Microsoft.Extensions.Configuration;
|
||||||
using Microsoft.Extensions.Hosting;
|
using Microsoft.Extensions.Hosting;
|
||||||
using Microsoft.Extensions.Logging;
|
using Microsoft.Extensions.Logging;
|
||||||
using MQTTnet;
|
using MQTTnet;
|
||||||
using MQTTnet.Client;
|
using MQTTnet.Client;
|
||||||
using MQTTnet.Protocol;
|
using MQTTnet.Protocol;
|
||||||
using System;
|
|
||||||
using System.Linq;
|
|
||||||
using System.Text;
|
|
||||||
using System.Text.Json;
|
|
||||||
using System.Threading;
|
|
||||||
using System.Threading.Tasks;
|
|
||||||
using ZR.Common.MqttHelper;
|
using ZR.Common.MqttHelper;
|
||||||
using ZR.Model.dc;
|
using ZR.Model.dc;
|
||||||
using ZR.Model.Dto;
|
using ZR.Model.Dto;
|
||||||
@@ -294,7 +294,9 @@ namespace ZR.Service.mqtt
|
|||||||
|
|
||||||
foreach (var item in result.Items)
|
foreach (var item in result.Items)
|
||||||
{
|
{
|
||||||
_logger.LogInformation($"订阅结果:{item.TopicFilter.Topic} -> {item.ResultCode}");
|
_logger.LogInformation(
|
||||||
|
$"订阅结果:{item.TopicFilter.Topic} -> {item.ResultCode}"
|
||||||
|
);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -318,7 +320,6 @@ namespace ZR.Service.mqtt
|
|||||||
var unsubscribeOptions = unsubscribeOptionsBuilder.Build();
|
var unsubscribeOptions = unsubscribeOptionsBuilder.Build();
|
||||||
await _mqttClient.UnsubscribeAsync(unsubscribeOptions);
|
await _mqttClient.UnsubscribeAsync(unsubscribeOptions);
|
||||||
_logger.LogInformation($"已取消订阅: {string.Join(", ", removedTopics)}");
|
_logger.LogInformation($"已取消订阅: {string.Join(", ", removedTopics)}");
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// 更新本地已订阅主题列表
|
// 更新本地已订阅主题列表
|
||||||
@@ -371,10 +372,22 @@ namespace ZR.Service.mqtt
|
|||||||
// plc网关抓取数据上传
|
// plc网关抓取数据上传
|
||||||
if (topic.Contains("device/data/push"))
|
if (topic.Contains("device/data/push"))
|
||||||
{
|
{
|
||||||
DeviceUploadDataGatWayDto deviceUploadDataGatWayDto = JsonSerializer.Deserialize<DeviceUploadDataGatWayDto>(payload);
|
DeviceUploadDataGatWayDto deviceUploadDataGatWayDto =
|
||||||
|
JsonSerializer.Deserialize<DeviceUploadDataGatWayDto>(payload);
|
||||||
// 这里添加设备消息处理逻辑
|
// 这里添加设备消息处理逻辑
|
||||||
string deviceCode = topic.Split("/")[2];
|
string deviceCode = topic.Split("/")[2];
|
||||||
DeviceUploadData deviceUploadData = new()
|
|
||||||
|
DeviceUploadData lastDeviceUploadData = Context
|
||||||
|
.Queryable<DeviceUploadData>()
|
||||||
|
.Where(it => it.DeviceCode == deviceCode)
|
||||||
|
.OrderByDescending(it => it.UploadTime)
|
||||||
|
.First();
|
||||||
|
DateTime lastTime = lastDeviceUploadData.UploadTime;
|
||||||
|
|
||||||
|
// 每一小时上传一下
|
||||||
|
|
||||||
|
DeviceUploadData deviceUploadData =
|
||||||
|
new()
|
||||||
{
|
{
|
||||||
FactoryCode = "上海干巷",
|
FactoryCode = "上海干巷",
|
||||||
WorkshopCode = "涂装车间",
|
WorkshopCode = "涂装车间",
|
||||||
@@ -383,7 +396,9 @@ namespace ZR.Service.mqtt
|
|||||||
DictCode = "device_dict_plc_001",
|
DictCode = "device_dict_plc_001",
|
||||||
Remark = "网关采集设备数据",
|
Remark = "网关采集设备数据",
|
||||||
UploadTime = DateTime.Now,
|
UploadTime = DateTime.Now,
|
||||||
CollectionTime = DateTimeOffset.FromUnixTimeMilliseconds(deviceUploadDataGatWayDto.Time).LocalDateTime,
|
CollectionTime = DateTimeOffset
|
||||||
|
.FromUnixTimeMilliseconds(deviceUploadDataGatWayDto.Time)
|
||||||
|
.LocalDateTime,
|
||||||
Value01 = deviceUploadDataGatWayDto.DeviceParams.Value01.ToString(),
|
Value01 = deviceUploadDataGatWayDto.DeviceParams.Value01.ToString(),
|
||||||
Value02 = deviceUploadDataGatWayDto.DeviceParams.Value02.ToString(),
|
Value02 = deviceUploadDataGatWayDto.DeviceParams.Value02.ToString(),
|
||||||
Value03 = deviceUploadDataGatWayDto.DeviceParams.Value03.ToString(),
|
Value03 = deviceUploadDataGatWayDto.DeviceParams.Value03.ToString(),
|
||||||
@@ -486,6 +501,7 @@ namespace ZR.Service.mqtt
|
|||||||
_logger.LogError(ex, "清理MQTT客户端资源时出错");
|
_logger.LogError(ex, "清理MQTT客户端资源时出错");
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
public void Dispose()
|
public void Dispose()
|
||||||
{
|
{
|
||||||
Dispose(true);
|
Dispose(true);
|
||||||
|
|||||||
Reference in New Issue
Block a user