From 3c47276406cc04ccc4b9652e44c53d3d62990d8c Mon Sep 17 00:00:00 2001 From: "David P.G" Date: Thu, 11 Sep 2025 09:42:44 +0800 Subject: [PATCH] =?UTF-8?q?=E5=88=9D=E6=AD=A5=E5=AE=8C=E6=88=90=E5=8E=86?= =?UTF-8?q?=E5=8F=B2=E8=AE=B0=E5=BD=95=E5=AD=98=E5=82=A8=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Services/Processors/HistoryProcessor.cs | 113 ++++++++++-------- .../Processors/UpdateDbVariableProcessor.cs | 11 +- DMS.Infrastructure/Profiles/MappingProfile.cs | 2 + DMS.WPF/App.xaml.cs | 2 +- 4 files changed, 69 insertions(+), 59 deletions(-) diff --git a/DMS.Application/Services/Processors/HistoryProcessor.cs b/DMS.Application/Services/Processors/HistoryProcessor.cs index 8292971..0adf06d 100644 --- a/DMS.Application/Services/Processors/HistoryProcessor.cs +++ b/DMS.Application/Services/Processors/HistoryProcessor.cs @@ -1,6 +1,9 @@ using DMS.Application.Interfaces; using DMS.Application.Models; +using DMS.Core.Interfaces; using DMS.Core.Models; +using System.Collections.Concurrent; +using Microsoft.Extensions.Logging; namespace DMS.Application.Services.Processors; @@ -9,77 +12,91 @@ public class HistoryProcessor : IVariableProcessor, IDisposable private const int BATCH_SIZE = 50; // 批量写入的阈值 private const int TIMER_INTERVAL_MS = 30 * 1000; // 30秒 - // private readonly ConcurrentQueue _queue = new(); + private readonly ConcurrentQueue _queue = new(); private readonly Timer _timer; + private readonly IRepositoryManager _repositoryManager; + private readonly ILogger _logger; - public HistoryProcessor() + public HistoryProcessor(IRepositoryManager repositoryManager, ILogger logger) { + _repositoryManager = repositoryManager; + _logger = logger; _timer = new Timer(async _ => await FlushQueueToDatabase(), null, Timeout.Infinite, Timeout.Infinite); _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) { - // 只有当数据发生变化时才记录历史 - // if (!context.Data.IsSave) // 如果数据已经被其他处理器处理过或者不需要保存，则跳过 - // { - // return; - // } - // - // // 将 Variable 转换为 DbVariableHistory - // var historyData = new DbVariableHistory - // { - // Name = context.Data.Name, - // NodeId = context.Data.NodeId, - // DataValue = context.Data.DataValue, - // VariableId = context.Data.Id, - // Timestamp = DateTime.Now // 记录当前时间 - // }; - // - // _queue.Enqueue(historyData); - // - // if (_queue.Count >= BATCH_SIZE) - // { - // await FlushQueueToDatabase(); - // } + // 只有当数据需要保存时才记录历史 + if (!context.Data.IsHistoryEnabled) // 如果数据已经被其他处理器处理过或者不需要保存，则跳过 + { + _logger.LogDebug("变量 {VariableName} (ID: {VariableId}) 历史记录已禁用，跳过处理", context.Data.Name, context.Data.Id); + return; + } + + // 将 VariableDto 转换为 VariableHistory + var historyData = new VariableHistory + { + VariableId = context.Data.Id, + Value = context.Data.DataValue?.ToString() ?? string.Empty, + Timestamp = DateTime.Now // 记录当前时间 + }; + + _queue.Enqueue(historyData); + _logger.LogDebug("变量 {VariableName} (ID: {VariableId}) 历史数据已入队，队列数量: {QueueCount}", context.Data.Name, context.Data.Id, _queue.Count); + + if (_queue.Count >= BATCH_SIZE) + { + _logger.LogInformation("达到批量大小 ({BatchSize})，正在刷新队列到数据库", BATCH_SIZE); + await FlushQueueToDatabase(); + } } private async Task FlushQueueToDatabase() { // 停止定时器，防止在写入过程中再次触发 - // _timer.Change(Timeout.Infinite, Timeout.Infinite); - // - // var itemsToProcess = new List(); - // while (_queue.TryDequeue(out var item)) - // { - // itemsToProcess.Add(item); - // } - // - // if (itemsToProcess.Any()) - // { - // try - // { - // using var db = DbContext.GetInstance(); - // await db.Insertable(itemsToProcess).ExecuteCommandAsync(); - // NlogHelper.Info($"成功批量插入 {itemsToProcess.Count} 条历史变量数据。"); - // } - // catch (Exception ex) - // { - // NlogHelper.Error( $"批量插入历史变量数据时发生错误: {ex.Message}",ex); - // // 错误处理：可以将未成功插入的数据重新放回队列，或者记录到日志中以便后续处理 - // // 为了简单起见，这里不重新入队，避免无限循环 - // } - // } + _timer.Change(Timeout.Infinite, Timeout.Infinite); + _logger.LogDebug("数据库刷新期间定时器已停止"); + + var itemsToProcess = new List(); + while (_queue.TryDequeue(out var item)) + { + itemsToProcess.Add(item); + } + + if (itemsToProcess.Any()) + { + try + { + await _repositoryManager.VariableHistories.AddBatchAsync(itemsToProcess); + _logger.LogInformation("成功插入 {Count} 条变量历史记录到数据库", itemsToProcess.Count); + } + catch (Exception ex) + { + _logger.LogError(ex, "批量插入 {Count} 条变量历史记录时发生错误: {ErrorMessage}", itemsToProcess.Count, ex.Message); + // 错误处理：可以将未成功插入的数据重新放回队列，或者记录到日志中以便后续处理 + // 为了简单起见，这里不重新入队，避免无限循环 + } + } + else + { + _logger.LogDebug("队列中没有需要处理的项目"); + } // 重新启动定时器 _timer.Change(TIMER_INTERVAL_MS, TIMER_INTERVAL_MS); + _logger.LogDebug("定时器已重启，间隔 {TimerInterval}ms", TIMER_INTERVAL_MS); } public void Dispose() { + _logger.LogInformation("正在释放 HistoryProcessor，刷新队列中剩余的 {Count} 个项目", _queue.Count); _timer?.Dispose(); // 在 Dispose 时，尝试将剩余数据写入数据库 - FlushQueueToDatabase().Wait(); + FlushQueueToDatabase().Wait(); + _logger.LogInformation("HistoryProcessor 已释放"); } } \ No newline at end of file diff --git a/DMS.Application/Services/Processors/UpdateDbVariableProcessor.cs b/DMS.Application/Services/Processors/UpdateDbVariableProcessor.cs index df18b74..72f212f 100644 --- a/DMS.Application/Services/Processors/UpdateDbVariableProcessor.cs +++ b/DMS.Application/Services/Processors/UpdateDbVariableProcessor.cs @@ -19,16 +19,7 @@ public class UpdateDbVariableProcessor : IVariableProcessor { 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) { diff --git a/DMS.Infrastructure/Profiles/MappingProfile.cs b/DMS.Infrastructure/Profiles/MappingProfile.cs index 260bfe1..9d313e1 100644 --- a/DMS.Infrastructure/Profiles/MappingProfile.cs +++ b/DMS.Infrastructure/Profiles/MappingProfile.cs @@ -35,5 +35,7 @@ public class MappingProfile : Profile CreateMap().ReverseMap(); CreateMap().ReverseMap(); + + CreateMap().ReverseMap(); } } diff --git a/DMS.WPF/App.xaml.cs b/DMS.WPF/App.xaml.cs index 927fb2e..5532ad0 100644 --- a/DMS.WPF/App.xaml.cs +++ b/DMS.WPF/App.xaml.cs @@ -72,9 +72,9 @@ public partial class App : System.Windows.Application var dataProcessingService = Host.Services.GetRequiredService(); dataProcessingService.AddProcessor(Host.Services.GetRequiredService()); dataProcessingService.AddProcessor(Host.Services.GetRequiredService()); + dataProcessingService.AddProcessor(Host.Services.GetRequiredService()); dataProcessingService.AddProcessor(Host.Services.GetRequiredService()); dataProcessingService.AddProcessor(Host.Services.GetRequiredService()); - //dataProcessingService.AddProcessor(Host.Services.GetRequiredService()); } catch (Exception exception) {