Skip to content

T
Thick-Common
Commit d4e889dc authored by 潘栩锋's avatar 潘栩锋 🚴
Browse files
Options

修复 tcpclient 连接断开时的线程异常问题

parent e6c70111
Hide whitespace changes
Inline Side-by-side
Showing with 52 additions and 364 deletions
GeneralGommunication/GComm_TcpClient.cs
View file @ d4e889dc
...@@ -10,354 +10,6 @@ using System.Threading.Tasks; ...@@ -10,354 +10,6 @@ using System.Threading.Tasks;
namespace GeneralGommunication namespace GeneralGommunication
{ {
//public class GComm_TcpClient : IGeneralComm
//{
// public Logger logger = NLog.LogManager.GetCurrentClassLogger();
// /// <summary>
// /// IP 地址 and 端口号
// /// </summary>
// public string Addr { get; set; }
// IPEndPoint RemoteEp => Misc.StringConverter.ToIPEndPoint(Addr);
// /// <summary>
// /// 是否异常
// /// </summary>
// public bool IsError => !string.IsNullOrEmpty(ErrMsg);
// /// <summary>
// /// 异常信息
// /// </summary>
// public string ErrMsg { get; private set; }
// /// <summary>
// /// 运行中
// /// </summary>
// public bool IsRunning { get; private set; }
// /// <summary>
// /// 连接成功
// /// </summary>
// public bool IsConnected { get; private set; }
// /// <summary>
// /// 接收task调用
// /// </summary>
// public event IGeneralCommDataReceivedHandler DataReceived;
// public event PropertyChangedEventHandler PropertyChanged;
// Socket sock;
// CancellationTokenSource cts_readTask;
// CancellationTokenSource cts_waitForSend;
// CancellationTokenSource cts_sendTask;
// public GComm_TcpClient()
// {
// }
// /// <summary>
// /// 开始
// /// </summary>
// public void Start()
// {
// if (IsRunning)
// return;
// IsRunning = true;
// cts_readTask = new CancellationTokenSource();
// Task.Factory.StartNew(OnTask, cts_readTask.Token);
// }
// /// <summary>
// /// 接收
// /// </summary>
// public void Stop()
// {
// if (!IsRunning)
// return;
// IsRunning = false;
// IsConnected = false;
// cts_readTask.Cancel();
// if (sock != null && sock.Connected)
// {
// sock.Close();
// sock = null;
// }
// }
// void OnTask()
// {
// while (!cts_readTask.IsCancellationRequested)
// {
// ConnectTask();
// if (IsConnected)
// {
// sendBuf.Clear();
// //启动发送task
// cts_sendTask = new CancellationTokenSource();
// var sendtask = Task.Factory.StartNew(SendTask, cts_sendTask.Token);
// //进入接收task
// ReceiveTask();
// //退出了，肯定连接断开了
// //需要等待 SendTask 也退出了
// cts_sendTask.Cancel();
// sendtask.Wait();
// if (sock != null)
// {
// sock.Close();
// }
// }
// //休息一会儿，在重连
// try
// {
// Task.Delay(2000, cts_readTask.Token).Wait();
// }
// catch (Exception e)
// {
// //被打断了
// break;
// }
// }
// IsConnected = false;
// }
// /// <summary>
// /// 连接任务
// /// </summary>
// void ConnectTask()
// {
// sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
// sock.Blocking = true;
// //1秒内，需要收到信息
// sock.ReceiveTimeout = 1000;
// //1秒内，必须发送完
// sock.SendTimeout = 1000;
// while (!cts_readTask.IsCancellationRequested)
// {
// try
// {
// sock.Connect(RemoteEp);
// IsConnected = true;
// ErrMsg = null;
// return;
// }
// catch (SocketException e)
// {
// //连接出错
// //等待重试
// ErrMsg = e.Message;
// }
// try
// {
// Task.Delay(2000, cts_readTask.Token).Wait();
// }
// catch (Exception e)
// {
// //被打断
// return;
// }
// }
// }
// /// <summary>
// /// 接收任务
// /// </summary>
// void ReceiveTask()
// {
// byte[] buf = new byte[sock.ReceiveBufferSize];
// while (!cts_readTask.IsCancellationRequested)
// {
// int len;
// try
// {
// len = sock.Receive(buf);
// if (len == 0)
// {
// continue;//没收到数据，继续等
// }
// }
// catch (SocketException e)
// {
// if (e.SocketErrorCode == SocketError.TimedOut)
// continue;//超时而已
// else if ((e.SocketErrorCode == SocketError.ConnectionAborted)
// || (e.SocketErrorCode == SocketError.ConnectionRefused)
// || (e.SocketErrorCode == SocketError.ConnectionReset))
// {
// IsConnected = false;
// return;
// }
// //Console.WriteLine($"ReceiveTask() {e}");
// //肯定断开连接了
// IsConnected = false;
// break;
// }
// logger.Debug($"ReceiveTask len={len}");
// DataReceived?.Invoke(this, buf.Take(len).ToArray());
// }
// IsConnected = false;
// }
// /// <summary>
// /// 发送任务
// /// </summary>
// void SendTask()
// {
// while (!cts_sendTask.IsCancellationRequested)
// {
// CancellationTokenSource cts3;
// while (true)
// {
// byte[] buffer;
// lock (sendBuf)
// {
// if (sendBuf.Count <= 0)
// break;
// buffer = sendBuf.ToArray();
// }
// int slen;
// try
// {
// slen = sock.Send(buffer);
// logger.Debug($"SendTask len={slen}");
// }
// catch (Exception e)
// {
// //连接断开了
// IsConnected = false;
// break;
// }
// if (slen <= 0)
// {
// //连接断开了
// IsConnected = false;
// break;
// }
// else
// {
// lock (sendBuf)
// {
// if (slen == sendBuf.Count())
// {
// sendBuf.Clear();
// break;
// }
// else
// {
// sendBuf.RemoveRange(0, slen);
// }
// }
// }
// }
// //重新等 下次被呼醒
// lock (sendBuf)
// {
// // 呼醒 发送task
// if (cts_waitForSend == null)
// {
// cts_waitForSend = new CancellationTokenSource();
// }
// //cts_waitForSend 与 cts_sendTask 合体, 任意一个都会呼醒 delay
// cts3 = CancellationTokenSource.CreateLinkedTokenSource(cts_waitForSend.Token, cts_sendTask.Token);
// }
// try
// {
// Task.Delay(-1, cts3.Token).Wait();
// }
// catch (Exception e)
// {
// //被打断， 有数据需要发送
// }
// cts_waitForSend = null;
// }
// }
// /// <summary>
// /// 无限大，发送缓存
// /// </summary>
// List<byte> sendBuf = new List<byte>();
// ///// <summary>
// ///// 发送数据
// ///// </summary>
// ///// <param name="buf"></param>
// //public void Write(IEnumerable<byte> buf)
// //{
// // if (!IsConnected)
// // return;
// // lock (sendBuf)
// // {
// // //放入，发送缓存区
// // sendBuf.AddRange(buf);
// // //呼醒 发送task
// // if (cts_waitForSend == null)
// // {
// // cts_waitForSend = new CancellationTokenSource();
// // }
// // cts_waitForSend.Cancel();
// // }
// //}
// /// <summary>
// /// 发送数据
// /// </summary>
// /// <param name="buf"></param>
// public void Write(IEnumerable<byte> buf)
// {
// if (!IsConnected)
// return;
// int slen;
// try
// {
// slen = sock.Send(buf.ToArray());
// logger.Debug($"SendTask len={slen}");
// }
// catch (Exception e)
// {
// //连接断开了
// IsConnected = false;
// return;
// }
// if (slen <= 0)
// {
// //连接断开了
// IsConnected = false;
// return;
// }
// }
// /// <summary>
// /// 清空输入缓存
// /// </summary>
// public void DiscardInBuffer()
// {
// //TODO
// }
//}
public class GComm_TcpClient : IGeneralComm public class GComm_TcpClient : IGeneralComm
{ {
public Logger logger;// = NLog.LogManager.GetCurrentClassLogger(); public Logger logger;// = NLog.LogManager.GetCurrentClassLogger();
...@@ -399,6 +51,11 @@ namespace GeneralGommunication ...@@ -399,6 +51,11 @@ namespace GeneralGommunication
Socket sock; Socket sock;
CancellationTokenSource cts_readTask; CancellationTokenSource cts_readTask;
/// <summary>
/// sock 的线程锁
/// </summary>
object sock_lock = new object();
Task task = null;
public GComm_TcpClient() public GComm_TcpClient()
{ {
...@@ -412,9 +69,21 @@ namespace GeneralGommunication ...@@ -412,9 +69,21 @@ namespace GeneralGommunication
if (IsRunning) if (IsRunning)
return; return;
IsRunning = true; IsRunning = true;
cts_readTask = new CancellationTokenSource();
Task.Factory.StartNew(OnTask, cts_readTask.Token); if (task == null || task.IsCompleted)
{
//线程已经完成, 创建个新的
cts_readTask = new CancellationTokenSource();
task = Task.Factory.StartNew(OnTask, cts_readTask.Token);
}
else
{
//线程还没结束，等待结束后，再创建新的
task.Wait();
cts_readTask = new CancellationTokenSource();
task = Task.Factory.StartNew(OnTask, cts_readTask.Token);
}
} }
/// <summary> /// <summary>
...@@ -427,13 +96,35 @@ namespace GeneralGommunication ...@@ -427,13 +96,35 @@ namespace GeneralGommunication
IsRunning = false; IsRunning = false;
IsConnected = false; IsConnected = false;
cts_readTask.Cancel(); cts_readTask.Cancel();
if (sock != null && sock.Connected) closeSock();
}
/// <summary>
/// 关闭sock，打断接收，发送中的等待，非OnTask调用
/// </summary>
void closeSock()
{
lock (sock_lock)
{ {
sock.Close(); if (sock != null && sock.Connected)
sock = null; {
sock.Close();
}
}
}
/// <summary>
/// 释放sock， 只能在OnTask中调用
/// </summary>
void disposeSock()
{
lock (sock_lock)//lock住，防止外部线程在 sock==null 时,调用sock.Close()
{
if (sock != null)
{
sock.Close();
sock = null;
}
} }
} }
void OnTask() void OnTask()
{ {
while (!cts_readTask.IsCancellationRequested) while (!cts_readTask.IsCancellationRequested)
...@@ -446,14 +137,10 @@ namespace GeneralGommunication ...@@ -446,14 +137,10 @@ namespace GeneralGommunication
ReceiveTask(); ReceiveTask();
//退出了，肯定连接断开了 //退出了，肯定连接断开了
disposeSock();
if (sock != null)
{
sock.Close();
}
} }
//休息一会儿，在重连 //休息一会儿，再重连
try try
{ {
Task.Delay(2000, cts_readTask.Token).Wait(); Task.Delay(2000, cts_readTask.Token).Wait();
...@@ -464,7 +151,7 @@ namespace GeneralGommunication ...@@ -464,7 +151,7 @@ namespace GeneralGommunication
break; break;
} }
} }
disposeSock();
IsConnected = false; IsConnected = false;
} }
...@@ -473,12 +160,11 @@ namespace GeneralGommunication ...@@ -473,12 +160,11 @@ namespace GeneralGommunication
/// </summary> /// </summary>
void ConnectTask() void ConnectTask()
{ {
while (!cts_readTask.IsCancellationRequested) while (!cts_readTask.IsCancellationRequested)
{ {
try try
{ {
//sock 的 创建，销毁，只能在 这个线程操作
sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp); sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
sock.Blocking = true; sock.Blocking = true;
...@@ -500,16 +186,18 @@ namespace GeneralGommunication ...@@ -500,16 +186,18 @@ namespace GeneralGommunication
} }
else else
{ {
sock.Close(); disposeSock();
ErrMsg = "连接超时"; ErrMsg = "连接超时";
} }
} }
catch (Exception e) catch (Exception e)
{ {
disposeSock();
//连接出错 //连接出错
//等待重试 //等待重试
ErrMsg = e.Message; ErrMsg = e.Message;
} }
try try
{ {
Task.Delay(2000, cts_readTask.Token).Wait(); Task.Delay(2000, cts_readTask.Token).Wait();
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment