277 lines
12 KiB
C#
277 lines
12 KiB
C#
using Common.Attributes;
|
||
using DeviceCommand.Base;
|
||
using Logger;
|
||
using Model.Models;
|
||
using Prism.Events;
|
||
using System;
|
||
using System.Collections.Concurrent;
|
||
using System.Collections.Generic;
|
||
using System.Linq;
|
||
using System.Linq.Expressions;
|
||
using System.Reflection;
|
||
using System.Threading;
|
||
using System.Threading.Tasks;
|
||
using System.Windows.Threading;
|
||
using UIShare.PubEvent;
|
||
|
||
namespace UIShare.GlobalVariable
|
||
{
|
||
/// <summary>
|
||
/// 硬件数据广播器（中间层，全局单例）：
|
||
/// 直接基于 GlobalInfo.HardwarePool 中的设备指纹字典进行统一采样，
|
||
/// 每个物理设备只采样一次，然后向所有引用该设备的作用域分别广播 HardwareDataReportedEvent。
|
||
/// 解决多作用域共享同一物理设备时的重复采样问题。
|
||
/// DeviceCommand 层无需引用 Prism，完全解耦。
|
||
/// </summary>
|
||
public class HardwareDataBroadcaster : IDisposable
|
||
{
|
||
private readonly IEventAggregator _eventAggregator;
|
||
private readonly GlobalInfo _globalInfo;
|
||
|
||
/// <summary>采样定时器</summary>
|
||
private readonly DispatcherTimer _sampleTimer;
|
||
|
||
/// <summary>
|
||
/// 已注册的采样项：(Fingerprint, MethodName, 编译后的委托, DeviceInstance)
|
||
/// 委托签名统一为 Func<object, CancellationToken, Task<double>>
|
||
/// - 第一个参数 = 设备实例（open delegate 风格，编译时已做类型转换）
|
||
/// - 第二个参数 = CancellationToken（无参方法会忽略它）
|
||
/// - 返回值 = Task<double>（Task<string> 方法自动 TryParse 为 double）
|
||
/// </summary>
|
||
private readonly List<(string Fingerprint, string MethodName, Func<object, CancellationToken, Task<double>> Invoker, object Device)> _registeredMethods = new();
|
||
|
||
private CancellationTokenSource? _cts;
|
||
private bool _disposed;
|
||
|
||
/// <summary>是否已完成过至少一次 Discover（幂等标记，避免多作用域重复调用时 Clear + 重扫）</summary>
|
||
private bool _discovered;
|
||
|
||
/// <summary>连续失败次数达到此阈值后，暂停该通道采样并发布报警</summary>
|
||
private const int FailureThreshold = 5;
|
||
|
||
/// <summary>每通道连续失败计数（key = fingerprint + "|" + methodName）</summary>
|
||
private readonly ConcurrentDictionary<string, int> _failureCounts = new();
|
||
|
||
/// <summary>已因连续失败而被暂停的通道（key 同上）</summary>
|
||
private readonly ConcurrentDictionary<string, bool> _suspendedChannels = new();
|
||
|
||
/// <summary>采样间隔（默认 1000ms）</summary>
|
||
public TimeSpan SampleInterval
|
||
{
|
||
get => _sampleTimer.Interval;
|
||
set => _sampleTimer.Interval = value;
|
||
}
|
||
|
||
public HardwareDataBroadcaster(GlobalInfo globalInfo, IEventAggregator eventAggregator)
|
||
{
|
||
_globalInfo = globalInfo;
|
||
_eventAggregator = eventAggregator;
|
||
|
||
_sampleTimer = new DispatcherTimer(DispatcherPriority.Normal)
|
||
{
|
||
Interval = TimeSpan.FromMilliseconds(1000)
|
||
};
|
||
_sampleTimer.Tick += OnSampleTick;
|
||
}
|
||
|
||
/// <summary>
|
||
/// 将 MethodInfo 编译为强类型委托，后续调用零反射开销。
|
||
/// 统一签名为 Func<object, CancellationToken, Task<double>>：
|
||
/// - Task<double> 方法：直接返回 double 值
|
||
/// - Task<string> 方法：自动 TryParse 为 double
|
||
/// - 无参方法：忽略 CancellationToken
|
||
/// - 带 CancellationToken 方法：直接透传
|
||
/// </summary>
|
||
private static Func<object, CancellationToken, Task<double>> BuildInvoker(MethodInfo method)
|
||
{
|
||
var deviceParam = Expression.Parameter(typeof(object), "device");
|
||
var ctParam = Expression.Parameter(typeof(CancellationToken), "ct");
|
||
|
||
// 将 object 转型为方法的声明类型
|
||
var castDevice = Expression.Convert(deviceParam, method.DeclaringType!);
|
||
|
||
// 根据方法签名决定传参
|
||
var parms = method.GetParameters();
|
||
Expression[] callArgs = parms.Length == 1
|
||
? new Expression[] { ctParam }
|
||
: Array.Empty<Expression>();
|
||
|
||
var call = Expression.Call(castDevice, method, callArgs);
|
||
|
||
if (method.ReturnType == typeof(Task<double>))
|
||
{
|
||
// Task<double> 方法：直接返回
|
||
return Expression.Lambda<Func<object, CancellationToken, Task<double>>>(
|
||
call, deviceParam, ctParam).Compile();
|
||
}
|
||
else
|
||
{
|
||
// Task<string> 方法：await 后 TryParse 为 double
|
||
return async (device, ct) =>
|
||
{
|
||
// 通过 MethodInfo.Invoke 调用（仅 Task<string> 路径，性能可接受）
|
||
var task = (Task<string>)method.Invoke(device,
|
||
parms.Length == 1 ? new object[] { ct } : Array.Empty<object>())!;
|
||
string raw = await task.ConfigureAwait(false);
|
||
return double.TryParse(raw, out double v) ? v : double.NaN;
|
||
};
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// 扫描 GlobalInfo.HardwarePool 中所有已创建的物理设备，
|
||
/// 反查可监测方法并编译为委托。每个指纹只注册一次，反射只执行一次。
|
||
/// <para>幂等：首次调用后再次调用不会 Clear 重扫，避免多作用域重复 Discover 导致短暂采样中断。</para>
|
||
/// </summary>
|
||
public void Discover()
|
||
{
|
||
if (_discovered) return;
|
||
_discovered = true;
|
||
_registeredMethods.Clear();
|
||
|
||
foreach (var poolEntry in _globalInfo.HardwarePool)
|
||
{
|
||
string fingerprint = poolEntry.Key;
|
||
var lazy = poolEntry.Value;
|
||
|
||
// 只监控已经实例化的设备
|
||
if (!lazy.IsValueCreated || lazy.Value == null) continue;
|
||
|
||
var device = lazy.Value;
|
||
var deviceType = device.GetType();
|
||
|
||
var methods = deviceType.GetMethods(BindingFlags.Public | BindingFlags.Instance)
|
||
.Where(m =>
|
||
{
|
||
if (m.GetCustomAttribute<MonitorableAttribute>() == null) return false;
|
||
// 同时支持 Task<string> 和 Task<double> 返回类型
|
||
var rt = m.ReturnType;
|
||
if (rt != typeof(Task<string>) && rt != typeof(Task<double>)) return false;
|
||
var parms = m.GetParameters();
|
||
return parms.Length == 0 ||
|
||
(parms.Length == 1 && parms[0].ParameterType == typeof(CancellationToken));
|
||
});
|
||
|
||
foreach (var method in methods)
|
||
{
|
||
var invoker = BuildInvoker(method);
|
||
_registeredMethods.Add((fingerprint, method.Name, invoker, device));
|
||
}
|
||
}
|
||
}
|
||
|
||
/// <summary>获取指定硬件指纹当前被哪些作用域引用</summary>
|
||
private List<string> GetScopesForFingerprint(string fingerprint)
|
||
{
|
||
if (_globalInfo.DeviceAndScopeDic.TryGetValue(fingerprint, out var lazy))
|
||
{
|
||
var scopeList = lazy.Value;
|
||
lock (scopeList) return scopeList.ToList();
|
||
}
|
||
return new List<string>();
|
||
}
|
||
|
||
/// <summary>启动采样广播（幂等）</summary>
|
||
public void Start()
|
||
{
|
||
if (_disposed) return;
|
||
_cts ??= new CancellationTokenSource();
|
||
if (!_sampleTimer.IsEnabled) _sampleTimer.Start();
|
||
}
|
||
|
||
/// <summary>停止采样</summary>
|
||
public void Stop()
|
||
{
|
||
_sampleTimer.Stop();
|
||
}
|
||
|
||
private void OnSampleTick(object? sender, EventArgs e)
|
||
{
|
||
if (_registeredMethods.Count == 0 || _disposed) return;
|
||
|
||
var token = _cts?.Token ?? CancellationToken.None;
|
||
var now = DateTime.Now;
|
||
|
||
foreach (var entry in _registeredMethods)
|
||
{
|
||
string channelKey = entry.Fingerprint + "|" + entry.MethodName;
|
||
|
||
// 已暂停的通道跳过采样
|
||
if (_suspendedChannels.ContainsKey(channelKey)) continue;
|
||
|
||
// fire-and-forget：每个通道独立采样，完成后自行广播
|
||
_ = Task.Run(async () =>
|
||
{
|
||
try
|
||
{
|
||
// 直接委托调用，零反射开销（Task<double> 路径）或自动 TryParse（Task<string> 路径）
|
||
double value = await entry.Invoker(entry.Device, token).ConfigureAwait(false);
|
||
|
||
if (double.IsNaN(value)) return;
|
||
|
||
// 采样成功，重置失败计数
|
||
_failureCounts.TryRemove(channelKey, out _);
|
||
|
||
// 向所有引用该物理设备的作用域分别广播
|
||
var scopes = GetScopesForFingerprint(entry.Fingerprint);
|
||
foreach (var scope in scopes)
|
||
{
|
||
_eventAggregator.GetEvent<HardwareDataReportedEvent>().Publish(new HardwareReportArgs
|
||
{
|
||
Scope = scope,
|
||
HardwareFingerprint = entry.Fingerprint,
|
||
MethodName = entry.MethodName,
|
||
Value = value,
|
||
Time = now
|
||
});
|
||
|
||
// 同步检查该作用域 ValueLimitList 是否超限
|
||
string MonitorStatus= ValueLimitAlarmHelper.CheckAlarm(scope, entry.Fingerprint, entry.MethodName, value, _globalInfo);
|
||
if (MonitorStatus != "" && MonitorStatus != "未报警")
|
||
{
|
||
_eventAggregator.GetEvent<AlarmEvent>().Publish((scope,entry.Fingerprint, MonitorStatus));
|
||
}
|
||
}
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
int count = _failureCounts.AddOrUpdate(channelKey, 1, (_, c) => c + 1);
|
||
LoggerHelper.Warn($"采样通道 [{entry.Fingerprint}/{entry.MethodName}] 第 {count} 次失败: {ex.Message}");
|
||
|
||
if (count >= FailureThreshold)
|
||
{
|
||
_suspendedChannels.TryAdd(channelKey, true);
|
||
LoggerHelper.Error($"采样通道 [{entry.Fingerprint}/{entry.MethodName}] 连续失败 {count} 次，已暂停采样");
|
||
|
||
// 向所有引用该设备的作用域发布报警
|
||
var scopes = GetScopesForFingerprint(entry.Fingerprint);
|
||
foreach (var scope in scopes)
|
||
{
|
||
_eventAggregator.GetEvent<AlarmEvent>().Publish(
|
||
(scope, entry.Fingerprint, $"通道 {entry.MethodName} 连续失败 {count} 次，已暂停"));
|
||
}
|
||
}
|
||
}
|
||
}, token);
|
||
}
|
||
|
||
// 不在 UI 线程 await——任务在线程池上自行完成并广播事件，
|
||
// DispatcherTimer 按设定间隔准时触发下一次 Tick。
|
||
}
|
||
|
||
public void Dispose()
|
||
{
|
||
if (_disposed) return;
|
||
_disposed = true;
|
||
_sampleTimer.Stop();
|
||
_sampleTimer.Tick -= OnSampleTick;
|
||
try { _cts?.Cancel(); } catch { }
|
||
_cts?.Dispose();
|
||
_cts = null;
|
||
_registeredMethods.Clear();
|
||
_failureCounts.Clear();
|
||
_suspendedChannels.Clear();
|
||
}
|
||
}
|
||
}
|