xiaowei/DMS
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

初步完成历史记录存储功能

Browse Source
This commit is contained in:
David P.G
2025-09-11 09:42:44 +08:00
parent a43b978097
commit 3c47276406
4 changed files with 69 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

111
DMS.Application/Services/Processors/HistoryProcessor.cs
View File

@@ -1,6 +1,9 @@
using DMS.Application.Interfaces; using DMS.Application.Interfaces;
using DMS.Application.Models; using DMS.Application.Models;
using DMS.Core.Interfaces;
using DMS.Core.Models; using DMS.Core.Models;
using System.Collections.Concurrent;
using Microsoft.Extensions.Logging;
namespace DMS.Application.Services.Processors; namespace DMS.Application.Services.Processors;
@@ -9,77 +12,91 @@ public class HistoryProcessor : IVariableProcessor, IDisposable
private const int BATCH_SIZE = 50; // 批量写入的阈值 private const int BATCH_SIZE = 50; // 批量写入的阈值
private const int TIMER_INTERVAL_MS = 30 * 1000; // 30秒 private const int TIMER_INTERVAL_MS = 30 * 1000; // 30秒
// private readonly ConcurrentQueue<DbVariableHistory> _queue = new(); private readonly ConcurrentQueue<VariableHistory> _queue = new();
private readonly Timer _timer; private readonly Timer _timer;
private readonly IRepositoryManager _repositoryManager;
private readonly ILogger<HistoryProcessor> _logger;
public HistoryProcessor() public HistoryProcessor(IRepositoryManager repositoryManager, ILogger<HistoryProcessor> logger)
{ {
_repositoryManager = repositoryManager;
_logger = logger;
_timer = new Timer(async _ => await FlushQueueToDatabase(), null, Timeout.Infinite, Timeout.Infinite); _timer = new Timer(async _ => await FlushQueueToDatabase(), null, Timeout.Infinite, Timeout.Infinite);
_timer.Change(TIMER_INTERVAL_MS, TIMER_INTERVAL_MS); // 启动定时器 _timer.Change(TIMER_INTERVAL_MS, TIMER_INTERVAL_MS); // 启动定时器
_logger.LogInformation("HistoryProcessor 初始化，批量大小 {BatchSize}，定时器间隔 {TimerInterval}ms", BATCH_SIZE, TIMER_INTERVAL_MS);
} }
public async Task ProcessAsync(VariableContext context) public async Task ProcessAsync(VariableContext context)
{ {
// 只有当数据发生变化时才记录历史 // 只有当数据需要保存时才记录历史
// if (!context.Data.IsSave) // 如果数据已经被其他处理器处理过或者不需要保存，则跳过 if (!context.Data.IsHistoryEnabled) // 如果数据已经被其他处理器处理过或者不需要保存，则跳过
// { {
// return; _logger.LogDebug("变量 {VariableName} (ID: {VariableId}) 历史记录已禁用，跳过处理", context.Data.Name, context.Data.Id);
// } return;
// }
// // 将 Variable 转换为 DbVariableHistory
// var historyData = new DbVariableHistory // 将 VariableDto 转换为 VariableHistory
// { var historyData = new VariableHistory
// Name = context.Data.Name, {
// NodeId = context.Data.NodeId, VariableId = context.Data.Id,
// DataValue = context.Data.DataValue, Value = context.Data.DataValue?.ToString() ?? string.Empty,
// VariableId = context.Data.Id, Timestamp = DateTime.Now // 记录当前时间
// Timestamp = DateTime.Now // 记录当前时间 };
// };
// _queue.Enqueue(historyData);
// _queue.Enqueue(historyData); _logger.LogDebug("变量 {VariableName} (ID: {VariableId}) 历史数据已入队，队列数量: {QueueCount}", context.Data.Name, context.Data.Id, _queue.Count);
//
// if (_queue.Count >= BATCH_SIZE) if (_queue.Count >= BATCH_SIZE)
// { {
// await FlushQueueToDatabase(); _logger.LogInformation("达到批量大小 ({BatchSize})，正在刷新队列到数据库", BATCH_SIZE);
// } await FlushQueueToDatabase();
}
} }
private async Task FlushQueueToDatabase() private async Task FlushQueueToDatabase()
{ {
// 停止定时器，防止在写入过程中再次触发 // 停止定时器，防止在写入过程中再次触发
// _timer.Change(Timeout.Infinite, Timeout.Infinite); _timer.Change(Timeout.Infinite, Timeout.Infinite);
// _logger.LogDebug("数据库刷新期间定时器已停止");
// var itemsToProcess = new List<DbVariableHistory>();
// while (_queue.TryDequeue(out var item)) var itemsToProcess = new List<VariableHistory>();
// { while (_queue.TryDequeue(out var item))
// itemsToProcess.Add(item); {
// } itemsToProcess.Add(item);
// }
// if (itemsToProcess.Any())
// { if (itemsToProcess.Any())
// try {
// { try
// using var db = DbContext.GetInstance(); {
// await db.Insertable(itemsToProcess).ExecuteCommandAsync(); await _repositoryManager.VariableHistories.AddBatchAsync(itemsToProcess);
// NlogHelper.Info($"成功批量插入 {itemsToProcess.Count} 条历史变量数据。"); _logger.LogInformation("成功插入 {Count} 条变量历史记录到数据库", itemsToProcess.Count);
// } }
// catch (Exception ex) catch (Exception ex)
// { {
// NlogHelper.Error( $"批量插入历史变量数据时发生错误: {ex.Message}",ex); _logger.LogError(ex, "批量插入 {Count} 条变量历史记录时发生错误: {ErrorMessage}", itemsToProcess.Count, ex.Message);
// // 错误处理：可以将未成功插入的数据重新放回队列，或者记录到日志中以便后续处理 // 错误处理：可以将未成功插入的数据重新放回队列，或者记录到日志中以便后续处理
// // 为了简单起见，这里不重新入队，避免无限循环 // 为了简单起见，这里不重新入队，避免无限循环
// } }
// } }
else
{
_logger.LogDebug("队列中没有需要处理的项目");
}
// 重新启动定时器 // 重新启动定时器
_timer.Change(TIMER_INTERVAL_MS, TIMER_INTERVAL_MS); _timer.Change(TIMER_INTERVAL_MS, TIMER_INTERVAL_MS);
_logger.LogDebug("定时器已重启，间隔 {TimerInterval}ms", TIMER_INTERVAL_MS);
} }
public void Dispose() public void Dispose()
{ {
_logger.LogInformation("正在释放 HistoryProcessor，刷新队列中剩余的 {Count} 个项目", _queue.Count);
_timer?.Dispose(); _timer?.Dispose();
// 在 Dispose 时，尝试将剩余数据写入数据库 // 在 Dispose 时，尝试将剩余数据写入数据库
FlushQueueToDatabase().Wait(); FlushQueueToDatabase().Wait();
_logger.LogInformation("HistoryProcessor 已释放");
} }
} }

