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

超时bug修改

Browse Source
This commit is contained in:
hsc
2026-06-24 16:49:39 +08:00
parent 4deb785135
commit 72f8b8c5b3
2 changed files with 174 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

125
DeviceCommand/Base/TCP.cs
View File

@@ -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
}
}
/// <summary>
/// 核心内部方法：无锁状态下安全重置并重建连接
/// </summary>
private async Task<bool> 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<byte>() : 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<byte>() : 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<string> ReadAsync(string delimiter = "\n", CancellationToken ct = default)
@@ -165,7 +217,6 @@ namespace DeviceCommand.Base
}
}
// 核心优化：确保发送与读取在同一组锁生命周期内
public async Task<string> WriteReadAsync(string command, string delimiter = "\n", CancellationToken ct = default)
{
await _commLock.WaitAsync(ct);