From cdfb906112d10c637260c14e970af5a0ee77b993 Mon Sep 17 00:00:00 2001 From: "David P.G" Date: Thu, 2 Oct 2025 17:35:35 +0800 Subject: [PATCH] =?UTF-8?q?=20=20feat(=E5=A4=84=E7=90=86=E5=99=A8):=20?= =?UTF-8?q?=E5=A2=9E=E5=BC=BA=E5=8F=98=E9=87=8F=E5=A4=84=E7=90=86=E9=93=BE?= =?UTF-8?q?=E5=B9=B6=E5=AE=9E=E7=8E=B0=E6=89=B9=E9=87=8F=E6=9B=B4=E6=96=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 UpdateDbVariableProcessor 处理器，通过队列和定时器实现数据库的批量更新，以降低负载。 - 重构 ValueConvertProcessor 处理器，使其能够解析 ConversionFormula 公式，计算出最终的 DisplayValue。 - 扩展 IVariableRepository 仓储接口，添加 UpdateBatchAsync 方法，并使用SqlSugar实现高效的批量更新。 - 优化 VariableContext 模型，将 NewValue 类型统一为 string，简化了数据流并提升了类型安全。 --- DMS.Application/Models/VariableContext.cs | 4 +- .../Processors/CheckValueChangedProcessor.cs | 4 +- .../Processors/UpdateDbVariableProcessor.cs | 90 ++++++++++++++++-- .../Processors/ValueConvertProcessor.cs | 93 ++++++++++++------- .../Repositories/IVariableRepository.cs | 7 ++ .../Repositories/VariableRepository.cs | 16 ++++ .../Services/OpcUa/OpcUaServiceManager.cs | 2 +- .../S7/OptimizedS7BackgroundService.cs | 3 +- 8 files changed, 169 insertions(+), 50 deletions(-) diff --git a/DMS.Application/Models/VariableContext.cs b/DMS.Application/Models/VariableContext.cs index 4a81a1d..170d8c9 100644 --- a/DMS.Application/Models/VariableContext.cs +++ b/DMS.Application/Models/VariableContext.cs @@ -7,10 +7,10 @@ namespace DMS.Application.Models { public VariableDto Data { get; set; } - public object NewValue { get; set; } + public string NewValue { get; set; } public bool IsHandled { get; set; } - public VariableContext(VariableDto data, object newValue=null) + public VariableContext(VariableDto data, string newValue="") { Data = data; IsHandled = false; // 默认未处理 diff --git a/DMS.Application/Services/Processors/CheckValueChangedProcessor.cs b/DMS.Application/Services/Processors/CheckValueChangedProcessor.cs index c93c0e6..40aa75c 100644 --- a/DMS.Application/Services/Processors/CheckValueChangedProcessor.cs +++ b/DMS.Application/Services/Processors/CheckValueChangedProcessor.cs @@ -7,9 +7,7 @@ namespace DMS.Application.Services.Processors; public class CheckValueChangedProcessor : IVariableProcessor { - public CheckValueChangedProcessor() - { - } + public Task ProcessAsync(VariableContext context) { // Variable newVariable = context.Data; diff --git a/DMS.Application/Services/Processors/UpdateDbVariableProcessor.cs b/DMS.Application/Services/Processors/UpdateDbVariableProcessor.cs index 72f212f..d619b39 100644 --- a/DMS.Application/Services/Processors/UpdateDbVariableProcessor.cs +++ b/DMS.Application/Services/Processors/UpdateDbVariableProcessor.cs @@ -1,29 +1,103 @@ -using System.Threading.Tasks; +using System.Collections.Concurrent; +using AutoMapper; +using DMS.Application.DTOs; using DMS.Application.Interfaces; using DMS.Application.Models; +using DMS.Core.Interfaces; using DMS.Core.Models; using Microsoft.Extensions.Logging; namespace DMS.Application.Services.Processors; -public class UpdateDbVariableProcessor : IVariableProcessor +/// +/// 负责将变量的当前值批量更新到数据库。 +/// +public class UpdateDbVariableProcessor : IVariableProcessor, IDisposable { - private readonly ILogger _logger; + private const int BATCH_SIZE = 50; // 批量更新的阈值 + private const int TIMER_INTERVAL_MS = 30 * 1000; // 30秒 - public UpdateDbVariableProcessor(ILogger logger) + private readonly ConcurrentQueue _queue = new(); + private readonly Timer _timer; + private readonly IRepositoryManager _repositoryManager; + private readonly ILogger _logger; + private readonly IMapper _mapper; + + public UpdateDbVariableProcessor(IRepositoryManager repositoryManager, ILogger logger, IMapper mapper) { + _repositoryManager = repositoryManager; _logger = logger; + _mapper = mapper; + + _timer = new Timer(async _ => await FlushQueueToDatabase(), null, Timeout.Infinite, Timeout.Infinite); + _timer.Change(TIMER_INTERVAL_MS, TIMER_INTERVAL_MS); // 启动定时器 + + _logger.LogInformation("UpdateDbVariableProcessor 初始化，批量大小 {BatchSize}，定时器间隔 {TimerInterval}ms", BATCH_SIZE, TIMER_INTERVAL_MS); } public async Task ProcessAsync(VariableContext context) { - try + // 检查新值是否有效，以及是否与旧值不同 + if (context.NewValue == null || Equals(context.Data.DataValue, context.NewValue?.ToString())) { - + return; // 值未变或新值无效，跳过 } - catch (Exception ex) + + // 用新值更新上下文中的数据，确保处理链的后续环节能看到最新值 + context.Data.DataValue = context.NewValue?.ToString(); + + _queue.Enqueue(context.Data); + + if (_queue.Count >= BATCH_SIZE) { - _logger.LogError(ex, $"更新数据库变量 {context.Data.Name} 失败: {ex.Message}"); + _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()) + { + // 去重：对于同一个变量，我们只关心其在本次批次中的最后一次更新值 + var uniqueItems = itemsToProcess + .GroupBy(v => v.Id) + .Select(g => g.Last()) + .ToList(); + + try + { + // **依赖于仓储层实现真正的批量更新** + var variableModels = _mapper.Map>(uniqueItems); + await _repositoryManager.Variables.UpdateBatchAsync(variableModels); + _logger.LogInformation("成功批量更新 {Count} 条变量记录到数据库", uniqueItems.Count); + } + catch (Exception ex) + { + _logger.LogError(ex, "批量更新 {Count} 条变量记录时发生错误: {ErrorMessage}", uniqueItems.Count, ex.Message); + // 错误处理策略：可以考虑将未成功的项重新入队，或记录到死信队列 + } + } + + // 重新启动定时器 + _timer.Change(TIMER_INTERVAL_MS, TIMER_INTERVAL_MS); + } + + public void Dispose() + { + _logger.LogInformation("正在释放 UpdateDbVariableProcessor，刷新队列中剩余的 {Count} 个项目", _queue.Count); + _timer?.Dispose(); + // 在 Dispose 时，尝试将剩余数据写入数据库 + FlushQueueToDatabase().Wait(); + _logger.LogInformation("UpdateDbVariableProcessor 已释放"); + } } \ No newline at end of file diff --git a/DMS.Application/Services/Processors/ValueConvertProcessor.cs b/DMS.Application/Services/Processors/ValueConvertProcessor.cs index fe3c739..83abf68 100644 --- a/DMS.Application/Services/Processors/ValueConvertProcessor.cs +++ b/DMS.Application/Services/Processors/ValueConvertProcessor.cs @@ -1,4 +1,5 @@ -using System.Globalization; +using System.Data; +using System.Globalization; using DMS.Application.DTOs; using DMS.Application.Interfaces; using DMS.Application.Models; @@ -14,123 +15,147 @@ public class ValueConvertProcessor : IVariableProcessor { _logger = logger; } - public async Task ProcessAsync(VariableContext context) + + public Task ProcessAsync(VariableContext context) { var oldValue = context.Data.DataValue; + + // 步骤 1: 将原始值转换为 DataValue 和 NumericValue ConvertS7ValueToStringAndNumeric(context.Data, context.NewValue); + + // 步骤 2: 根据公式计算 DisplayValue + CalculateDisplayValue(context.Data); + context.Data.UpdatedAt = DateTime.Now; - // 如何值没有变化则中断处理 - if (context.Data.DataValue==oldValue) + + // 如果值没有变化则中断处理链 + if (context.Data.DataValue == oldValue) { context.IsHandled = true; } + + return Task.CompletedTask; } + + /// + /// 根据转换公式计算用于UI显示的DisplayValue + /// + /// 需要处理的变量DTO + private void CalculateDisplayValue(VariableDto variable) + { + // 默认情况下，显示值等于原始数据值 + variable.DisplayValue = variable.DataValue; + + // 如果没有转换公式，则直接返回 + if (string.IsNullOrWhiteSpace(variable.ConversionFormula)) + { + return; + } + + try + { + // 将公式中的 'x' 替换为实际的数值 + // 使用 InvariantCulture 确保小数点是 '.' + string expression = variable.ConversionFormula.ToLowerInvariant() + .Replace("x", variable.NumericValue.ToString(CultureInfo.InvariantCulture)); + + // 使用 DataTable.Compute 来安全地计算表达式 + var result = new DataTable().Compute(expression, null); + + // 将计算结果格式化后赋给 DisplayValue + if (result is double || result is decimal || result is float) + { + variable.DisplayValue = string.Format("{0:F2}", result); // 默认格式化为两位小数，可根据需要调整 + } + else + { + variable.DisplayValue = result.ToString(); + } + } + catch (Exception ex) + { + _logger.LogError(ex, $"为变量 {variable.Name} (ID: {variable.Id}) 计算DisplayValue时出错。公式: '{variable.ConversionFormula}'"); + // 如果计算出错，DisplayValue 将保持为原始的 DataValue，保证程序健壮性 + } + } + /// /// 将从 S7 读取的对象值转换为字符串表示和数值表示 /// /// 关联的变量 DTO /// 从 S7 读取的原始对象值 - /// (字符串表示, 数值表示) private void ConvertS7ValueToStringAndNumeric(VariableDto variable, object value) { if (value == null) - return ; + return; - // 首先根据 value 的实际运行时类型进行匹配和转换 string directConversion = null; double numericValue = 0.0; - bool numericParsed = false; switch (value) { case double d: directConversion = d.ToString("G17", CultureInfo.InvariantCulture); numericValue = d; - numericParsed = true; break; case float f: directConversion = f.ToString("G9", CultureInfo.InvariantCulture); numericValue = f; - numericParsed = true; break; case int i: directConversion = i.ToString(CultureInfo.InvariantCulture); numericValue = i; - numericParsed = true; break; case uint ui: directConversion = ui.ToString(CultureInfo.InvariantCulture); numericValue = ui; - numericParsed = true; break; case short s: directConversion = s.ToString(CultureInfo.InvariantCulture); numericValue = s; - numericParsed = true; break; case ushort us: directConversion = us.ToString(CultureInfo.InvariantCulture); numericValue = us; - numericParsed = true; break; case byte b: directConversion = b.ToString(CultureInfo.InvariantCulture); numericValue = b; - numericParsed = true; break; case sbyte sb: directConversion = sb.ToString(CultureInfo.InvariantCulture); numericValue = sb; - numericParsed = true; break; case long l: directConversion = l.ToString(CultureInfo.InvariantCulture); numericValue = l; - numericParsed = true; break; case ulong ul: directConversion = ul.ToString(CultureInfo.InvariantCulture); numericValue = ul; - numericParsed = true; break; case bool boolValue: directConversion = boolValue.ToString().ToLowerInvariant(); numericValue = boolValue ? 1.0 : 0.0; - numericParsed = true; break; case string str: directConversion = str; - // 尝试从字符串解析数值 if (double.TryParse(str, NumberStyles.Float, CultureInfo.InvariantCulture, out var parsedFromStr)) { numericValue = parsedFromStr; - numericParsed = true; } break; default: - // 对于未预期的类型，记录日志 _logger.LogWarning($"变量 {variable.Name} 读取到未预期的数据类型: {value.GetType().Name}, 值: {value}"); directConversion = value.ToString() ?? string.Empty; - // 尝试从 ToString() 结果解析数值 if (double.TryParse(directConversion, NumberStyles.Float, CultureInfo.InvariantCulture, out var parsedFromObj)) { numericValue = parsedFromObj; - numericParsed = true; } break; } - // 如果直接转换成功，直接返回 - - // 如果直接转换未能解析数值，并且变量有明确的 DataType，可以尝试更精细的解析 - // (这部分逻辑在上面的 switch 中已经处理了大部分情况，这里作为后备) - // 在这个实现中，我们主要依赖于 value 的实际类型进行转换，因为这通常更可靠。 - // 如果需要，可以根据 variable.DataType 添加额外的解析逻辑。 - - // 返回最终结果 variable.DataValue = directConversion ?? value.ToString() ?? string.Empty; variable.NumericValue = numericValue; } - - } \ No newline at end of file diff --git a/DMS.Core/Interfaces/Repositories/IVariableRepository.cs b/DMS.Core/Interfaces/Repositories/IVariableRepository.cs index 73f0857..21401ac 100644 --- a/DMS.Core/Interfaces/Repositories/IVariableRepository.cs +++ b/DMS.Core/Interfaces/Repositories/IVariableRepository.cs @@ -25,5 +25,12 @@ namespace DMS.Core.Interfaces.Repositories /// OPC UA NodeId列表。 /// 找到的变量实体列表。 Task> GetByOpcUaNodeIdsAsync(List opcUaNodeIds); + + /// + /// 异步批量更新变量。 + /// + /// 要更新的变量实体集合。 + /// 受影响的行数。 + Task UpdateBatchAsync(IEnumerable variables); } } \ No newline at end of file diff --git a/DMS.Infrastructure/Repositories/VariableRepository.cs b/DMS.Infrastructure/Repositories/VariableRepository.cs index e06235a..a825b98 100644 --- a/DMS.Infrastructure/Repositories/VariableRepository.cs +++ b/DMS.Infrastructure/Repositories/VariableRepository.cs @@ -170,4 +170,20 @@ public class VariableRepository : BaseRepository, IVariableRepositor .ToListAsync(); return _mapper.Map>(dbVariables); } + + /// + /// 异步批量更新变量。 + /// + /// 要更新的变量实体集合。 + /// 受影响的行数。 + public async Task UpdateBatchAsync(IEnumerable variables) + { + var stopwatch = new Stopwatch(); + stopwatch.Start(); + var dbVariables = _mapper.Map>(variables); + var result = await _dbContext.GetInstance().Updateable(dbVariables).ExecuteCommandAsync(); + stopwatch.Stop(); + _logger.LogInformation($"Batch update {typeof(DbVariable)}, Count={dbVariables.Count}, 耗时：{stopwatch.ElapsedMilliseconds}ms"); + return result; + } } \ No newline at end of file diff --git a/DMS.Infrastructure/Services/OpcUa/OpcUaServiceManager.cs b/DMS.Infrastructure/Services/OpcUa/OpcUaServiceManager.cs index d0466e2..b4027a8 100644 --- a/DMS.Infrastructure/Services/OpcUa/OpcUaServiceManager.cs +++ b/DMS.Infrastructure/Services/OpcUa/OpcUaServiceManager.cs @@ -561,7 +561,7 @@ namespace DMS.Infrastructure.Services.OpcUa variable.Name, variable.Id, opcUaNode.NodeId, context.Device.Name); // 推送到数据处理队列 - await _dataProcessingService.EnqueueAsync(new VariableContext(variable, opcUaNode.Value)); + await _dataProcessingService.EnqueueAsync(new VariableContext(variable, opcUaNode.Value?.ToString())); _logger.LogDebug("HandleDataChanged: 变量 {VariableName} 的值已推送到数据处理队列", variable.Name); break; diff --git a/DMS.Infrastructure/Services/S7/OptimizedS7BackgroundService.cs b/DMS.Infrastructure/Services/S7/OptimizedS7BackgroundService.cs index 1ce3afe..f24624b 100644 --- a/DMS.Infrastructure/Services/S7/OptimizedS7BackgroundService.cs +++ b/DMS.Infrastructure/Services/S7/OptimizedS7BackgroundService.cs @@ -242,9 +242,8 @@ public class OptimizedS7BackgroundService : BackgroundService { if (readResults.TryGetValue(variable.S7Address, out var value)) { - // 将更新后的数据推入处理队列。 - await _dataProcessingService.EnqueueAsync(new VariableContext(variable, value)); + await _dataProcessingService.EnqueueAsync(new VariableContext(variable, value?.ToString())); } // else // {