9
DMS.Application/Services/Processors/UpdateDbVariableProcessor.cs
View File

@@ -19,16 +19,7 @@ public class UpdateDbVariableProcessor : IVariableProcessor
{ {
try try
{ {
// 假设 DataServices 有一个方法来更新 Variable
// await _dataServices.UpdateVariableAsync(context.Data);
// _logger.LogInformation($"数据库变量 {context.Data.Name} 更新成功，值为: {context.Data.DataValue}");
// if (!_dataServices.AllVariables.TryGetValue(context.Data.Id, out Variable oldVariable))
// {
// _logger.LogWarning($"数据库更新完成修改变量值是否改变时在_dataServices.AllVariables中找不到Id:{context.Data.Id},Name:{context.Data.Name}的变量。");
// context.IsHandled = true;
// }
// oldVariable.DataValue = context.Data.DataValue;
} }
catch (Exception ex) catch (Exception ex)
{ {

2
DMS.Infrastructure/Profiles/MappingProfile.cs
View File

@@ -35,5 +35,7 @@ public class MappingProfile : Profile
CreateMap<DbMenu, MenuBean>().ReverseMap(); CreateMap<DbMenu, MenuBean>().ReverseMap();
CreateMap<DbNlog, Nlog>().ReverseMap(); CreateMap<DbNlog, Nlog>().ReverseMap();
CreateMap<DbVariableHistory, VariableHistory>().ReverseMap();
} }
} }

2
DMS.WPF/App.xaml.cs
View File

@@ -72,9 +72,9 @@ public partial class App : System.Windows.Application
var dataProcessingService = Host.Services.GetRequiredService<IDataProcessingService>(); var dataProcessingService = Host.Services.GetRequiredService<IDataProcessingService>();
dataProcessingService.AddProcessor(Host.Services.GetRequiredService<CheckValueChangedProcessor>()); dataProcessingService.AddProcessor(Host.Services.GetRequiredService<CheckValueChangedProcessor>());
dataProcessingService.AddProcessor(Host.Services.GetRequiredService<LoggingProcessor>()); dataProcessingService.AddProcessor(Host.Services.GetRequiredService<LoggingProcessor>());
dataProcessingService.AddProcessor(Host.Services.GetRequiredService<HistoryProcessor>());
dataProcessingService.AddProcessor(Host.Services.GetRequiredService<MqttPublishProcessor>()); dataProcessingService.AddProcessor(Host.Services.GetRequiredService<MqttPublishProcessor>());
dataProcessingService.AddProcessor(Host.Services.GetRequiredService<UpdateDbVariableProcessor>()); dataProcessingService.AddProcessor(Host.Services.GetRequiredService<UpdateDbVariableProcessor>());
//dataProcessingService.AddProcessor(Host.Services.GetRequiredService<HistoryProcessor>());
} }
catch (Exception exception) catch (Exception exception)
{ {