283 lines
12 KiB
C#
283 lines
12 KiB
C#
using Aitex.Core.RT.Event;
|
||
using Aitex.Core.RT.Log;
|
||
using Aitex.Core.Util;
|
||
using MECF.Framework.Common.Equipment;
|
||
using System;
|
||
using System.Collections.Generic;
|
||
using System.Diagnostics;
|
||
using System.Linq;
|
||
using System.Text;
|
||
using System.Threading.Tasks;
|
||
|
||
namespace Aitex.Core.RT.DataCollection.HighPerformance
|
||
{
|
||
/// <summary>
|
||
/// 数据收集器缓存对象。
|
||
/// </summary>
|
||
/// <remarks>
|
||
/// 每个Table对应一个DataRecordCache对象。该对象负责缓存数据，而不提供任何写数据库相关操作。
|
||
/// <see cref="DataTraceManager"/>周期性调用<see cref="GetInsertSql"/>方法获取数据库写入表达式进行持久化操作。
|
||
/// </remarks>
|
||
internal class DataTraceCache : IDataTraceCache
|
||
{
|
||
#region Variables
|
||
|
||
internal const int SQL_BUILD_DURATION_TOO_SLOW_MS = 500;
|
||
internal const int MIN_CACHE_SIZE = 3000;
|
||
internal const int MIN_CACHE_PERIOD_MS = 10;
|
||
internal const int MIN_PERSIST_ITEMS = 100;
|
||
private const int MAX_SIZE_SQL_EXPR = 1000000; // approx. 5MB
|
||
|
||
private readonly object _syncRoot = new();
|
||
private readonly int _maxItemsPerPersist;
|
||
private readonly int _minCachePeriodMs;
|
||
private readonly int _maxCacheSize;
|
||
private readonly string _tableName;
|
||
private readonly string _insertExpression;
|
||
private readonly Queue<long> _qCachedTimestamps;
|
||
private readonly Stopwatch _stopwatch = new();
|
||
private readonly StringBuilder _sqlExpr = new (MAX_SIZE_SQL_EXPR); // preallocate 5MB
|
||
private readonly StringBuilder _sqlValuesList = new (MAX_SIZE_SQL_EXPR); // preallocate 5MB
|
||
|
||
private readonly R_TRIG _rTrigCacheHalfFull = new ();
|
||
private readonly R_TRIG _rTrigCacheAlmostFull = new ();
|
||
private readonly R_TRIG _rTrigCacheFull = new ();
|
||
private readonly R_TRIG _rTrigBuildSqlExprTooSlow = new ();
|
||
|
||
private int _lastPersistRows = 0;
|
||
private double _lastCachePeriodMs = 0;
|
||
|
||
#endregion
|
||
|
||
#region Constructors
|
||
|
||
/// <summary>
|
||
/// 创建数据记录器缓存对象。
|
||
/// </summary>
|
||
/// <param name="tableName">当前Cache对应的数据库表名称。</param>
|
||
/// <param name="category">当前Cache对应的分组名称。</param>
|
||
/// <param name="dataBuffers">数据获取器对象集合，保存当前Cache需要缓存的数据源。</param>
|
||
/// <param name="minCachePeriodMs">最小缓存周期，如果外部频繁调用<see cref="Cache"/>方法请求缓存数据，该周期内的重复请求被忽略，以避免高频缓存导致性能下降。</param>
|
||
/// <param name="maxItemsPerPersist">每次持久化的最大行数，避免一次性对数据库写入太多数据造成性能问题。</param>
|
||
/// <param name="maxCacheSize">数据缓存的最大项目数。</param>
|
||
/// <exception cref="ArgumentException"></exception>
|
||
public DataTraceCache(string tableName, string category, IReadOnlyList<IDataBuffer> dataBuffers, int minCachePeriodMs = 50, int maxItemsPerPersist = 2000, int maxCacheSize = MIN_CACHE_SIZE)
|
||
{
|
||
Debug.Assert(dataBuffers is { Count: > 0 }, "Incorrect Data Holders List.");
|
||
|
||
Category = category;
|
||
DataBuffers = dataBuffers;
|
||
_tableName = tableName;
|
||
_maxItemsPerPersist = maxItemsPerPersist < MIN_PERSIST_ITEMS ? MIN_PERSIST_ITEMS : maxItemsPerPersist;
|
||
_minCachePeriodMs = minCachePeriodMs < MIN_CACHE_PERIOD_MS ? MIN_CACHE_PERIOD_MS : minCachePeriodMs;
|
||
_maxCacheSize = maxCacheSize < MIN_CACHE_SIZE ? MIN_CACHE_SIZE : maxCacheSize;
|
||
_qCachedTimestamps = new Queue<long>(_maxCacheSize);
|
||
|
||
// 必须传入有效的数据收集器列表。
|
||
if (dataBuffers == null || dataBuffers.Count == 0)
|
||
throw new ArgumentException("数据缓冲器列表不能为空。", nameof(dataBuffers));
|
||
|
||
// 检查数据收集器列表中是否存在重复的序号。
|
||
if (dataBuffers.GroupBy(dc => dc.Index).Any(g => g.Count() > 1))
|
||
throw new ArgumentException("数据缓冲器列表中存在重复的序号。", nameof(dataBuffers));
|
||
|
||
// 检查数据收集器列表中是否存在重复的名称。
|
||
if (dataBuffers.GroupBy(dc => dc.Name).Any(g => g.Count() > 1))
|
||
throw new ArgumentException("数据缓冲器列表中存在重复的名称。", nameof(dataBuffers));
|
||
|
||
// 检查数据收集器列表中是否存在空的获取器。
|
||
if (dataBuffers.Any(dc => dc.Read == null))
|
||
throw new ArgumentException("数据缓冲器列表中存在空的缓冲器。", nameof(dataBuffers));
|
||
|
||
_insertExpression = BuildInsertExpression();
|
||
}
|
||
|
||
#endregion
|
||
|
||
#region Properties
|
||
|
||
/// <summary>
|
||
/// 返回当前缓存服务的模组名称。
|
||
/// </summary>
|
||
public string Category { get; }
|
||
|
||
/// <summary>
|
||
/// 返回数据获取器列表。
|
||
/// </summary>
|
||
public IReadOnlyList<IDataBuffer> DataBuffers { get; }
|
||
|
||
#endregion
|
||
|
||
#region Methods
|
||
|
||
/// <summary>
|
||
/// 创建Insert语句。
|
||
/// </summary>
|
||
/// <returns></returns>
|
||
private string BuildInsertExpression()
|
||
{
|
||
var sqlExpr = new StringBuilder();
|
||
|
||
lock (_syncRoot)
|
||
{
|
||
sqlExpr.Append($"INSERT INTO \"{_tableName}\"(\"time\" ");
|
||
foreach (var holder in DataBuffers)
|
||
{
|
||
sqlExpr.Append($",\"{holder.Name}\"");
|
||
}
|
||
|
||
sqlExpr.Append(")");
|
||
}
|
||
|
||
return sqlExpr.ToString();
|
||
}
|
||
|
||
/// <summary>
|
||
/// 立即缓存当前数据。
|
||
/// </summary>
|
||
public void Cache()
|
||
{
|
||
lock (_syncRoot)
|
||
{
|
||
#region Performance Monitor
|
||
|
||
// 限制最大Cache频率
|
||
if(_stopwatch.IsRunning && _stopwatch.ElapsedMilliseconds < _minCachePeriodMs)
|
||
return;
|
||
|
||
//TODO 需要限制Q的最大数据量，并且检测HalfFull和Full事件，并触发警告，可能PC性能出现问题
|
||
if (_qCachedTimestamps.Count >= _maxCacheSize)
|
||
{
|
||
// 缓存已满，不再缓存数据
|
||
_rTrigCacheFull.CLK = true;
|
||
if(_rTrigCacheFull.Q)
|
||
EV.PostWarningLog(ModuleName.System.ToString(), $"[{Category}]DataRecorderCache Full, Capacity {_maxCacheSize} items");
|
||
return;
|
||
}
|
||
|
||
if (_qCachedTimestamps.Count >= _maxCacheSize / 2)
|
||
{
|
||
// 缓存的数据量已经超过一半，触发警告
|
||
_rTrigCacheHalfFull.CLK = true;
|
||
if(_rTrigCacheHalfFull.Q)
|
||
EV.PostWarningLog(ModuleName.System.ToString(), $"[{Category}]DataRecorderCache Half full, Capacity {_maxCacheSize} items");
|
||
}
|
||
else if (_qCachedTimestamps.Count >= _maxCacheSize * 0.9)
|
||
{
|
||
// 缓存的数据量已经超过90%，触发警告
|
||
_rTrigCacheAlmostFull.CLK = true;
|
||
if(_rTrigCacheAlmostFull.Q)
|
||
EV.PostWarningLog(ModuleName.System.ToString(), $"[{Category}]DataRecorderCache Almost full, Usage {_qCachedTimestamps.Count} of {_maxCacheSize} items");
|
||
}
|
||
else
|
||
{
|
||
_rTrigCacheFull.CLK = false;
|
||
_rTrigCacheHalfFull.CLK = false;
|
||
_rTrigCacheAlmostFull.CLK = false;
|
||
}
|
||
|
||
#endregion
|
||
|
||
_lastCachePeriodMs = _stopwatch.ElapsedMilliseconds;
|
||
_stopwatch.Restart();
|
||
var ts = DateTime.Now.Ticks;
|
||
if (_qCachedTimestamps.Contains(ts))
|
||
{
|
||
LOG.Error($"时间戳{ts}已经存在。");
|
||
return;
|
||
}
|
||
|
||
var ret = Parallel.ForEach(DataBuffers, holder => { holder.Cache(ts); });
|
||
|
||
if (ret.IsCompleted)
|
||
{
|
||
_qCachedTimestamps.Enqueue(ts);
|
||
}
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// 获取将缓存数据写入数据库的SQL语句。
|
||
/// </summary>
|
||
/// <returns></returns>
|
||
/// <exception cref="NotImplementedException"></exception>
|
||
public string GetInsertSql()
|
||
{
|
||
lock (_syncRoot)
|
||
{
|
||
// 如果没数据，直接退出
|
||
if (_qCachedTimestamps.Count == 0)
|
||
return "";
|
||
|
||
var sw = new Stopwatch();
|
||
sw.Start();
|
||
|
||
var pickedCount = 0;
|
||
_sqlValuesList.Clear();
|
||
|
||
while (true)
|
||
{
|
||
// 没有缓存的数据了，退出
|
||
if (_qCachedTimestamps.Count == 0)
|
||
break;
|
||
|
||
var ts = _qCachedTimestamps.Dequeue();
|
||
{
|
||
_sqlValuesList.Append("(");
|
||
_sqlValuesList.Append($"'{ts}',");
|
||
foreach (var holder in DataBuffers)
|
||
{
|
||
var value = holder.Get(ts);
|
||
_sqlValuesList.Append(value);
|
||
_sqlValuesList.Append(",");
|
||
}
|
||
|
||
_sqlValuesList.Remove(_sqlValuesList.Length - 1, 1);
|
||
_sqlValuesList.Append("),");
|
||
}
|
||
|
||
pickedCount++;
|
||
|
||
if (pickedCount >= _maxItemsPerPersist)
|
||
break;
|
||
}
|
||
|
||
if (_sqlValuesList.Length <= 0)
|
||
{
|
||
_lastPersistRows = 0;
|
||
return "";
|
||
}
|
||
else
|
||
{
|
||
_lastPersistRows = pickedCount;
|
||
}
|
||
|
||
_sqlValuesList.Remove(_sqlValuesList.Length - 1, 1);
|
||
_sqlValuesList.Append(";");
|
||
|
||
// 拼接完整的SQL语句
|
||
_sqlExpr.Clear();
|
||
_sqlExpr.Append(_insertExpression);
|
||
_sqlExpr.Append(" VALUES ");
|
||
_sqlExpr.Append(_sqlValuesList);
|
||
|
||
sw.Stop();
|
||
|
||
//TODO 加入性能监视器，如果执行时间过长，则引发Warn事件；
|
||
_rTrigBuildSqlExprTooSlow.CLK = sw.ElapsedMilliseconds > SQL_BUILD_DURATION_TOO_SLOW_MS;
|
||
if (_rTrigBuildSqlExprTooSlow.Q)
|
||
LOG.Warning($"{nameof(DataTraceCache)} Build sql expression too slow, took {sw.ElapsedMilliseconds:F1}ms which was greater than {SQL_BUILD_DURATION_TOO_SLOW_MS:F1}ms");
|
||
// EV.PostWarningLog(ModuleName.System.ToString(), $"DataRecorderCache Build SQL Expression Too Slow, Took {sw.ElapsedMilliseconds:F1}ms greater than {SQL_BUILD_DURATION_TOO_SLOW_MS:F1}ms");
|
||
|
||
Debug.WriteLine(
|
||
$"[{DateTime.Now:yyyy-MM-dd HH:mm:ss.fff}] GenSQL:Ln{_lastPersistRows}/{sw.ElapsedMilliseconds}ms/{_sqlExpr.Length} bytes, LastCachePeriod:{_lastCachePeriodMs}ms, CacheRemained: {_qCachedTimestamps.Count}",
|
||
$"{nameof(DataTraceCache)} - {Category}");
|
||
|
||
return _sqlExpr.ToString();
|
||
}
|
||
}
|
||
|
||
#endregion
|
||
}
|
||
}
|