C#实现异步阻塞TCP（SocketAsyncEventArgs，SendAsync，Receive
1.服务器端代码
public class TcpServiceSocketAsync
{
//接收数据事件
public Action<string> recvMessageEvent = null;
//发送结果事件
public Action<int> sendResultEvent = null;
//监听socket
private Socket listenSocket = null;
//允许连接到tcp服务器的tcp客户端数量
private int numConnections = 0;
//用于socket发送和接受的缓存区大小
private int bufferSize = 0;
//socket缓冲区管理对象
private BufferManager bufferManager = null;
//SocketAsyncEventArgs池
private SocketAsyncEventArgsPool socketAsyncEventArgsPool = null;
//当前连接的tcp客户端数量
private int numberAcceptedClients = 0;
//控制tcp客户端连接数量的信号量
private Semaphore maxNumberAcceptedClients = null;
//用于socket发送数据的SocketAsyncEventArgs集合
private List<SocketAsyncEventArgs> sendAsyncEventArgs = null;
//tcp服务器ip
private string ip = "";
//tcp服务器端口
private int port = 0;
/// <summary>
/// 构造函数
/// </summary>
/// <param name="numConnections">允许连接到tcp服务器的tcp客户端数量</param>
/// <param name="bufferSize">用于socket发送和接受的缓存区大小</param>
public TcpServiceSocketAsync(int numConnections, int bufferSize)
{
if (numConnections <= 0 || numConnections > int.MaxValue)
throw new ArgumentOutOfRangeException("_numConnections is out of range");
if (bufferSize <= 0 || bufferSize > int.MaxValue)
throw new ArgumentOutOfRangeException("_receiveBufferSize is out of range");
this.numConnections = numConnections;
this.bufferSize = bufferSize;
bufferManager = new BufferManager(numConnections * bufferSize * 2, bufferSize);
socketAsyncEventArgsPool = new SocketAsyncEventArgsPool(numConnections);
maxNumberAcceptedClients = new Semaphore(numConnections, numConnections);
sendAsyncEventArgs = new List<SocketAsyncEventArgs>();
}
/// <summary>
/// 初始化数据（bufferManager，socketAsyncEventArgsPool）
/// </summary>
public void Init()
{
numberAcceptedClients = 0;
bufferManager.InitBuffer();
SocketAsyncEventArgs readWriteEventArg;
for (int i = 0; i < numConnections * 2; i++)
{
readWriteEventArg = new SocketAsyncEventArgs();
readWriteEventArg.Completed += new EventHandler<SocketAsyncEventArgs>(IO_Completed);
readWriteEventArg.UserToken = new AsyncUserToken();
bufferManager.SetBuffer(readWriteEventArg);
socketAsyncEventArgsPool.Push(readWriteEventArg);
}
}
/// <summary>
/// 开启tcp服务器，等待tcp客户端连接
/// </summary>
/// <param name="ip"></param>
/// <param name="port"></param>
public void Start(string ip, int port)
{
if (string.IsNullOrEmpty(ip))
throw new ArgumentNullException("ip cannot be null");
if (port < 1 || port > 65535)
throw new ArgumentOutOfRangeException("port is out of range");
this.ip = ip;
this.port = port;
try
{
listenSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
IPAddress address = IPAddress.Parse(ip);
IPEndPoint endpoint = new IPEndPoint(address, port);
listenSocket.Bind(endpoint);//绑定地址
listenSocket.Listen(int.MaxValue);//开始监听
StartAccept(null);
}
catch (Exception ex)
{
throw ex;
}
}
/// <summary>
/// 关闭tcp服务器
/// </summary>
public void CloseSocket()
{
if (listenSocket == null)
return;
try
{
foreach (var e in sendAsyncEventArgs)
{
((AsyncUserToken)e.UserToken).Socket.Shutdown(SocketShutdown.Both);
}
listenSocket.Shutdown(SocketShutdown.Both);
}
catch { }
try
{
foreach (var e in sendAsyncEventArgs)
{
((AsyncUserToken)e.UserToken).Socket.Close();
}
listenSocket.Close();
}
catch { }
try
{
foreach (var e in sendAsyncEventArgs)
{
e.Dispose();
}
}
catch { }
sendAsyncEventArgs.Clear();
socketAsyncEventArgsPool.Clear();
bufferManager.FreeAllBuffer();
maxNumberAcceptedClients.Release(numberAcceptedClients);
}
/// <summary>
/// 重新启动tcp服务器
/// </summary>
public void Restart()
{
CloseSocket();
Init();
Start(ip, port);
}
/// <summary>
/// 开始等待tcp客户端连接
/// </summary>
/// <param name="acceptEventArg"></param>
private void StartAccept(SocketAsyncEventArgs acceptEventArg)
{
if (acceptEventArg == null)
{
acceptEventArg = new SocketAsyncEventArgs();
acceptEventArg.Completed += new EventHandler<SocketAsyncEventArgs>(AcceptEventArg_Completed);
}
else
{
// socket must be cleared since the context object is being reused
acceptEventArg.AcceptSocket = null;
}
maxNumberAcceptedClients.WaitOne();
bool willRaiseEvent = listenSocket.AcceptAsync(acceptEventArg);
if (!willRaiseEvent)
{
ProcessAccept(acceptEventArg);
}
}
/// <summary>
/// Socket.AcceptAsync完成回调函数
/// </summary>
/// <param name="sender"></param>
/// <param name="e"></param>
private void AcceptEventArg_Completed(object sender, SocketAsyncEventArgs e)
{
ProcessAccept(e);
}
/// <summary>
/// 接受到tcp客户端连接，进行处理
/// </summary>
/// <param name="e"></param>
private void ProcessAccept(SocketAsyncEventArgs e)
{
Interlocked.Increment(ref numberAcceptedClients);
//设置用于接收的SocketAsyncEventArgs的socket，可以接受数据
SocketAsyncEventArgs recvEventArgs = socketAsyncEventArgsPool.Pop();
((AsyncUserToken)recvEventArgs.UserToken).Socket = e.AcceptSocket;
//设置用于发送的SocketAsyncEventArgs的socket，可以发送数据
SocketAsyncEventArgs sendEventArgs = socketAsyncEventArgsPool.Pop();
((AsyncUserToken)sendEventArgs.UserToken).Socket = e.AcceptSocket;
sendAsyncEventArgs.Add(sendEventArgs);
StartRecv(recvEventArgs);
StartAccept(e);
}
/// <summary>
/// tcp服务器开始接受tcp客户端发送的数据
/// </summary>
private void StartRecv(SocketAsyncEventArgs e)
{
bool willRaiseEvent = ((AsyncUserToken)e.UserToken).Socket.ReceiveAsync(e);
if (!willRaiseEvent)
{
ProcessReceive(e);
}
}
/// <summary>
/// socket.sendAsync和socket.recvAsync的完成回调函数
/// </summary>
/// <param name="sender"></param>
/// <param name="e"></param>
private void IO_Completed(object sender, SocketAsyncEventArgs e)
{
switch (e.LastOperation)
{
case SocketAsyncOperation.Receive:
ProcessReceive(e);
break;
case SocketAsyncOperation.Send:
ProcessSend(e);
break;
default:
throw new ArgumentException("The last operation completed on the socket was not a receive or send");
}
}
/// <summary>
/// 处理接受到的tcp客户端数据
/// </summary>
/// <param name="e"></param>
private void ProcessReceive(SocketAsyncEventArgs e)
{
AsyncUserToken token = (AsyncUserToken)e.UserToken;
if (e.BytesTransferred > 0 && e.SocketError == SocketError.Success)
{
if (recvMessageEvent != null)
//一定要指定GetString的长度
recvMessageEvent(Encoding.UTF8.GetString(e.Buffer, e.Offset, e.BytesTransferred));
StartRecv(e);
}
else
{
CloseClientSocket(e);
}
}
/// <summary>
/// 处理tcp服务器发送的结果
/// </summary>
/// <param name="e"></param>
private void ProcessSend(SocketAsyncEventArgs e)
{
AsyncUserToken token = (AsyncUserToken)e.UserToken;
if (e.SocketError == SocketError.Success)
{
if (sendResultEvent != null)
sendResultEvent(e.BytesTransferred);
}
else
{
if (sendResultEvent != null)
sendResultEvent(e.BytesTransferred);
CloseClientSocket(e);
}
}
/// <summary>
/// 关闭一个与tcp客户端连接的socket
/// </summary>
/// <param name="e"></param>
private void CloseClientSocket(SocketAsyncEventArgs e)
{
AsyncUserToken token = e.UserToken as AsyncUserToken;
try
{
//关闭socket时，单独使用socket.close()通常会造成资源提前被释放，应该在关闭socket之前，先使用shutdown进行接受或者发送的禁用，再使用socket进行释放
token.Socket.Shutdown(SocketShutdown.Both);
}
catch { }
token.Socket.Close();
Interlocked.Decrement(ref numberAcceptedClients);
socketAsyncEventArgsPool.Push(e);
maxNumberAcceptedClients.Release();
if (e.LastOperation == SocketAsyncOperation.Send)
sendAsyncEventArgs.Remove(e);
}
/// <summary>
/// 给全部tcp客户端发送数据
/// </summary>
/// <param name="message"></param>
public void SendMessageToAllClients(string message)
{
if (string.IsNullOrEmpty(message))
throw new ArgumentNullException("message cannot be null");
foreach (var e in sendAsyncEventArgs)
{
byte[] buff = Encoding.UTF8.GetBytes(message);
if (buff.Length > bufferSize)
throw new ArgumentOutOfRangeException("message is out off range");
buff.CopyTo(e.Buffer, e.Offset);
e.SetBuffer(e.Offset, buff.Length);
bool willRaiseEvent = ((AsyncUserToken)e.UserToken).Socket.SendAsync(e);
if (!willRaiseEvent)
{
ProcessSend(e);
}
}
}
}
2.客户端代码
public class TcpClientSocketAsync
{
//接收数据事件
public Action<string> recvMessageEvent = null;
//发送结果事件
public Action<int> sendResultEvent = null;
//接受缓存数组
private byte[] recvBuff = null;
//发送缓存数组
private byte[] sendBuff = null;
//连接socket
private Socket connectSocket = null;
//用于发送数据的SocketAsyncEventArgs
private SocketAsyncEventArgs sendEventArg = null;
//用于接收数据的SocketAsyncEventArgs
private SocketAsyncEventArgs recvEventArg = null;
//tcp服务器ip
private string ip = "";
//tcp服务器端口
private int port = 0;
/// <summary>
/// 构造函数
/// </summary>
/// <param name="bufferSize">用于socket发送和接受的缓存区大小</param>
public TcpClientSocketAsync(int bufferSize)
{
//设置用于发送数据的SocketAsyncEventArgs
sendBuff = new byte[bufferSize];
sendEventArg = new SocketAsyncEventArgs();
sendEventArg.Completed += new EventHandler<SocketAsyncEventArgs>(IO_Completed);
sendEventArg.SetBuffer(sendBuff, 0, bufferSize);
//设置用于接受数据的SocketAsyncEventArgs
recvBuff = new byte[bufferSize];
recvEventArg = new SocketAsyncEventArgs();
recvEventArg.Completed += new EventHandler<SocketAsyncEventArgs>(IO_Completed);
recvEventArg.SetBuffer(recvBuff, 0, bufferSize);
}
/// <summary>
/// 开启tcp客户端，连接tcp服务器
/// </summary>
/// <param name="ip"></param>
/// <param name="port"></param>
public void Start(string ip, int port)
{
if (string.IsNullOrEmpty(ip))
throw new ArgumentNullException("ip cannot be null");
if (port < 1 || port > 65535)
throw new ArgumentOutOfRangeException("port is out of range");
this.ip = ip;
this.port = port;
try
{
connectSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
IPAddress address = IPAddress.Parse(ip);
IPEndPoint endpoint = new IPEndPoint(address, port);
//连接tcp服务器
SocketAsyncEventArgs connectEventArg = new SocketAsyncEventArgs();
connectEventArg.Completed += ConnectEventArgs_Completed;
connectEventArg.RemoteEndPoint = endpoint;//设置要连接的tcp服务器地址
bool willRaiseEvent = connectSocket.ConnectAsync(connectEventArg);
if (!willRaiseEvent)
ProcessConnect(connectEventArg);
}
catch (Exception ex)
{
throw ex;
}
}
/// <summary>
/// 发送数据到tcp服务器
/// </summary>
/// <param name="message">要发送的数据</param>
public void SendMessage(string message)
{
if (string.IsNullOrEmpty(message))
throw new ArgumentNullException("message cannot be null");
if (connectSocket == null)
throw new Exception("socket cannot be null");
byte[] buff = Encoding.UTF8.GetBytes(message);
buff.CopyTo(sendBuff, 0);
sendEventArg.SetBuffer(0, buff.Length);
bool willRaiseEvent = connectSocket.SendAsync(sendEventArg);
if (!willRaiseEvent)
{
ProcessSend(sendEventArg);
}
}
/// <summary>
/// 关闭tcp客户端
/// </summary>
public void CloseSocket()
{
if (connectSocket == null)
return;
try
{
//关闭socket时，单独使用socket.close()通常会造成资源提前被释放，应该在关闭socket之前，先使用shutdown进行接受或者发送的禁用，再使用socket进行释放
connectSocket.Shutdown(SocketShutdown.Both);
}
catch { }
try
{
connectSocket.Close();
}
catch { }
}
/// <summary>
/// 重启tcp客户端，重新连接tcp服务器
/// </summary>
public void Restart()
{
CloseSocket();
Start(ip, port);
}
/// <summary>
/// Socket.ConnectAsync完成回调函数
/// </summary>
/// <param name="sender"></param>
/// <param name="e"></param>
private void ConnectEventArgs_Completed(object sender, SocketAsyncEventArgs e)
{
ProcessConnect(e);
}
private void ProcessConnect(SocketAsyncEventArgs e)
{
StartRecv();
}
/// <summary>
/// tcp客户端开始接受tcp服务器发送的数据
/// </summary>
private void StartRecv()
{
bool willRaiseEvent = connectSocket.ReceiveAsync(recvEventArg);
if (!willRaiseEvent)
{
ProcessReceive(recvEventArg);
}
}
/// <summary>
/// socket.sendAsync和socket.recvAsync的完成回调函数
/// </summary>
/// <param name="sender"></param>
/// <param name="e"></param>
private void IO_Completed(object sender, SocketAsyncEventArgs e)
{
switch (e.LastOperation)
{
case SocketAsyncOperation.Receive:
ProcessReceive(e);
break;
case SocketAsyncOperation.Send:
ProcessSend(e);
break;
default:
throw new ArgumentException("The last operation completed on the socket was not a receive or send");
}
}
/// <summary>
/// 处理接受到的tcp服务器数据
/// </summary>
/// <param name="e"></param>
private void ProcessReceive(SocketAsyncEventArgs e)
{
AsyncUserToken token = (AsyncUserToken)e.UserToken;
if (e.BytesTransferred > 0 && e.SocketError == SocketError.Success)
{
if (recvMessageEvent != null)
//一定要指定GetString的长度，否则整个bugger都要转成string
recvMessageEvent(Encoding.UTF8.GetString(e.Buffer, 0, e.BytesTransferred));
StartRecv();
}
else
{
Restart();
}
}
/// <summary>
/// 处理tcp客户端发送的结果
/// </summary>
/// <param name="e"></param>
private void ProcessSend(SocketAsyncEventArgs e)
{
AsyncUserToken token = (AsyncUserToken)e.UserToken;
if (e.SocketError == SocketError.Success)
{
if (sendResultEvent != null)
sendResultEvent(e.BytesTransferred);
}
else
{
if (sendResultEvent != null)
sendResultEvent(-1);
Restart();
}
}
}
3.使用
（1）服务器端
public partial class Form1 : Form
{
TcpServiceSocketAsync tcpServiceSocket = null;
private readonly string ip = "192.168.172.142";
private readonly int port = 8090;
public Form1()
{
InitializeComponent();
tcpServiceSocket = new TcpServiceSocketAsync(10, 1024);
tcpServiceSocket.recvMessageEvent += new Action<string>(Recv);
tcpServiceSocket.Init();
}
private void Recv(string message)
{
this.BeginInvoke(new Action(() =>
{
tbRecv.Text += message + "
";
}));
}
private void btnStart_Click(object sender, EventArgs e)
{
tcpServiceSocket.Start(ip, port);
}
private void btnSend_Click(object sender, EventArgs e)
{
string message = tbSend.Text.Trim();
if (string.IsNullOrEmpty(message))
return;
tcpServiceSocket.SendMessageToAllClients(message);
tbSend.Text = "";
}
}
（2）客户端
public partial class Form1 : Form
{
private TcpClientSocketAsync tcpClientSocket = null;
private readonly string ip = "192.168.172.142";
private readonly int port = 8090;
private readonly int buffSize = 1024;
public Form1()
{
InitializeComponent();
tcpClientSocket = new TcpClientSocketAsync(buffSize);
tcpClientSocket.recvMessageEvent += new Action<string>(Recv);
}
private void Recv(string message)
{
this.BeginInvoke(new Action(() =>
{
tbRecv.Text += message + "
";
}));
}
private void btnStart_Click(object sender, EventArgs e)
{
tcpClientSocket.Start(ip, port);
}
private void btnSend_Click(object sender, EventArgs e)
{
string message = tbSend.Text.Trim();
if (string.IsNullOrEmpty(message))
return;
tcpClientSocket.SendMessage(message);
tbSend.Text = "";
}
}
4.演示
参考：
https://docs.microsoft.com/zh-cn/dotnet/api/system.net.sockets.socketasynceventargs?view=netframework-4.8