高性能網絡服務器實現（C#封裝IOCP）

時間 2019-12-30

原文原文鏈接

之前發在博客園的文章，今天被人翻出來，發現這個在CSDN的下載量和評論都不錯，所以搬過來與你們分享。爲了方便你們下載源代碼，我把代碼託管到開源中國了，呵呵！
有點懶，之前寫博客都沒堅持住，我在這邊是看成日記功能，也就沒把之前的東東整理過來。git

最近有項目要作一個高性能網絡服務器，決定下功夫搞定完成端口（IOCP），最終花了一個星期終於把它弄清楚了，並用C++寫了一個版本，效率很不錯。
但，從項目的整體需求來考慮，最終決定上.net平臺，所以又花了一天一晚上弄出了一個C#版，在這與你們分享。服務器

一些心得體會：

一、在C#中，不用去面對完成端口的操做系統內核對象，Microsoft已經爲咱們提供了SocketAsyncEventArgs類，它封裝了IOCP的使用。請參考：http://msdn.microsoft.com/zh-cn/library/system.net.sockets.socketasynceventargs.aspx?cs-save-lang=1&cs-lang=cpp#code-snippet-1。網絡

二、個人SocketAsyncEventArgsPool類使用List對象來存儲對客戶端來通訊的SocketAsyncEventArgs對象，它至關於直接使用內核對象時的IoContext。我這樣設計比用堆棧來實現的好處理是，我能夠在SocketAsyncEventArgsPool池中找到任何一個與服務器鏈接的客戶，主動向它發信息。而用堆棧來實現的話，要主動給客戶發信息，則還要設計一個結構來存儲已鏈接上服務器的客戶。socket

三、對每個客戶端無論還發送仍是接收，我使用同一個SocketAsyncEventArgs對象，對每個客戶端來講，通訊是同步進行的，也就是說服務器高度保證同一個客戶鏈接上要麼在投遞發送請求，並等待；或者是在投遞接收請求，等待中。本例只作echo服務器，還未考慮由服務器主動向客戶發送信息。async

四、SocketAsyncEventArgs的UserToken被直接設定爲被接受的客戶端Socket。函數

五、沒有使用BufferManager 類，由於我在初始化時給每個SocketAsyncEventArgsPool中的對象分配一個緩衝區，發送時使用Arrary.Copy來進行字符拷貝，不去改變緩衝區的位置，只改變使用的長度，所以在下次投遞接收請求時恢復緩衝區長度就能夠了！若是要主動給客戶發信息的話，能夠new一個SocketAsyncEventArgs對象，或者在初始化中創建幾個來專門用於主動發送信息，由於這種需求通常是進行信息羣發，創建一個對象能夠用於不少次信息發送，整體來看，這種花銷不大，還減去了字符拷貝和消耗。性能

六、測試結果：（在個人筆記本上時行的，個人本本是T420 I7 8G內存）測試

100客戶 100，000（十萬次）不間斷的發送接收數據（發送和接收之間沒有Sleep，就一個一循環，不斷的發送與接收）
耗時3004.6325 秒完成
總共 10,000,000 一千萬次訪問
平均每分完成 199,691.6 次發送與接收
平均每秒完成 3,328.2 次發送與接收ui

整個運行過程當中，內存消耗在開始兩三分種後就保持穩定再也不增漲。this

看了一下對每一個客戶端的延遲最多不超過2毫秒，CPU佔用在8%左右。

七、下載地址：http://download.csdn.net/detail/ztk12/4928644
八、源代碼託管： http://git.oschina.net/zhoutk/IocpServer

九、主要源碼：
IoContextPool.cs

IoContextPool.cs
using System;
using System.Collections.Generic;
using System.Text;
using System.Net.Sockets;

namespace IocpServer
{
    /// <summary>
    /// 與每一個客戶Socket相關聯，進行Send和Receive投遞時所須要的參數
    /// </summary>
    internal sealed class IoContextPool
    {
        List<SocketAsyncEventArgs> pool;        //爲每個Socket客戶端分配一個SocketAsyncEventArgs，用一個List管理，在程序啓動時創建。
        Int32 capacity;                         //pool對象池的容量
        Int32 boundary;                         //已分配和未分配對象的邊界，大的是已經分配的，小的是未分配的

        internal IoContextPool(Int32 capacity)
        {
            this.pool = new List<SocketAsyncEventArgs>(capacity);
            this.boundary = 0;
            this.capacity = capacity;
        }

