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
// {