diff --git a/DeviceCommand/Base/Serial_Port.cs b/DeviceCommand/Base/Serial_Port.cs index 9b4e0ba..1d73cb0 100644 --- a/DeviceCommand/Base/Serial_Port.cs +++ b/DeviceCommand/Base/Serial_Port.cs @@ -1,5 +1,6 @@ using Model.Models; using System; +using System.IO; using System.IO.Ports; using System.Text; using System.Threading; @@ -7,7 +8,7 @@ using System.Threading.Tasks; namespace DeviceCommand.Base { - public class Serial_Port : ISerialPort + public class Serial_Port : ISerialPort, IDisposable { public string PortName { get; set; } = "COM1"; public int BaudRate { get; set; } = 9600; @@ -58,8 +59,10 @@ namespace DeviceCommand.Base _serialPort.DataBits = DataBits; _serialPort.StopBits = StopBits; _serialPort.Parity = Parity; - _serialPort.ReadTimeout = ReadTimeout; - _serialPort.WriteTimeout = WriteTimeout; + + // 允许底层保留默认设置，控制权交给外层 WaitAsync + _serialPort.ReadTimeout = SerialPort.InfiniteTimeout; + _serialPort.WriteTimeout = SerialPort.InfiniteTimeout; _serialPort.Open(); return true; @@ -72,19 +75,33 @@ namespace DeviceCommand.Base public virtual void Close() { - if (_serialPort.IsOpen) _serialPort.Close(); + if (_serialPort?.IsOpen == true) _serialPort.Close(); } - // 内部无锁发送方法，供原子组合操作调用 private async Task LoglessSendAsync(string data, CancellationToken ct) { if (!_serialPort.IsOpen) throw new InvalidOperationException("串口未打开。"); byte[] bytes = Encoding.UTF8.GetBytes(data); - using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct); - if (WriteTimeout > 0) cts.CancelAfter(WriteTimeout); - await _serialPort.BaseStream.WriteAsync(bytes, 0, bytes.Length, cts.Token); + try + { + if (WriteTimeout > 0) + { + // 核心修改：使用 WaitAsync 精准控制写入超时 + await _serialPort.BaseStream.WriteAsync(bytes, 0, bytes.Length, ct) + .WaitAsync(TimeSpan.FromMilliseconds(WriteTimeout), ct) + .ConfigureAwait(false); + } + else + { + await _serialPort.BaseStream.WriteAsync(bytes, 0, bytes.Length, ct).ConfigureAwait(false); + } + } + catch (TimeoutException ex) + { + throw new TimeoutException($"串口写入数据超时（{WriteTimeout} ms）", ex); + } } public async Task SendAsync(string data, CancellationToken ct = default) @@ -100,7 +117,6 @@ namespace DeviceCommand.Base } } - // 内部无锁读取方法，利用 BaseStream 挂起线程，高性能不吃 CPU private async Task LoglessReadAsync(string delimiter, CancellationToken ct) { if (!_serialPort.IsOpen) throw new InvalidOperationException("串口未打开。"); @@ -109,24 +125,44 @@ namespace DeviceCommand.Base var sb = new StringBuilder(); byte[] buffer = new byte[1024]; - using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct); - if (ReadTimeout > 0) cts.CancelAfter(ReadTimeout); - - while (!cts.Token.IsCancellationRequested) + try { - // 核心优化：利用流异步挂起，替代原先的 BytesToRead 循环延时 - int bytesRead = await _serialPort.BaseStream.ReadAsync(buffer, 0, buffer.Length, cts.Token); - if (bytesRead == 0) continue; - - sb.Append(Encoding.UTF8.GetString(buffer, 0, bytesRead)); - - int index = sb.ToString().IndexOf(delimiter, StringComparison.Ordinal); - if (index >= 0) + while (true) { - return sb.ToString(0, index).Trim(); + ct.ThrowIfCancellationRequested(); + + int bytesRead; + if (ReadTimeout > 0) + { + // 核心修改：使用 WaitAsync 精准控制单次异步读取超时 + bytesRead = await _serialPort.BaseStream.ReadAsync(buffer, 0, buffer.Length, ct) + .WaitAsync(TimeSpan.FromMilliseconds(ReadTimeout), ct) + .ConfigureAwait(false); + } + else + { + bytesRead = await _serialPort.BaseStream.ReadAsync(buffer, 0, buffer.Length, ct).ConfigureAwait(false); + } + + if (bytesRead == 0) continue; + + sb.Append(Encoding.UTF8.GetString(buffer, 0, bytesRead)); + + string currentText = sb.ToString(); + int index = currentText.IndexOf(delimiter, StringComparison.Ordinal); + if (index >= 0) + { + return currentText.Substring(0, index).Trim(); + } } } - throw new TimeoutException("读取数据超时"); + catch (TimeoutException ex) + { + // 关键防污染策略：串口发生命令错、超时不响应时，WaitAsync 会立功捕获异常 + // 我们直接清空串口驱动内部的输入缓冲区，把残余垃圾数据物理抹除，确保下一条指令安全、不掉线 + ClearHardwareBuffer(); + throw new TimeoutException($"串口读取超时（未收到结束符 '{delimiter}'），等待时间：{ReadTimeout} ms", ex); + } } public async Task ReadAsync(string delimiter = "\n", CancellationToken ct = default) @@ -142,12 +178,14 @@ namespace DeviceCommand.Base } } - // 核心优化：保证多线程环境下发送和等待回包是一个原子过程 public async Task WriteReadAsync(string command, string delimiter = "\n", CancellationToken ct = default) { await commLock.WaitAsync(ct); try { + // 发送新命令前先冲洗一下缓冲区，双重保险 + ClearHardwareBuffer(); + await LoglessSendAsync(command, ct); return await LoglessReadAsync(delimiter, ct); } @@ -157,9 +195,32 @@ namespace DeviceCommand.Base } } + /// + /// 物理强制清空串口驱动层及软件层的缓冲区 + /// + private void ClearHardwareBuffer() + { + try + { + if (_serialPort?.IsOpen == true) + { + _serialPort.DiscardInBuffer(); // 清空硬件接收缓冲区 + _serialPort.DiscardOutBuffer(); // 清空硬件发送缓冲区 + } + } + catch + { + // 忽略清理异常 + } + } + public void Dispose() { - _serialPort?.Dispose(); + if (_serialPort != null) + { + if (_serialPort.IsOpen) _serialPort.Close(); + _serialPort.Dispose(); + } commLock?.Dispose(); } } diff --git a/DeviceCommand/Base/TCP.cs b/DeviceCommand/Base/TCP.cs index 0f7d067..c064410 100644 --- a/DeviceCommand/Base/TCP.cs +++ b/DeviceCommand/Base/TCP.cs @@ -8,7 +8,7 @@ using System.Threading.Tasks; namespace DeviceCommand.Base { - public class Tcp : ITcp + public class Tcp : ITcp, IDisposable { public string IPAddress { get; set; } = "127.0.0.1"; public int Port { get; set; } = 502; @@ -46,15 +46,7 @@ namespace DeviceCommand.Base await _commLock.WaitAsync(ct); try { - if (_tcpClient.Connected) return true; - - // 修复：释放并彻底清空旧的连接实例，否则复用引发异常 - _tcpClient.Close(); - _tcpClient.Dispose(); - - _tcpClient = new TcpClient(); - await _tcpClient.ConnectAsync(IPAddress, Port, ct); - return true; + return await ResetConnectionAsync(ct); } finally { @@ -62,9 +54,25 @@ namespace DeviceCommand.Base } } + /// + /// 核心内部方法：无锁状态下安全重置并重建连接 + /// + private async Task ResetConnectionAsync(CancellationToken ct) + { + if (_tcpClient != null) + { + _tcpClient.Close(); + _tcpClient.Dispose(); + } + + _tcpClient = new TcpClient(); + await _tcpClient.ConnectAsync(IPAddress, Port, ct).ConfigureAwait(false); + return true; + } + public virtual void Close() { - if (_tcpClient.Connected) _tcpClient.Close(); + if (_tcpClient?.Connected == true) _tcpClient.Close(); } private async Task LoglessSendAsync(byte[] buffer, CancellationToken ct) @@ -72,10 +80,26 @@ namespace DeviceCommand.Base if (!IsConnected) throw new InvalidOperationException("TCP未连接。"); NetworkStream stream = _tcpClient.GetStream(); - using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct); - if (SendTimeout > 0) cts.CancelAfter(SendTimeout); - - await stream.WriteAsync(buffer, 0, buffer.Length, cts.Token); + try + { + if (SendTimeout > 0) + { + // 使用 WaitAsync 控制发送超时 + await stream.WriteAsync(buffer, 0, buffer.Length, ct) + .WaitAsync(TimeSpan.FromMilliseconds(SendTimeout), ct) + .ConfigureAwait(false); + } + else + { + await stream.WriteAsync(buffer, 0, buffer.Length, ct).ConfigureAwait(false); + } + } + catch (TimeoutException ex) + { + // 发送超时，安全起见直接断开重建 + await ResetConnectionAsync(ct); + throw new TimeoutException($"TCP 发送数据超时（{SendTimeout} ms），连接已重置以确保安全。", ex); + } } public async Task SendAsync(byte[] buffer, CancellationToken ct = default) @@ -107,17 +131,30 @@ namespace DeviceCommand.Base byte[] buffer = new byte[length]; int offset = 0; - using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct); - if (ReceiveTimeout > 0) cts.CancelAfter(ReceiveTimeout); - - while (offset < length) + try { - int read = await stream.ReadAsync(buffer, offset, length - offset, cts.Token); - if (read == 0) break; - offset += read; - } + while (offset < length) + { + ct.ThrowIfCancellationRequested(); - return offset == 0 ? Array.Empty() : buffer[..offset]; + // 计算剩余超时时间（这里简单使用配置值，若要求极精准可加入不长计算） + int read = ReceiveTimeout > 0 + ? await stream.ReadAsync(buffer, offset, length - offset, ct) + .WaitAsync(TimeSpan.FromMilliseconds(ReceiveTimeout), ct) + .ConfigureAwait(false) + : await stream.ReadAsync(buffer, offset, length - offset, ct).ConfigureAwait(false); + + if (read == 0) throw new IOException("远程主机已关闭连接"); + offset += read; + } + return offset == 0 ? Array.Empty() : buffer[..offset]; + } + catch (TimeoutException ex) + { + // 关键安全重置：WaitAsync 超时后抛弃老连接，防止未完成的 Read 污染后续 Buffer + await ResetConnectionAsync(ct); + throw new TimeoutException($"TCP 读取定长数据超时（{ReceiveTimeout} ms），连接已重置避免数据错乱。", ex); + } } finally { @@ -134,22 +171,37 @@ namespace DeviceCommand.Base byte[] buffer = new byte[1024]; NetworkStream stream = _tcpClient.GetStream(); - using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct); - if (ReceiveTimeout > 0) cts.CancelAfter(ReceiveTimeout); - - while (!cts.Token.IsCancellationRequested) + try { - int bytesRead = await stream.ReadAsync(buffer, 0, buffer.Length, cts.Token); - if (bytesRead == 0) throw new IOException("远程主机已关闭连接"); + while (true) + { + ct.ThrowIfCancellationRequested(); - sb.Append(Encoding.UTF8.GetString(buffer, 0, bytesRead)); + int bytesRead = ReceiveTimeout > 0 + ? await stream.ReadAsync(buffer, 0, buffer.Length, ct) + .WaitAsync(TimeSpan.FromMilliseconds(ReceiveTimeout), ct) + .ConfigureAwait(false) + : await stream.ReadAsync(buffer, 0, buffer.Length, ct).ConfigureAwait(false); - int index = sb.ToString().IndexOf(delimiter, StringComparison.Ordinal); - if (index >= 0) - return sb.ToString(0, index).Trim(); + if (bytesRead == 0) throw new IOException("远程主机已关闭连接"); + + sb.Append(Encoding.UTF8.GetString(buffer, 0, bytesRead)); + + string currentText = sb.ToString(); + int index = currentText.IndexOf(delimiter, StringComparison.Ordinal); + if (index >= 0) + { + return currentText.Substring(0, index).Trim(); + } + } + } + catch (TimeoutException ex) + { + // 当 SCPI 仪表由于命令发错不响应时，WaitAsync 会立功并抛出 TimeoutException + // 此时我们在内部秒速“重置连接”，上层业务只会收到一个干净的超时报错，但由于锁和自动重连，下一次通信直接恢复！ + await ResetConnectionAsync(ct); + throw new TimeoutException($"SCPI 读取超时（未收到结束符 '{delimiter}'，等待：{ReceiveTimeout} ms），通信链路已自动刷新。", ex); } - - throw new TimeoutException("读取数据超时"); } public async Task ReadAsync(string delimiter = "\n", CancellationToken ct = default) @@ -165,7 +217,6 @@ namespace DeviceCommand.Base } } - // 核心优化：确保发送与读取在同一组锁生命周期内 public async Task WriteReadAsync(string command, string delimiter = "\n", CancellationToken ct = default) { await _commLock.WaitAsync(ct);