feat: 移植DeviceHealthMonitor心跳重连机制并集成到DeviceManager

This commit is contained in:
2026-09-14 13:46:45 +08:00
parent c4fc2d80fd
commit 31255fadeb
2 changed files with 279 additions and 0 deletions
@@ -0,0 +1,255 @@
using DeviceCommand.Base;
using Logger;
using Model.Models;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using System.Windows;
namespace UIShare.GlobalVariable
{
/// <summary>
/// 设备健康监控器:独立于监控采样的心跳重连机制。
/// <para>
/// 设计原则——与检测值零冲突:
/// <list type="bullet">
/// <item>健康检查平时只读 <see cref="IBaseInterface.IsConnected"/>(纯本地属性,零网络开销,不碰通信锁)</item>
/// <item>仅在 IsConnected==false 时才获取 _commLock 执行重连,与监控采样天然串行化</item>
/// <item>ModbusTcp 的 ConnectAsync 是幂等的(已连接时直接返回),不会中断正在进行的监控</item>
/// <item>TCP 的 ConnectAsync 会重置连接,但监控的 catch 容忍单次失败,下次 tick 自动恢复</item>
/// </list>
/// </para>
/// </summary>
public class DeviceHealthMonitor : IDisposable
{
private readonly IDictionary<string, IBaseInterface> _deviceMap;
private readonly SystemConfig _systemConfig;
private readonly string _scopeName;
/// <summary>健康检查定时器</summary>
private Timer? _healthCheckTimer;
/// <summary>每个设备的连续失败计数</summary>
private readonly ConcurrentDictionary<string, int> _failureCounts = new();
/// <summary>每个设备的重连尝试次数(用于计算退避阈值)</summary>
private readonly ConcurrentDictionary<string, int> _reconnectAttempts = new();
/// <summary>基础失败阈值(首次重连触发值)</summary>
private const int BaseFailureThreshold = 3;
/// <summary>退避上限(最大阈值)</summary>
private const int MaxBackoffThreshold = 15;
/// <summary>健康检查间隔(毫秒)</summary>
private readonly int _checkIntervalMs;
private bool _disposed;
private readonly object _startStopLock = new();
/// <summary>
/// 创建健康监控器实例。
/// </summary>
/// <param name="deviceMap">当前作用域的设备字典</param>
/// <param name="systemConfig">当前作用域的系统配置(用于更新 DeviceInfoVM.IsConnected</param>
/// <param name="scopeName">作用域名称(日志标识)</param>
/// <param name="checkIntervalMs">健康检查间隔,默认 5000ms</param>
public DeviceHealthMonitor(
IDictionary<string, IBaseInterface> deviceMap,
SystemConfig systemConfig,
string scopeName,
int checkIntervalMs = 5000)
{
_deviceMap = deviceMap;
_systemConfig = systemConfig;
_scopeName = scopeName;
_checkIntervalMs = checkIntervalMs;
}
/// <summary>启动健康监控(幂等:多次调用只启动一次)</summary>
public void Start()
{
lock (_startStopLock)
{
if (_disposed || _healthCheckTimer != null) return;
_healthCheckTimer = new Timer(
OnHealthCheckTick,
null,
TimeSpan.FromSeconds(10), // 首次检查延迟 10 秒,避免启动时设备尚未连接完成
TimeSpan.FromMilliseconds(_checkIntervalMs));
LoggerHelper.Info($"[{_scopeName}] 设备健康监控已启动,检查间隔={_checkIntervalMs}ms,基础重连阈值={BaseFailureThreshold}次(指数退避上限={MaxBackoffThreshold}");
}
}
/// <summary>停止健康监控</summary>
public void Stop()
{
lock (_startStopLock)
{
_healthCheckTimer?.Change(Timeout.Infinite, Timeout.Infinite);
_healthCheckTimer?.Dispose();
_healthCheckTimer = null;
_failureCounts.Clear();
_reconnectAttempts.Clear();
}
}
/// <summary>
/// 健康检查核心逻辑:遍历设备,检查连接状态,失败计数超阈值则重连。
/// <para>TCP 设备额外进行主动探活(*IDN?),以检测死连接(对端崩溃但 TCP 未收到 FIN)。</para>
/// </summary>
private async void OnHealthCheckTick(object? state)
{
if (_disposed || _deviceMap.Count == 0) return;
// 快照避免枚举期间字典被修改
var snapshot = _deviceMap.ToArray();
foreach (var kvp in snapshot)
{
if (_disposed) return;
string deviceName = kvp.Key;
var device = kvp.Value;
try
{
bool alive = device.IsConnected;
// TCP 设备主动探活:IsConnected 只反映上次操作状态,无法检测死连接
if (alive && device is Tcp tcpDevice)
{
alive = await ProbeTcpDeviceAsync(tcpDevice);
}
if (alive)
{
// 连接正常:清零失败计数与退避
_failureCounts.TryRemove(deviceName, out _);
_reconnectAttempts.TryRemove(deviceName, out _);
continue;
}
// 连接断开:累加失败计数
int failures = _failureCounts.AddOrUpdate(deviceName, 1, (_, count) => count + 1);
// 计算当前退避阈值:基础值 + 重连尝试次数 × 2,上限为 MaxBackoffThreshold
int attempts = _reconnectAttempts.GetOrAdd(deviceName, 0);
int currentThreshold = Math.Min(BaseFailureThreshold + attempts * 2, MaxBackoffThreshold);
if (failures < currentThreshold)
{
LoggerHelper.Warn(
$"[{_scopeName}] 设备 [{deviceName}] 连接断开,等待重连中 ({failures}/{currentThreshold})");
continue;
}
// 达到阈值:执行重连
await ReconnectDeviceAsync(deviceName, device);
}
catch (Exception ex)
{
LoggerHelper.Error($"[{_scopeName}] 设备 [{deviceName}] 健康检查异常:{ex.Message}");
}
}
}
/// <summary>
/// 对 TCP 设备执行轻量级主动探活(SCPI *IDN?),2 秒超时。
/// <para>通过 WriteReadAsync 内部获取 _commLock,与监控采样天然串行化。</para>
/// </summary>
/// <returns>true: 探活成功(连接确实存活); false: 探活失败(死连接)</returns>
private async Task<bool> ProbeTcpDeviceAsync(Tcp tcpDevice)
{
try
{
using var probeCts = new CancellationTokenSource(TimeSpan.FromSeconds(2));
string resp = await tcpDevice.WriteReadAsync("*IDN?\n", "\n", probeCts.Token);
return !string.IsNullOrWhiteSpace(resp);
}
catch
{
return false;
}
}
/// <summary>
/// 重连单个设备。
/// <para>
/// 冲突避免机制:ConnectAsync 内部获取设备的 _commLock
/// 如果此时监控采样正在通信,重连会等待锁释放后再执行,
/// 保证同一时刻只有一个操作在使用通信链路。
/// </para>
/// </summary>
private async Task ReconnectDeviceAsync(string deviceName, IBaseInterface device)
{
try
{
int attempts = _reconnectAttempts.AddOrUpdate(deviceName, 1, (_, c) => c + 1);
int nextThreshold = Math.Min(BaseFailureThreshold + attempts * 2, MaxBackoffThreshold);
LoggerHelper.Info($"[{_scopeName}] 设备 [{deviceName}] 连续失败触发重连(第 {attempts} 次重连,下次阈值={nextThreshold}...");
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
bool ok = await device.ConnectAsync(cts.Token);
// 同步更新 DeviceInfoVM 的 UI 状态
UpdateDeviceInfoState(deviceName, ok);
if (ok)
{
_failureCounts.TryRemove(deviceName, out _);
_reconnectAttempts.TryRemove(deviceName, out _);
LoggerHelper.Info($"[{_scopeName}] 设备 [{deviceName}] 重连成功");
}
else
{
LoggerHelper.Warn($"[{_scopeName}] 设备 [{deviceName}] 重连失败,将在下次检查时重试");
}
}
catch (OperationCanceledException)
{
UpdateDeviceInfoState(deviceName, false);
LoggerHelper.Warn($"[{_scopeName}] 设备 [{deviceName}] 重连超时(10s),将在下次检查时重试");
}
catch (Exception ex)
{
UpdateDeviceInfoState(deviceName, false);
LoggerHelper.Error($"[{_scopeName}] 设备 [{deviceName}] 重连异常:{ex.Message}");
}
}
/// <summary>
/// 通过 UI 线程更新 SystemConfig.DeviceList 中对应设备的 IsConnected 状态(驱动 UI 刷新)。
/// <para>Timer 回调运行在线程池线程上,必须切回 UI 线程才能触发 PropertyChanged。</para>
/// </summary>
private void UpdateDeviceInfoState(string deviceName, bool isConnected)
{
var info = _systemConfig?.DeviceList?
.FirstOrDefault(d => d != null &&
string.Equals(d.DeviceName, deviceName, StringComparison.OrdinalIgnoreCase));
if (info == null) return;
var dispatcher = Application.Current?.Dispatcher;
if (dispatcher == null || dispatcher.CheckAccess())
{
// 已在 UI 线程(或无 Dispatcher),直接赋值
info.IsConnected = isConnected;
}
else
{
// 切回 UI 线程赋值,避免跨线程 PropertyChanged 异常
dispatcher.BeginInvoke(() => info.IsConnected = isConnected);
}
}
public void Dispose()
{
if (_disposed) return;
_disposed = true;
Stop();
}
}
}
+24
View File
@@ -28,6 +28,9 @@ namespace UIShare.GlobalVariable
private readonly string _scopeName;
private readonly IEventAggregator _eventAggregator;
/// <summary>设备健康监控器:独立心跳检测 + 自动重连,与监控采样互不冲突</summary>
private DeviceHealthMonitor? _healthMonitor;
/// <summary>按 DeviceName 索引的设备字典,便于业务层按名取实例。</summary>
public IDictionary<string, IBaseInterface> DeviceMap { get; private set; }
= new Dictionary<string, IBaseInterface>(StringComparer.OrdinalIgnoreCase);
@@ -252,6 +255,9 @@ namespace UIShare.GlobalVariable
}
await Task.WhenAll(tasks);
// 所有设备连接完成后,启动健康监控(心跳重连)
StartHealthMonitor();
}
@@ -320,6 +326,7 @@ namespace UIShare.GlobalVariable
/// </summary>
public async Task CloseAllDevicesAsync()
{
StopHealthMonitor();
List<Task> tasks = new List<Task>();
lock (_lockObj)
@@ -617,8 +624,25 @@ namespace UIShare.GlobalVariable
return Activator.CreateInstance(type, cfg) as IBaseInterface;
}
/// <summary>启动设备健康监控(心跳检测 + 自动重连)</summary>
private void StartHealthMonitor()
{
StopHealthMonitor();
_healthMonitor = new DeviceHealthMonitor(DeviceMap, _systemConfig, _scopeName);
_healthMonitor.Start();
}
/// <summary>停止设备健康监控</summary>
private void StopHealthMonitor()
{
_healthMonitor?.Stop();
_healthMonitor?.Dispose();
_healthMonitor = null;
}
public void Dispose()
{
StopHealthMonitor();
if (string.IsNullOrEmpty(_scopeName)) return;
// 遍历当前作用域用到的所有指纹,逐一移除本作用域的引用