        /// <summary>
        /// 往pool對象池中增長新創建的對象，由於這個程序在啓動時會創建好全部對象，
        /// 故這個方法只在初始化時會被調用,所以，沒有加鎖。
        /// </summary>
        /// <param name="arg"></param>
        /// <returns></returns>
        internal bool Add(SocketAsyncEventArgs arg)
        {
            if (arg != null && pool.Count < capacity)
            {
                pool.Add(arg);
                boundary++;
                return true;
            }
            else
                return false;
        }

        /// <summary>
        /// 取出集合中指定對象，內部使用
        /// </summary>
        /// <param name="index"></param>
        /// <returns></returns>
        //internal SocketAsyncEventArgs Get(int index)
        //{
        //    if (index >= 0 && index < capacity)
        //        return pool[index];
        //    else
        //        return null;
        //}

        /// <summary>
        /// 從對象池中取出一個對象，交給一個socket來進行投遞請求操做
        /// </summary>
        /// <returns></returns>
        internal SocketAsyncEventArgs Pop()
        {
            lock (this.pool)
            {
                if (boundary > 0)
                {
                    --boundary;
                    return pool[boundary];
                }
                else
                    return null;
            }
        }

        /// <summary>
        /// 一個socket客戶斷開，與其相關的IoContext被釋放，從新投入Pool中，備用。
        /// </summary>
        /// <param name="arg"></param>
        /// <returns></returns>
        internal bool Push(SocketAsyncEventArgs arg)
        {
            if (arg != null)
            {
                lock (this.pool)
                {
                    int index = this.pool.IndexOf(arg, boundary);  //找出被斷開的客戶,此處必定能查到，所以index不可能爲-1，一定要大於0。
                    if (index == boundary)         //正好是邊界元素
                        boundary++;
                    else
                    {
                        this.pool[index] = this.pool[boundary];     //將斷開客戶移到邊界上，邊界右移
                        this.pool[boundary++] = arg;
                    }
                }
                return true;
            }
            else
                return false;
        }
    }
}

IoServer.cs

IoServer.cs
using System;
using System.Collections.Generic;
using System.Text;
using System.Net.Sockets;
using System.Threading;
using System.Net;

namespace IocpServer
{
    /// <summary>
    /// 基於SocketAsyncEventArgs 實現 IOCP 服務器
    /// </summary>
    internal sealed class IoServer
    {
        /// <summary>
        /// 監聽Socket，用於接受客戶端的鏈接請求
        /// </summary>
        private Socket listenSocket;

        /// <summary>
        /// 用於服務器執行的互斥同步對象
        /// </summary>
        private static Mutex mutex = new Mutex();

        /// <summary>
        /// 用於每一個I/O Socket操做的緩衝區大小
        /// </summary>
        private Int32 bufferSize;

        /// <summary>
        /// 服務器上鍊接的客戶端總數
        /// </summary>
        private Int32 numConnectedSockets;

        /// <summary>
        /// 服務器能接受的最大鏈接數量
        /// </summary>
        private Int32 numConnections;

        /// <summary>
        /// 完成端口上進行投遞所用的IoContext對象池
        /// </summary>
        private IoContextPool ioContextPool;

        public MainForm mainForm;

        /// <summary>
        /// 構造函數，創建一個未初始化的服務器實例
        /// </summary>
        /// <param name="numConnections">服務器的最大鏈接數據</param>
        /// <param name="bufferSize"></param>
        internal IoServer(Int32 numConnections, Int32 bufferSize)
        {
            this.numConnectedSockets = 0;
            this.numConnections = numConnections;
            this.bufferSize = bufferSize;

            this.ioContextPool = new IoContextPool(numConnections);

            // 爲IoContextPool預分配SocketAsyncEventArgs對象
            for (Int32 i = 0; i < this.numConnections; i++)
            {
                SocketAsyncEventArgs ioContext = new SocketAsyncEventArgs();
                ioContext.Completed += new EventHandler<SocketAsyncEventArgs>(OnIOCompleted);
                ioContext.SetBuffer(new Byte[this.bufferSize], 0, this.bufferSize);

                // 將預分配的對象加入SocketAsyncEventArgs對象池中
                this.ioContextPool.Add(ioContext);
            }
        }

        /// <summary>
        /// 當Socket上的發送或接收請求被完成時，調用此函數
        /// </summary>
        /// <param name="sender">激發事件的對象</param>
        /// <param name="e">與發送或接收完成操做相關聯的SocketAsyncEventArg對象</param>
        private void OnIOCompleted(object sender, SocketAsyncEventArgs e)
        {
            // Determine which type of operation just completed and call the associated handler.
            switch (e.LastOperation)
            {
                case SocketAsyncOperation.Receive:
                    this.ProcessReceive(e);
                    break;
                case SocketAsyncOperation.Send:
                    this.ProcessSend(e);
                    break;
                default:
                    throw new ArgumentException("The last operation completed on the socket was not a receive or send");
            }
        }

