huangsucan/ADP
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

上报数据方式优化

Browse Source
This commit is contained in:
hsc
2026-06-30 10:01:30 +08:00
parent 8f124fb08d
commit f27045153c
7 changed files with 368 additions and 190 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
MonitorModule/ViewModels/MonitorChannel.cs
View File

@@ -16,6 +16,8 @@ namespace MonitorModule.ViewModels
{
// ===== 标识 =====
public string DeviceName { get; init; } = string.Empty;
/// <summary>硬件指纹（物理设备唯一标识，用于匹配广播事件）</summary>
public string Fingerprint { get; init; } = string.Empty;
public string MethodName { get; init; } = string.Empty;
public string DisplayName { get; init; } = string.Empty;

325
MonitorModule/ViewModels/MonitorViewModel.cs
View File

@@ -10,6 +10,7 @@ using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Collections.ObjectModel;
using System.Diagnostics;
using System.Linq;
using System.Reflection;
using System.Threading;
@@ -75,7 +76,7 @@ namespace MonitorModule.ViewModels
public ICommand RefreshCommand { get; }
#endregion
#region 私有字段与并发缓存
#region 私有字段
private bool IsInitiated = false;
private IScopedProvider? _scope;
private DeviceManager? _deviceManager;
@@ -90,27 +91,24 @@ namespace MonitorModule.ViewModels
};
private int _colorIndex;
// 图表采样定时器 (100ms)
private readonly DispatcherTimer _sampleTimer;
private double _sampleTime;
private const double _sampleInterval = 0.1;
/// <summary>广播器（由 UIShare 层注入，每 Scope 一个实例）</summary>
private HardwareDataBroadcaster? _broadcaster;
/// <summary>反射方法缓存：Channel → (MethodInfo, DeviceInstance)</summary>
private readonly Dictionary<MonitorChannel, (MethodInfo Method, object Device)> _methodCache = new();
/// <summary>用于 OxyPlot X 轴相对秒数</summary>
private readonly Stopwatch _stopwatch = new();
/// <summary>OxyPlot 刷新定时器（仅负责滚动 X 轴 + InvalidatePlot）</summary>
private readonly DispatcherTimer _plotRefreshTimer;
/// <summary>事件订阅令牌，用于 Dispose 时取消订阅</summary>
private SubscriptionToken? _subscriptionToken;
// ==========================================
// 🟥 基于 Task.Run 批量入库的核心高并发结构
// 批量入库核心结构
// ==========================================
/// <summary>高并发无锁无阻塞队列</summary>
private readonly ConcurrentQueue<MonitorValueEntity> _insertQueue = new();
/// <summary>常驻后台的数据库消费任务</summary>
private Task? _dbFlushTask;
/// <summary>定量触发阈值</summary>
private const int BulkInsertThreshold = 50;
/// <summary>防并发消费锁标志</summary>
private int _isFlushing = 0;
#endregion
@@ -127,206 +125,139 @@ namespace MonitorModule.ViewModels
RefreshDataCommand = new DelegateCommand(OnRefreshData);
RefreshCommand = new DelegateCommand(OnExpand);
// 前端采样定时器保持 100ms
_sampleTimer = new DispatcherTimer { Interval = TimeSpan.FromSeconds(_sampleInterval) };
_sampleTimer.Tick += OnSampleTick;
// OxyPlot 刷新定时器（1000ms，只负责视觉更新）
_plotRefreshTimer = new DispatcherTimer(DispatcherPriority.Background)
{
Interval = TimeSpan.FromMilliseconds(1000)
};
_plotRefreshTimer.Tick += OnPlotRefreshTick;
}
/// <summary>
/// 页面销毁与清理
/// </summary>
public void Dispose()
{
// 1. 率先掐断所有底层的异步信号（引发 Task.Delay 抛出异常安全退出常驻循环）
_monitorCTS?.Cancel();
_monitorCTS?.Dispose();
_monitorCTS = null;
// 2. 停掉前端 UI 定时器
_sampleTimer?.Stop();
_sampleTimer.Tick -= OnSampleTick;
_plotRefreshTimer.Stop();
_plotRefreshTimer.Tick -= OnPlotRefreshTick;
_stopwatch.Stop();
// 取消 EventAggregator 订阅
if (_subscriptionToken != null)
_eventAggregator.GetEvent<HardwareDataReportedEvent>().Unsubscribe(_subscriptionToken);
// 3. 🟥 等待后台入库消费 Task 彻底结束，最大安全等待 1.5 秒
if (_dbFlushTask != null)
{
try
{
_dbFlushTask.Wait(TimeSpan.FromSeconds(1.5));
}
catch { /* 忽略退出时的线程取消异常 */ }
try { _dbFlushTask.Wait(TimeSpan.FromSeconds(1.5)); } catch { }
}
// 4. 释放容器作用域及清空集合
_scope?.Dispose();
Channels.Clear();
AvailableMethods.Clear();
_methodCache.Clear();
}
#region 🟥 核心逻辑：基于 Task.Run 的常驻后台消费循环
#region 事件驱动：接收广播数据
/// <summary>
/// 开启常驻后台的数据库消费线程
/// 由 HardwareDataBroadcaster 通过 EventAggregator 广播触发。
/// 已通过 ThreadOption.UIThread 确保在 UI 线程执行。
/// 通过 HardwareFingerprint + MethodName 匹配监测通道。
/// </summary>
private void OnHardwareDataReceived(HardwareReportArgs args)
{
// 仅处理属于当前 Scope 的事件
if (string.IsNullOrEmpty(TestStatus) || args.Scope != TestStatus) return;
// 以硬件指纹 + 方法名唯一匹配通道（与逻辑名解耦）
var channel = Channels.FirstOrDefault(c =>
c.Fingerprint == args.HardwareFingerprint &&
c.MethodName == args.MethodName &&
c.IsMonitored);
if (channel == null) return;
double time = _stopwatch.Elapsed.TotalSeconds;
// 记录数据点（线程安全：Record 内部 ConcurrentQueue + Series 操作）
channel.Record(time, args.Value);
// 入队数据库批量写入
var entity = new MonitorValueEntity
{
MonitorName = channel.DisplayName,
MonitorValue = args.Value,
Scope = args.Scope,
CreateTime = DateTime.Now,
IsDel = 0
};
_insertQueue.Enqueue(entity);
if (_insertQueue.Count >= BulkInsertThreshold)
{
_ = Task.Run(async () => await DoFlushWorkAsync().ConfigureAwait(false));
}
}
#endregion
#region OxyPlot 视觉刷新定时器
private void OnPlotRefreshTick(object? sender, EventArgs e)
{
if (!_stopwatch.IsRunning) return;
double elapsed = _stopwatch.Elapsed.TotalSeconds;
var xAxis = Plot.Axes.FirstOrDefault(a => a.Position == AxisPosition.Bottom);
if (xAxis != null)
{
double window = 10000 * 0.1; // 20s 视窗
xAxis.Minimum = Math.Max(0, elapsed - window);
xAxis.Maximum = elapsed + 0.5;
}
Plot.InvalidatePlot(true);
}
#endregion
#region 后台数据库批量入库
private void StartDbFlushWorker(CancellationToken token)
{
_dbFlushTask = Task.Run(async () =>
{
// 只要未触发取消信号，就一直在后台默默轮询
while (!token.IsCancellationRequested)
{
try
{
// 💡 核心策略一：定时兜底。每隔 1 秒强制检查一次队列。
// ConfigureAwait(false) 彻底丢弃 UI 上下文，拥抱线程池极致速度。
await Task.Delay(TimeSpan.FromSeconds(1), token).ConfigureAwait(false);
// 批量提取并刷入数据库
await DoFlushWorkAsync().ConfigureAwait(false);
}
catch (OperationCanceledException)
{
break; // 正常收到退出信号，跳出循环
}
catch (OperationCanceledException) { break; }
catch (Exception ex)
{
System.Diagnostics.Debug.WriteLine($"[DB Worker Error] 后台写入数据库异常: {ex.Message}");
System.Diagnostics.Debug.WriteLine($"[DB Worker Error] {ex.Message}");
}
}
// 💡 核心策略二：临终遗言。当页面被关闭后，把队列里剩余的所有漏网之鱼彻底一次性洗干净。
await DoFlushWorkAsync().ConfigureAwait(false);
}, token);
}
/// <summary>
/// 纯粹的数据消费与批量 BulkCopy 入库
/// </summary>
private async Task DoFlushWorkAsync()
{
if (_insertQueue.IsEmpty) return;
// CAS 原子自增锁，确保同一时间只有一个线程在向 SqlSugar 投递这批数据
if (Interlocked.CompareExchange(ref _isFlushing, 1, 0) != 0) return;
try
{
var listToInsert = new List<MonitorValueEntity>();
// 一口气掏空当前并发队列里的所有实体
while (_insertQueue.TryDequeue(out var entity))
{
listToInsert.Add(entity);
}
while (_insertQueue.TryDequeue(out var entity)) listToInsert.Add(entity);
if (listToInsert.Count > 0)
{
// 调用你的 Service 层的 SqlSugar 批量写入
var result = await _monitorValueService.InsertRangeAsync(listToInsert).ConfigureAwait(false);
if (!result.IsSuccess)
{
System.Diagnostics.Debug.WriteLine($"[DB Bulk Error] 批量入库失败: {result.Msg}");
}
}
}
finally
{
Interlocked.Exchange(ref _isFlushing, 0); // 释放锁
}
finally { Interlocked.Exchange(ref _isFlushing, 0); }
}
#endregion
#region 采样定时器与高并发处理（100ms）
private async void OnSampleTick(object? sender, EventArgs e)
{
_sampleTime += _sampleInterval;
if (Channels.Count == 0 || _methodCache.Count == 0) return;
var sampleTasks = new List<Task>();
var token = _monitorCTS?.Token ?? CancellationToken.None;
// 1. 并行发起所有通道的数据采集
foreach (var channel in Channels)
{
if (!channel.IsMonitored) continue;
if (!_methodCache.TryGetValue(channel, out var entry)) continue;
sampleTasks.Add(Task.Run(async () =>
{
try
{
var (method, device) = entry;
var parmsInfo = method.GetParameters();
object?[]? invokeArgs = parmsInfo.Length == 1 && parmsInfo[0].ParameterType == typeof(CancellationToken)
? new object[] { token }
: null;
if (method.Invoke(device, invokeArgs) is Task<string> task)
{
// 硬件读取，彻底跑在后台线程池
string raw = await task.ConfigureAwait(false);
if (double.TryParse(raw, out double value))
{
// A. 更新本地图表数据缓存
channel.Record(_sampleTime, value);
// B. 封装实体
var entity = new MonitorValueEntity
{
MonitorName = channel.DisplayName,
MonitorValue = value,
CreateTime = DateTime.Now,
IsDel = 0
};
// C. 🟥 直接压入无锁队列，耗时极其微小，绝不卡硬件采集
_insertQueue.Enqueue(entity);
// D. 🟥 核心策略三：定量触发。如果采集数据爆棚，瞬间堆积超过50条，不等1秒定时器，直接开辟后台任务去写入
if (_insertQueue.Count >= BulkInsertThreshold)
{
_ = Task.Run(async () => await DoFlushWorkAsync().ConfigureAwait(false));
}
}
}
}
catch
{
// 某个硬件通道故障时不干扰其他通道
}
}, token));
}
// 2. 批量等待并发采样，加入 2 秒强制超时保护兜底
if (sampleTasks.Count > 0)
{
try
{
var delayTask = Task.Delay(TimeSpan.FromSeconds(2), token);
await Task.WhenAny(Task.WhenAll(sampleTasks), delayTask);
}
catch (OperationCanceledException)
{
return;
}
}
// 3. 【回归 UI 线程】执行 OxyPlot 坐标滚动与刷新
var xAxis = Plot.Axes.FirstOrDefault(a => a.Position == AxisPosition.Bottom);
if (xAxis != null)
{
double window = 200 * _sampleInterval;
xAxis.Minimum = Math.Max(0, _sampleTime - window);
xAxis.Maximum = _sampleTime + 0.5;
}
Plot.InvalidatePlot(true);
}
#endregion
#region Navigation 导航重写
#region Navigation 导航
public override void OnNavigatedTo(NavigationContext navigationContext)
{
base.OnNavigatedTo(navigationContext);
@@ -339,18 +270,28 @@ namespace MonitorModule.ViewModels
_scope = _globalInfo.ScopeDic[TestStatus];
_scopedContext = _scope.Resolve<ScopedContext>();
_deviceManager = _scope.Resolve<DeviceManager>();
_broadcaster = _scope.Resolve<HardwareDataBroadcaster>();
IsInitiated = true;
if (_monitorCTS == null) _monitorCTS = new CancellationTokenSource();
// 🟥 1. 先把后台数据库消费专线拉起来
// 1. 启动后台数据库消费
StartDbFlushWorker(_monitorCTS.Token);
// 2. 扫描可用硬件方法
// 2. 扫描可监测方法（供 UI 选择）
DiscoverAvailableMethods();
// 3. 开启 100ms 硬件采集
_sampleTimer.Start();
// 3. 启动广播器（Discover + Start）
_broadcaster.Discover();
_broadcaster.Start();
// 4. 订阅 HardwareDataReportedEvent（UI 线程回调）
_subscriptionToken = _eventAggregator.GetEvent<HardwareDataReportedEvent>()
.Subscribe(OnHardwareDataReceived, ThreadOption.UIThread);
// 5. 启动 OxyPlot 视觉刷新定时器
_stopwatch.Start();
_plotRefreshTimer.Start();
}
}
#endregion
@@ -388,7 +329,22 @@ namespace MonitorModule.ViewModels
}
#endregion
#region 设备方法发现
#region 设备方法发现（仅用于 UI 展示可选项）
/// <summary>
/// 通过 GlobalInfo.HardwarePool 构建 设备实例引用→硬件指纹 的反向查找表。
/// </summary>
private Dictionary<object, string> BuildInstanceToFingerprint()
{
var map = new Dictionary<object, string>(ReferenceEqualityComparer.Instance);
foreach (var poolEntry in _globalInfo.HardwarePool)
{
var lazy = poolEntry.Value;
if (lazy.IsValueCreated && lazy.Value != null)
map[lazy.Value] = poolEntry.Key;
}
return map;
}
private void DiscoverAvailableMethods()
{
AvailableMethods.Clear();
@@ -398,24 +354,26 @@ namespace MonitorModule.ViewModels
return;
}
var instanceToFp = BuildInstanceToFingerprint();
foreach (var kvp in _deviceManager.DeviceMap)
{
string deviceName = kvp.Key;
var device = kvp.Value;
var deviceType = device.GetType();
// 反查该实例的硬件指纹
if (!instanceToFp.TryGetValue(device, out string? fingerprint))
continue;
var methods = deviceType.GetMethods(BindingFlags.Public | BindingFlags.Instance)
.Where(m =>
{
var attribute = m.GetCustomAttribute<MonitorableAttribute>();
if (attribute == null) return false;
if (m.GetCustomAttribute<MonitorableAttribute>() == null) return false;
if (m.ReturnType != typeof(Task<string>)) return false;
var parms = m.GetParameters();
bool validParams = parms.Length == 0 ||
(parms.Length == 1 && parms[0].ParameterType == typeof(CancellationToken));
return validParams;
return parms.Length == 0 ||
(parms.Length == 1 && parms[0].ParameterType == typeof(CancellationToken));
});
foreach (var method in methods)
@@ -425,15 +383,15 @@ namespace MonitorModule.ViewModels
? $"{deviceName}.{attr.Description}"
: $"{deviceName}.{method.Name}";
var item = new AvailableMethodItem
AvailableMethods.Add(new AvailableMethodItem
{
DeviceName = deviceName,
Fingerprint = fingerprint,
MethodName = method.Name,
DisplayName = displayName,
MethodInfo = method,
Device = device
};
AvailableMethods.Add(item);
});
}
}
@@ -453,7 +411,8 @@ namespace MonitorModule.ViewModels
return;
}
if (Channels.Any(c => c.DeviceName == method.DeviceName && c.MethodName == method.MethodName))
// 以指纹+方法名去重，防止同一物理设备重复添加
if (Channels.Any(c => c.Fingerprint == method.Fingerprint && c.MethodName == method.MethodName))
{
StatusMessage = $"监测项 [{method.DisplayName}] 已存在";
return;
@@ -465,6 +424,7 @@ namespace MonitorModule.ViewModels
var channel = new MonitorChannel
{
DeviceName = method.DeviceName,
Fingerprint = method.Fingerprint,
MethodName = method.MethodName,
DisplayName = method.DisplayName,
Color = color,
@@ -480,8 +440,6 @@ namespace MonitorModule.ViewModels
};
Plot.Series.Add(channel.Series);
_methodCache[channel] = (method.MethodInfo, method.Device);
channel.PropertyChanged += (s, e) =>
{
if (e.PropertyName == nameof(MonitorChannel.IsDisplayed))
@@ -505,8 +463,6 @@ namespace MonitorModule.ViewModels
}
if (target.Series != null) Plot.Series.Remove(target.Series);
_methodCache.Remove(target);
Channels.Remove(target);
SelectedChannel = Channels.LastOrDefault();
@@ -550,11 +506,9 @@ namespace MonitorModule.ViewModels
};
var recent = channel.DataPoints.ToArray();
var startIdx = Math.Max(0, recent.Length - 200);
var startIdx = Math.Max(0, recent.Length - 10000);
for (int i = startIdx; i < recent.Length; i++)
{
channel.Series.Points.Add(new DataPoint(recent[i].Time, recent[i].DisplayValue));
}
Plot.Series.Add(channel.Series);
Plot.InvalidatePlot(true);
@@ -574,16 +528,13 @@ namespace MonitorModule.ViewModels
public void OnChannelMathChanged(MonitorChannel channel)
{
if (channel.Series == null) return;
var points = channel.DataPoints.ToArray();
channel.Series.Points.Clear();
var startIdx = Math.Max(0, points.Length - 200);
var startIdx = Math.Max(0, points.Length - 10000);
for (int i = startIdx; i < points.Length; i++)
{
channel.Series.Points.Add(new DataPoint(points[i].Time, points[i].DisplayValue));
}
Plot.InvalidatePlot(true);
}
#endregion
}
}
}