C#实现异步阻塞TCP

// This class creates a single large buffer which can be divided up
// and assigned to SocketAsyncEventArgs objects for use with each
// socket I/O operation. 
// This enables bufffers to be easily reused and guards against
// fragmenting heap memory.
//
// The operations exposed on the BufferManager class are not thread safe.
public class BufferManager
{
    //buffer缓冲区大小
    private int m_numBytes;
    //缓冲区
    private byte[] m_buffer;
    private Stack<int> m_freeIndexPool;
    private int m_currentIndex;
    private int m_bufferSize;
 
    public BufferManager(int totalBytes, int bufferSize)
    {
        m_numBytes = totalBytes;
        m_currentIndex = 0;
        m_bufferSize = bufferSize;
        m_freeIndexPool = new Stack<int>();
    }
 
    /// <summary>
    /// 给buffer分配缓冲区
    /// </summary>
    public void InitBuffer()
    {
        m_buffer = new byte[m_numBytes];
    }
 
    /// <summary>
    ///  将buffer添加到args的IO缓冲区中，并设置offset
    /// </summary>
    /// <param name="args"></param>
    /// <returns></returns>
    public bool SetBuffer(SocketAsyncEventArgs args)
    {
        if (m_freeIndexPool.Count > 0)
        {
            args.SetBuffer(m_buffer, m_freeIndexPool.Pop(), m_bufferSize);
        }
        else
        {
            if ((m_numBytes - m_bufferSize) < m_currentIndex)
            {
                return false;
            }
            args.SetBuffer(m_buffer, m_currentIndex, m_bufferSize);
            m_currentIndex += m_bufferSize;
        }
        return true;
    }
 
    /// <summary>
    /// 将buffer从args的IO缓冲区中释放
    /// </summary>
    /// <param name="args"></param>
    public void FreeBuffer(SocketAsyncEventArgs args)
    {
        m_freeIndexPool.Push(args.Offset);
        args.SetBuffer(null, 0, 0);
    }
 
    /// <summary>
    /// 释放全部buffer缓存
    /// </summary>
    public void FreeAllBuffer()
    {
        m_freeIndexPool.Clear();
        m_currentIndex = 0;
        m_buffer = null;
    }
}　

// Represents a collection of reusable SocketAsyncEventArgs objects. 
public class SocketAsyncEventArgsPool
{
    private Stack<SocketAsyncEventArgs> m_pool;
 
    // Initializes the object pool to the specified size
    //
    // The "capacity" parameter is the maximum number of
    // SocketAsyncEventArgs objects the pool can hold
    public SocketAsyncEventArgsPool(int capacity)
    {
        m_pool = new Stack<SocketAsyncEventArgs>(capacity);
    }
 
    // Add a SocketAsyncEventArg instance to the pool
    //
    //The "item" parameter is the SocketAsyncEventArgs instance
    // to add to the pool
    public void Push(SocketAsyncEventArgs item)
    {
        if (item == null) { throw new ArgumentNullException("Items added to a SocketAsyncEventArgsPool cannot be null"); }
        lock (m_pool)
        {
            m_pool.Push(item);
        }
    }
 
    // Removes a SocketAsyncEventArgs instance from the pool
    // and returns the object removed from the pool
    public SocketAsyncEventArgs Pop()
    {
        lock (m_pool)
        {
            return m_pool.Pop();
        }
    }
 
    /// <summary>
    /// 清空栈中元素
    /// </summary>
    public void Clear()
    {
        lock (m_pool)
        {
            m_pool.Clear();
        }
    }
 
    // The number of SocketAsyncEventArgs instances in the pool
    public int Count
    {
        get { return m_pool.Count; }
    }
 
}

public class AsyncUserToken
{
    private Socket socket = null;
    public Socket Socket { get => socket; set => socket = value; }
}

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);
            }
        }
    }
}

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();
        }
    }
}

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 + "\r\n";
        }));
    }
 
    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 = "";
    }
}

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 + "\r\n";
        }));
    }
 
    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 = "";
    }
}