        /// <summary>
        ///接收完成時處理函數
        /// </summary>
        /// <param name="e">與接收完成操做相關聯的SocketAsyncEventArg對象</param>
        private void ProcessReceive(SocketAsyncEventArgs e)
        {
            // 檢查遠程主機是否關閉鏈接
            if (e.BytesTransferred > 0)
            {
                if (e.SocketError == SocketError.Success)
                {
                    Socket s = (Socket)e.UserToken;
                    //判斷全部需接收的數據是否已經完成
                    if (s.Available == 0)
                    {
                        // 設置發送數據
                        Array.Copy(e.Buffer, 0, e.Buffer, e.BytesTransferred, e.BytesTransferred);
                        e.SetBuffer(e.Offset, e.BytesTransferred * 2);
                        if (!s.SendAsync(e))        //投遞發送請求，這個函數有可能同步發送出去，這時返回false，而且不會引起SocketAsyncEventArgs.Completed事件
                        {
                            // 同步發送時處理髮送完成事件
                            this.ProcessSend(e);
                        }
                    }
                    else if (!s.ReceiveAsync(e))    //爲接收下一段數據，投遞接收請求，這個函數有可能同步完成，這時返回false，而且不會引起SocketAsyncEventArgs.Completed事件
                    {
                        // 同步接收時處理接收完成事件
                        this.ProcessReceive(e);
                    }
                }
                else
                {
                    this.ProcessError(e);
                }
            }
            else
            {
                this.CloseClientSocket(e);
            }
        }

        /// <summary>
        /// 發送完成時處理函數
        /// </summary>
        /// <param name="e">與發送完成操做相關聯的SocketAsyncEventArg對象</param>
        private void ProcessSend(SocketAsyncEventArgs e)
        {
            if (e.SocketError == SocketError.Success)
            {
                Socket s = (Socket)e.UserToken;

                //接收時根據接收的字節數收縮了緩衝區的大小，所以投遞接收請求時，恢復緩衝區大小
                e.SetBuffer(0, bufferSize);
                if (!s.ReceiveAsync(e))     //投遞接收請求
                {
                    // 同步接收時處理接收完成事件
                    this.ProcessReceive(e);
                }
            }
            else
            {
                this.ProcessError(e);
            }
        }

        /// <summary>
        /// 處理socket錯誤
        /// </summary>
        /// <param name="e"></param>
        private void ProcessError(SocketAsyncEventArgs e)
        {
            Socket s = e.UserToken as Socket;
            IPEndPoint localEp = s.LocalEndPoint as IPEndPoint;

            this.CloseClientSocket(s, e);

            string outStr = String.Format("套接字錯誤 {0}, IP {1}, 操做 {2}。", (Int32)e.SocketError, localEp, e.LastOperation);
            mainForm.Invoke(mainForm.setlistboxcallback, outStr);
            //Console.WriteLine("Socket error {0} on endpoint {1} during {2}.", (Int32)e.SocketError, localEp, e.LastOperation);
        }

        /// <summary>
        /// 關閉socket鏈接
        /// </summary>
        /// <param name="e">SocketAsyncEventArg associated with the completed send/receive operation.</param>
        private void CloseClientSocket(SocketAsyncEventArgs e)
        {
            Socket s = e.UserToken as Socket;
            this.CloseClientSocket(s, e);
        }

        private void CloseClientSocket(Socket s, SocketAsyncEventArgs e)
        {
            Interlocked.Decrement(ref this.numConnectedSockets);

            // SocketAsyncEventArg 對象被釋放，壓入可重用隊列。
            this.ioContextPool.Push(e);            
            string outStr = String.Format("客戶 {0} 斷開, 共有 {1} 個鏈接。", s.RemoteEndPoint.ToString(), this.numConnectedSockets);
            mainForm.Invoke(mainForm.setlistboxcallback, outStr);            
            //Console.WriteLine("A client has been disconnected from the server. There are {0} clients connected to the server", this.numConnectedSockets);
            try
            {
                s.Shutdown(SocketShutdown.Send);
            }
            catch (Exception)
            {
                // Throw if client has closed, so it is not necessary to catch.
            }
            finally
            {
                s.Close();
            }
        }

        /// <summary>
        /// accept 操做完成時回調函數
        /// </summary>
        /// <param name="sender">Object who raised the event.</param>
        /// <param name="e">SocketAsyncEventArg associated with the completed accept operation.</param>
        private void OnAcceptCompleted(object sender, SocketAsyncEventArgs e)
        {
            this.ProcessAccept(e);
        }

        /// <summary>
        /// 監聽Socket接受處理
        /// </summary>
        /// <param name="e">SocketAsyncEventArg associated with the completed accept operation.</param>
        private void ProcessAccept(SocketAsyncEventArgs e)
        {
            Socket s = e.AcceptSocket;
            if (s.Connected)
            {
                try
                {
                    SocketAsyncEventArgs ioContext = this.ioContextPool.Pop();
                    if (ioContext != null)
                    {
                        // 從接受的客戶端鏈接中取數據配置ioContext

                        ioContext.UserToken = s;

                        Interlocked.Increment(ref this.numConnectedSockets);
                        string outStr = String.Format("客戶 {0} 連入, 共有 {1} 個鏈接。",  s.RemoteEndPoint.ToString(),this.numConnectedSockets);
                        mainForm.Invoke(mainForm.setlistboxcallback,outStr);
                        //Console.WriteLine("Client connection accepted. There are {0} clients connected to the server",
                            //this.numConnectedSockets);

                        if (!s.ReceiveAsync(ioContext))
                        {
                            this.ProcessReceive(ioContext);
                        }
                    }
                    else        //已經達到最大客戶鏈接數量，在這接受鏈接，發送「鏈接已經達到最大數」，而後斷開鏈接
                    {
                        s.Send(Encoding.Default.GetBytes("鏈接已經達到最大數!"));
                        string outStr = String.Format("鏈接已滿，拒絕 {0} 的鏈接。", s.RemoteEndPoint);
                        mainForm.Invoke(mainForm.setlistboxcallback, outStr);
                        s.Close();
                   }
                }
                catch (SocketException ex)
                {
                    Socket token = e.UserToken as Socket;
                    string outStr = String.Format("接收客戶 {0} 數據出錯, 異常信息： {1} 。", token.RemoteEndPoint, ex.ToString());
                    mainForm.Invoke(mainForm.setlistboxcallback, outStr);
                    //Console.WriteLine("Error when processing data received from {0}:\r\n{1}", token.RemoteEndPoint, ex.ToString());
                }
                catch (Exception ex)
                {
                    mainForm.Invoke(mainForm.setlistboxcallback, "異常：" + ex.ToString());
                }
                // 投遞下一個接受請求
                this.StartAccept(e);
            }
        }

        /// <summary>
        /// 從客戶端開始接受一個鏈接操做
        /// </summary>
        /// <param name="acceptEventArg">The context object to use when issuing 
        /// the accept operation on the server's listening socket.</param>
        private void StartAccept(SocketAsyncEventArgs acceptEventArg)
        {
            if (acceptEventArg == null)
            {
                acceptEventArg = new SocketAsyncEventArgs();
                acceptEventArg.Completed += new EventHandler<SocketAsyncEventArgs>(OnAcceptCompleted);
            }
            else
            {
                // 重用前進行對象清理
                acceptEventArg.AcceptSocket = null;
            }

            if (!this.listenSocket.AcceptAsync(acceptEventArg))
            {
                this.ProcessAccept(acceptEventArg);
            }
        }

        /// <summary>
        /// 啓動服務，開始監聽
        /// </summary>
        /// <param name="port">Port where the server will listen for connection requests.</param>
        internal void Start(Int32 port)
        {
            // 得到主機相關信息
            IPAddress[] addressList = Dns.GetHostEntry(Environment.MachineName).AddressList;
            IPEndPoint localEndPoint = new IPEndPoint(addressList[addressList.Length - 1], port);

            // 建立監聽socket
            this.listenSocket = new Socket(localEndPoint.AddressFamily, SocketType.Stream, ProtocolType.Tcp);
            this.listenSocket.ReceiveBufferSize = this.bufferSize;
            this.listenSocket.SendBufferSize = this.bufferSize;

            if (localEndPoint.AddressFamily == AddressFamily.InterNetworkV6)
            {
                // 配置監聽socket爲 dual-mode (IPv4 & IPv6) 
                // 27 is equivalent to IPV6_V6ONLY socket option in the winsock snippet below,
                this.listenSocket.SetSocketOption(SocketOptionLevel.IPv6, (SocketOptionName)27, false);
                this.listenSocket.Bind(new IPEndPoint(IPAddress.IPv6Any, localEndPoint.Port));
            }
            else
            {
                this.listenSocket.Bind(localEndPoint);
            }

            // 開始監聽
            this.listenSocket.Listen(this.numConnections);

            // 在監聽Socket上投遞一個接受請求。
            this.StartAccept(null);

            // Blocks the current thread to receive incoming messages.
            mutex.WaitOne();
        }

        /// <summary>
        /// 中止服務
        /// </summary>
        internal void Stop()
        {
            this.listenSocket.Close();
            mutex.ReleaseMutex();
        }

    }
}