Socket連接池

　　“池”這個概念好像最早是在操作系統的課上聽過的，到后來出來工作的第二天組長也跟我提起“池”這個東東。它給我的感覺是某種對象的集合，如果要用的話就取出，不用的話就放回。在學多線程的時候有接觸過線程池，在寫《Socket 一對多通信》的時候想到了Socket連接池這回事，不過在網上谷歌了一下，發現這類的文章貌似不多，看了一下園友的博文《高性能Socket設計實現》，獲益良多，下了份源碼來看，雖然有一部分看不明白，而且由於個人水平跑不了那份代碼，但是從中我學到了不少，至少我寫的“池”有一部分是用了這位田大哥的思想。

　　先來分析各個類之間的結構，整個連接池里面實際上是有兩個池，一個是在異步通信中可以重復利用的SocketAsyncEventArgs池(當然處於別的方面考慮，池里面並不是單純放SocketAsyncEventArgs的實例集合)，另一個是接收數據時用到的byte[]緩沖池。而這兩個池是外部不能訪問的，外部通過一個控制(或者叫管理)的類進行操作。繼承了前兩篇博文中的異步通信的思想，最底的一個類存放了一次連接的信息，包括兩個SocketAsyncEventArgs實例和通信另一端的Socket實例。下面將逐部分介紹。

 

    /// <summary>
    /// 連接單元
    /// </summary>
     class ConnectionUnit:IDisposable
    {
        private string _uid;//單元的編號，默認為-1
        private bool _state;//單元的狀態，true表示使用，false表示空閑
        private SocketAsyncEventArgs _sendArg;//專用於發送
        private SocketAsyncEventArgs _recArg;//專用於接收
        internal Socket client { get; set; }//客戶端的socket實例
        internal ArrayList tempArray { get; set; }//暫存已接收的數據，避免粘包用的


        public ConnectionUnit(string UID)
        {
            _uid = UID;
            tempArray = new ArrayList();
        }

        public ConnectionUnit() : this("-1") { }

        public void Dispose()
        {
            if (_sendArg != null)
                _sendArg.Dispose();
            if (_recArg != null)
                _recArg.Dispose();

            _sendArg = null;
            _recArg = null;
        }
    }

 

　　這個與之前一篇講異步通信的博文一樣，一個Socket兩個SocketAsyncEventArgs。為了取客戶端的Socket方便一點，兩個SocketAsyncEventArgs一個用於收，一個用於發，田大哥說這樣可以實現雙工。的確是這樣，當初把類定成這樣也是為了在收的同時也能發。以上代碼刪掉了把字段裝成屬性的那部分。

　　SocketAsyncEventArgsPool這個類就是上面鏈接單元的一個池，這個池才是整個連接池里最核心的池，也是真正意義上的池。池里面用了兩個集合來存放這一堆連接單元。分別是空閑棧(freeCollecton)和在線字典集(busyCollection)。對於這樣的一個池，我就認為，只要在我需要用的時候把對象取出來，頂多在去的時候給一些參數，讓池幫忙把這個對象配置好，我就拿來使用，等到我用完了，把他放回池里，池就把對象還原，等待下次再使用。正體感覺這個池對外有點像個棧，取的時候就Pop，放的時候就Push，此外還提供了兩個與棧不相干的的方法，根據編號獲取在線的連接單元和獲取所有在線連接單元的編號。

    class SocketAsyncEventArgsPool:IDisposable
    {
        private Dictionary<string, ConnectionUnit> busyCollection;
        private Stack<ConnectionUnit> freeCollecton;

        internal SocketAsyncEventArgsPool()
        {
            busyCollection = new Dictionary<string, ConnectionUnit>();
            freeCollecton = new Stack<ConnectionUnit>();
        }

        /// <summary>
        /// 取出
        /// </summary>
        internal ConnectionUnit Pop(string uid)
        {
            ConnectionUnit unit = freeCollecton.Pop();
            unit.State = true;
            unit.Uid = uid;
            busyCollection.Add(uid, unit);
            return unit;
        }

        /// <summary>
        /// 放回
        /// </summary>
        internal void Push(ConnectionUnit unit)
        {
            if (!string.IsNullOrEmpty(unit.Uid) && unit.Uid != "-1")
                busyCollection.Remove(unit.Uid);
            unit.Uid = "-1";
            unit.State = false;
            freeCollecton.Push(unit);
        }

        /// <summary>
        /// 獲取
        /// </summary>
        internal ConnectionUnit GetConnectionUnitByUID(string uid)
        {
            if (busyCollection.ContainsKey(uid))
                return busyCollection[uid];
            return null;
        }

        /// <summary>
        /// 
        /// </summary>
        internal string[] GetOnLineList()
        {
            return busyCollection.Keys.ToArray();
        }

        public void Dispose()
        {
            foreach (KeyValuePair<string,ConnectionUnit> item in busyCollection)
                item.Value.Dispose();

            busyCollection.Clear();

            while (freeCollecton.Count > 0)
                freeCollecton.Pop().Dispose();
        }

 

　　BufferManager這個是專給接收的SocketAsyncEventArgs用的緩沖池，整一個連接池里面所有接收用的緩沖區都用這個BufferManager，參照田大哥的思想，現在內存里開辟一大片區域存放byte，然后給每一個接收用的SocketAsyncEventArgs分配一塊。

    class BufferManager:IDisposable
    {
        private byte[] buffers;//緩沖池
        private int bufferSize;//每個單元使用的大小
        private int allSize;//池的總大小
        private int currentIndex;//當前可用的索引
        private Stack<int> freeIndexs;//已使用過的空閑索引

        /// <summary>
        /// 構造緩存池
        /// </summary>
        /// <param name="buffersSize">池總大小</param>
        /// <param name="defaultSize">默認單元大小</param>
        internal BufferManager(int buffersSize, int defaultSize) 
        {
            this.bufferSize=defaultSize;
            this.allSize=buffersSize;
            currentIndex=0;
            this.buffers = new byte[allSize];
            freeIndexs = new Stack<int>();
        }

        /// <summary>
        /// 給SocketAsyncEventArgs設置緩沖區
        /// </summary>
        internal bool SetBuffer(SocketAsyncEventArgs e)
        {
            //首先看看空閑棧里有沒有空閑的區域，有就使用
            if (freeIndexs.Count > 0)
            {
                e.SetBuffer(buffers, freeIndexs.Pop(), bufferSize);
            }
            else
            {
                //沒有就得從buffers里取，如果buffers用光了當然取不了
                if ((allSize - currentIndex) < bufferSize) return false;
                e.SetBuffer(buffers, currentIndex, bufferSize);
                currentIndex += bufferSize;
            }
            return true;
        }

        /// <summary>
        /// 釋放SocketAsyncEventArgs的緩沖區
        /// </summary>
        /// <param name="e"></param>
        internal void FreeBuffer(SocketAsyncEventArgs e)
        {
            //把索引放到空閑索引棧里面，供下次取的時候重復利用
            freeIndexs.Push(e.Offset);
            //同時清空這部分區域的數據免得上次使用時遺留的數據會摻
            //和到下次讀取的數據中
            for (int i = e.Offset; i < e.Offset + bufferSize; i++)
            {
                if (buffers[i] == 0) break;
                buffers[i] = 0;
            }
            e.SetBuffer(null, 0, 0);
        }

        public void Dispose()
        {
            buffers = null;
            freeIndexs = null;
        }
    }

 

　　其實上面兩個池都是很大程度參照了《高性能Socket設計實現》中的內容。下面這個SocketPoolController是對外的類，這個類的設計參照的就沒那么多了。而對於一個Socket通信(服務端的)來說，無非都是三件事，接受連接，接收數據，發送數據。這三件事我再操作類里面是這樣做的

　　接受連接：接受再運行池的時候就開始了，異步循環地接受，執行一次異步接受就阻塞，等到接受完成才被喚醒。

        /// <summary>
        /// 異步Accept客戶端的連接
        /// </summary>
        void MyAsyncAccept()
        {
            //這里使用Action的方式異步循環接受客戶端的連接
            //模仿同事的做法沒開線程，不知這種方式是好是壞
            Action callback = new Action(delegate()
            {
                while (true)
                {
                    //每次接受都要新開一個SocketAsyncEventArgs,否則會報錯
                    //其實我也想重復利用的
                    SocketAsyncEventArgs e = new SocketAsyncEventArgs();
                    e.Completed += new EventHandler<SocketAsyncEventArgs>(Accept_Completed);
                    
                    acceptLock.Reset();
                    server.AcceptAsync(e);
                    //在異步接受完成之前阻塞當前線程
                    acceptLock.WaitOne();
                }
            });
            callback.BeginInvoke(null, null);
        }

        void Accept_Completed(object sender, SocketAsyncEventArgs e)
        {
            Socket client = e.AcceptSocket;
            try
            {
                if (client.Connected)
                {
                    IPEndPoint point = client.RemoteEndPoint as IPEndPoint;
                    string uid = point.Address + ":" + point.Port;
                    ConnectionUnit unit = pool.Pop(uid);
                    unit.client = client;
                    unit.State = true;
                    unit.Uid = uid;
                    unit.RecArg.UserToken = unit;
                    unit.SendArg.UserToken = unit;
                    buffer.SetBuffer(unit.RecArg);

                    //在接受成功之后就開始接收數據了
                    client.ReceiveAsync(unit.RecArg);
                    //設置並發限制信號和增加當前連接數
                    semaphoreAccept.WaitOne();
                    Interlocked.Increment(ref currentConnect);

                    if (OnAccept != null) OnAccept(uid);
                }
                else if (client != null)
                {
                    client.Close();
                    client.Dispose();
                    client = null;
                }
            }
            catch (Exception ex) { Console.WriteLine(ex.ToString()); }
            //設置Accept信號，以便下次Accept的執行
            acceptLock.Set();
            e.Dispose();
        }

 

　　接收消息：在異步接受成功的時候開始接收，每次接收完成之后就進行下一次接收，直到客戶端斷開連接才終止。

        void RecArg_Completed(object sender, SocketAsyncEventArgs e)
        {
            Socket client = sender as Socket;
            ConnectionUnit unit = e.UserToken as ConnectionUnit;
            //這里大致與上一篇異步通信的一樣，只是對緩沖區的處理有一點差異
            if (e.SocketError == SocketError.Success)
            {
                int rec = e.BytesTransferred;
                if (rec == 0)
                {
                    CloseSocket(unit);
                    return;
                }
                if (client.Available > 0)
                {
                    unit.tempArray.AddRange(e.Buffer);
                    buffer.FreeBuffer(unit.RecArg);
                    buffer.SetBuffer(unit.RecArg);
                    client.SendAsync(unit.RecArg);
                    return;
                }
                byte[] data = e.Buffer;
                int len = rec;
                int offset = e.Offset;
                if (unit.tempArray.Count != 0)
                {
                    foreach (byte item in data)
                    {
                        if (item == 0) break;
                        unit.tempArray.Add(item);
                    }
                    data = unit.tempArray.ToArray(typeof(byte)) as byte[];
                    rec = data.Length;
                    offset = 0;
                    unit.tempArray.Clear();
                }

                string dataStr = Encoding.ASCII.GetString(data,offset,len);
                if (OnReceive != null)
                    OnReceive(unit.Uid, dataStr);

                if (!unit.State) return;
                buffer.FreeBuffer(e);
                buffer.SetBuffer(e);
                client.ReceiveAsync(e);
            }
            //這里還多個了一個關閉當前連接
            else
            {
                CloseSocket(unit);
            }
        }

        /// <summary>
        /// 關閉一個連接單元
        /// </summary>
        private void CloseSocket( ConnectionUnit unit )
        {
            //關閉並釋放客戶端socket的字眼
            if (unit.client != null)
            {
                unit.client.Shutdown(SocketShutdown.Both);
                unit.client.Dispose();
                unit.client = null;
            }
            //Console.WriteLine(unit.Uid+" disconnect ");
            //把連接放回連接池
            pool.Push(unit);
            //釋放並發信號
            semaphoreAccept.Release();
            //減少當前連接數
            Interlocked.Decrement(ref currentConnect);
        }

 

　　發送消息：外放方法，在需要的時候自行調用方法發送。

        /// <summary>
        /// 發送消息
        /// </summary>
        /// <param name="uid"></param>
        /// <param name="message"></param>
        public void SendMessage(string uid, string message)
        {
            sendLock.Reset();
            ConnectionUnit unit=pool.GetConnectionUnitByUID(uid);
            //如果獲取不了連接單元就不發送了，
            if (unit == null)
            { 
                if(OnSend!=null) OnSend(uid,"100");
                sendLock.Set();
                return;
            }
            byte[] datas = Encoding.ASCII.GetBytes(message);
            unit.SendArg.SetBuffer(datas, 0, datas.Length);
            unit.client.SendAsync(unit.SendArg);
            //阻塞當前線程，等到發送完成才釋放
            sendLock.WaitOne();
        }

        void SendArg_Completed(object sender, SocketAsyncEventArgs e)
        {
            Socket client = sender as Socket;
            ConnectionUnit unit = e.UserToken as ConnectionUnit;
            //這里的消息碼有三個，2字頭的是成功的，1字頭是不成功的
            //101是未知錯誤，100是客戶端不在線
            if (e.SocketError == SocketError.Success)
                if (OnSend != null) OnSend(unit.Uid, "200");
            else if (OnSend != null) OnSend(unit.Uid, "101");
            //釋放信號，以便下次發送消息執行
            sendLock.Set();
        }

 

　　下面則是類里面的一些字段信息和構造函數

        /// <summary>
        /// 初始化池的互斥體
        /// </summary>
        private Mutex mutex = new Mutex();

        /// <summary>
        /// Accept限制信號
        /// </summary>
        private Semaphore semaphoreAccept;

        /// <summary>
        /// Accept信號
        /// </summary>
        private static ManualResetEvent acceptLock = new ManualResetEvent(false);

        /// <summary>
        /// Send信號
        /// </summary>
        private static ManualResetEvent sendLock = new ManualResetEvent(false);

        /// <summary>
        /// 最大並發數(連接數)
        /// </summary>
        private int maxConnect;

        /// <summary>
        /// 當前連接數(並發數)
        /// </summary>
        private int currentConnect;

        /// <summary>
        /// 緩沖池
        /// </summary>
        private BufferManager buffer;

        /// <summary>
        /// SocketasyncEventArgs池
        /// </summary>
        private SocketAsyncEventArgsPool pool;

        /// <summary>
        /// 服務端Socket
        /// </summary>
        private Socket server;

        /// <summary>
        /// 完成接受的委托
        /// </summary>
        public delegate void AcceptHandler(string uid);

        /// <summary>
        /// 完成發送的委托
        /// </summary>
        public delegate void SendHandler(string uid, string result);

        /// <summary>
        /// 完成接收的委托
        /// </summary>
        public delegate void RecevieHandler(string uid, string data);

        /// <summary>
        /// 完成接受事件
        /// </summary>
        public event AcceptHandler OnAccept;

        /// <summary>
        /// 完成發送事件
        /// </summary>
        public event SendHandler OnSend;

        /// <summary>
        /// 完成接收事件
        /// </summary>
        public event RecevieHandler OnReceive;

        /// <summary>
        /// 構造函數
        /// </summary>
        /// <param name="buffersize">單元緩沖區大小</param>
        /// <param name="maxCount">並發總數</param>
        public SocketPoolController(int buffersize, int maxCount)
        {
            buffer = new BufferManager(buffersize * maxCount,buffersize);
            this.currentConnect = 0;
            this.maxConnect = maxCount;
            this.currentConnect = 0;
            this.pool = new SocketAsyncEventArgsPool();
            //設置並發數信號，經試驗過是並發數-1才對
            this.semaphoreAccept = new Semaphore(maxCount-1, maxCount-1);
            InitPool();
        }

 

構造函數里用到的方法

        /// <summary>
        /// 初始化SocketAsyncEventArgs池
        /// 這里主要是給空閑棧填充足夠的實例
        /// </summary>
        private void InitPool()
        {
            ConnectionUnit unit = null;
            for (int i = 0; i < maxConnect; i++)
            {
                unit = new ConnectionUnit();
                unit.Uid = "-1";
                unit.RecArg = new SocketAsyncEventArgs();
                unit.RecArg.Completed += new EventHandler<SocketAsyncEventArgs>(RecArg_Completed);
                unit.SendArg = new SocketAsyncEventArgs();
                unit.SendArg.Completed += new EventHandler<SocketAsyncEventArgs>(SendArg_Completed);
                this.pool.Push(unit);
            }
        }

 

　　其他外放專門控制池的方法

        /// <summary>
        /// 啟動池
        /// </summary>
        /// <param name="ipAddress">服務端的IP</param>
        /// <param name="port">端口</param>
        public void RunPool(string ipAddress, int port)
        {
            IPEndPoint endpoint = new IPEndPoint(IPAddress.Parse(ipAddress), port);
            server = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
            server.Bind(endpoint);
            server.Listen(100);

            //調用方法異步Accept客戶端的連接
            MyAsyncAccept();
            //設置信號，防止再池在已經啟動的情況下再次啟動
            mutex.WaitOne();
        }

        /// <summary>
        /// 停止池
        /// </summary>
        public void StopPool()
        {
            //把服務端的socket關了
            if (server != null)
                server.Close();
            //釋放互斥信號，等待下次啟動
            mutex.ReleaseMutex();
            //釋放資源
            Dispose();
        }

 

　　要把這個操作類與現實的事物類比的話，這個與我們平常跟某個或一堆人聊天時差不多，別人說的東西，不用你自己控制你都會聽得到(排除帶上耳機，塞住耳朵這種極端的情況)，所以接收消息那個方法就不需要了，而你要說什么這個要靠個人控制，而事件那些也好類比，OnAccept就相當於某人加入到這次聊天中你會做些什么(say hello 或者無視)，OnRecevie就聽到別人說什么自己有什么反應，OnSend就自己說完話之后有什么反應(有時候發現說錯了會糾正，有時覺得自己說得好笑的也笑一場諸如此類)。

　　在使用的時候可以這樣子

            SocketPoolController pool;
            pool = new SocketPoolController(32 * 1024, 1000);
            pool.OnReceive += new SocketPoolController.RecevieHandler(pool_OnReceive);
            pool.OnSend += new SocketPoolController.SendHandler(pool_OnSend);
            pool.OnAccept += new SocketPoolController.AcceptHandler(pool_OnAccept);
            pool.RunPool("127.0.0.1", 8081);
            Console.WriteLine("Pool has run\r\npress any key to stop...");
            Console.ReadKey();
            pool.StopPool();
            Console.WriteLine("Pool has stop\r\npress any key to exit...");
            Console.ReadLine();

 

　　在池里面有一部分地方看似跟用同步的差不多，像接受客戶端的連接，發送消息這些地方。可是用這種異步，萬一客戶端突然斷開連接也不會有同步那樣馬上拋異常。還有一點的是在這份代碼里面缺少了對異常的捕捉，有一部分錯誤我在測試的過程中設了判斷避開了。以前跟某只貓提過Socket編程會與多線程一起使用，我也覺得是這樣，在之前一篇博文 《Socket一對多通信》里我也用到線程，后來的異步通信也是有線程的，不過不是.net framework自行創建的。看了田大哥的博文給我的另一個收獲是信號量的使用，在以前不懂得使用信號量，只會設置一大堆標識狀態的布爾值或整形的變量來計數判斷。田大哥的博文介紹的時高性能的Socket，而我的這個應該性能不會高到哪里去。上面有什么說錯的請各位指出，有什么說漏的，請各位提點，多多指導。謝謝！

    /// <summary>
    /// 連接單元
    /// </summary>
    class ConnectionUnit:IDisposable
    {
        private string _uid;//單元的編號，默認為-1
        private bool _state;//單元的狀態，true表示使用，false表示空閑
        private SocketAsyncEventArgs _sendArg;//專用於發送
        private SocketAsyncEventArgs _recArg;//專用於接收
        internal Socket client { get; set; }//客戶端的socket實例
        internal ArrayList tempArray { get; set; }//暫存已接收的數據，避免粘包用的

        public string Uid
        {
            get { return _uid; }
            set { _uid = value; }
        }

        public ConnectionUnit(string UID)
        {
            _uid = UID;
            tempArray = new ArrayList();
        }

        public ConnectionUnit() : this("-1") { }

        public bool State
        {
            get { return _state; }
            set { _state = value; }
        }

        public SocketAsyncEventArgs SendArg
        {
            get { return _sendArg; }
            set { _sendArg = value; }
        }

        public SocketAsyncEventArgs RecArg
        {
            get { return _recArg; }
            set { _recArg = value; }
        }

        public void Dispose()
        {
            if (_sendArg != null)
                _sendArg.Dispose();
            if (_recArg != null)
                _recArg.Dispose();

            _sendArg = null;
            _recArg = null;
        }
    }

    class BufferManager:IDisposable
    {
        private byte[] buffers;
        private int bufferSize;
        private int allSize;
        private int currentIndex;
        private Stack<int> freeIndexs;

        /// <summary>
        /// 構造緩存池
        /// </summary>
        /// <param name="buffersSize">池總大小</param>
        /// <param name="defaultSize">默認單元大小</param>
        internal BufferManager(int buffersSize, int defaultSize) 
        {
            this.bufferSize=defaultSize;
            this.allSize=buffersSize;
            currentIndex=0;
            this.buffers = new byte[allSize];
            freeIndexs = new Stack<int>();
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="e"></param>
        /// <param name="offSet"></param>
        /// <returns></returns>
        internal bool SetBuffer(SocketAsyncEventArgs e)
        {
            if (freeIndexs.Count > 0)
            {
                e.SetBuffer(buffers, freeIndexs.Pop(), bufferSize);
            }
            else
            {
                if ((allSize - currentIndex) < bufferSize) return false;
                e.SetBuffer(buffers, currentIndex, bufferSize);
                currentIndex += bufferSize;
            }
            return true;
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="e"></param>
        internal void FreeBuffer(SocketAsyncEventArgs e)
        {
            freeIndexs.Push(e.Offset);
            for (int i = e.Offset; i < e.Offset + bufferSize; i++)
            {
                if (buffers[i] == 0) break;
                buffers[i] = 0;
            }
            e.SetBuffer(null, 0, 0);
        }

        public void Dispose()
        {
            buffers = null;
            freeIndexs = null;
        }
    }

    class SocketAsyncEventArgsPool:IDisposable
    {
        private Dictionary<string, ConnectionUnit> busyCollection;
        private Stack<ConnectionUnit> freeCollecton;

        internal SocketAsyncEventArgsPool()
        {
            busyCollection = new Dictionary<string, ConnectionUnit>();
            freeCollecton = new Stack<ConnectionUnit>();
        }

        /// <summary>
        /// 取出
        /// </summary>
        internal ConnectionUnit Pop(string uid)
        {
            ConnectionUnit unit = freeCollecton.Pop();
            unit.State = true;
            unit.Uid = uid;
            busyCollection.Add(uid, unit);
            return unit;
        }

        /// <summary>
        /// 放回
        /// </summary>
        internal void Push(ConnectionUnit unit)
        {
            if (!string.IsNullOrEmpty(unit.Uid) && unit.Uid != "-1")
                busyCollection.Remove(unit.Uid);
            unit.Uid = "-1";
            unit.State = false;
            freeCollecton.Push(unit);
        }

        /// <summary>
        /// 獲取
        /// </summary>
        internal ConnectionUnit GetConnectionUnitByUID(string uid)
        {
            if (busyCollection.ContainsKey(uid))
                return busyCollection[uid];
            return null;
        }

        /// <summary>
        /// 
        /// </summary>
        internal string[] GetOnLineList()
        {
            return busyCollection.Keys.ToArray();
        }

        public void Dispose()
        {
            foreach (KeyValuePair<string,ConnectionUnit> item in busyCollection)
                item.Value.Dispose();

            busyCollection.Clear();

            while (freeCollecton.Count > 0)
                freeCollecton.Pop().Dispose();
        }
    }

    public class SocketPoolController:IDisposable
    {
        /// <summary>
        /// 初始化池的互斥體
        /// </summary>
        private Mutex mutex = new Mutex();

        /// <summary>
        /// Accept限制信號
        /// </summary>
        private Semaphore semaphoreAccept;

        /// <summary>
        /// Accept信號
        /// </summary>
        private static ManualResetEvent acceptLock = new ManualResetEvent(false);

        /// <summary>
        /// Send信號
        /// </summary>
        private static ManualResetEvent sendLock = new ManualResetEvent(false);

        /// <summary>
        /// 最大並發數(連接數)
        /// </summary>
        private int maxConnect;

        /// <summary>
        /// 當前連接數(並發數)
        /// </summary>
        private int currentConnect;

        /// <summary>
        /// 緩沖池
        /// </summary>
        private BufferManager buffer;

        /// <summary>
        /// SocketasyncEventArgs池
        /// </summary>
        private SocketAsyncEventArgsPool pool;

        /// <summary>
        /// 服務端Socket
        /// </summary>
        private Socket server;

        /// <summary>
        /// 完成接受的委托
        /// </summary>
        public delegate void AcceptHandler(string uid);

        /// <summary>
        /// 完成發送的委托
        /// </summary>
        public delegate void SendHandler(string uid, string result);

        /// <summary>
        /// 完成接收的委托
        /// </summary>
        public delegate void RecevieHandler(string uid, string data);

        /// <summary>
        /// 完成接受事件
        /// </summary>
        public event AcceptHandler OnAccept;

        /// <summary>
        /// 完成發送事件
        /// </summary>
        public event SendHandler OnSend;

        /// <summary>
        /// 完成接收事件
        /// </summary>
        public event RecevieHandler OnReceive;

        /// <summary>
        /// 構造函數
        /// </summary>
        /// <param name="buffersize">單元緩沖區大小</param>
        /// <param name="maxCount">並發總數</param>
        public SocketPoolController(int buffersize, int maxCount)
        {
            buffer = new BufferManager(buffersize * maxCount,buffersize);
            this.currentConnect = 0;
            this.maxConnect = maxCount;
            this.currentConnect = 0;
            this.pool = new SocketAsyncEventArgsPool();
            //設置並發數信號，經試驗過是並發數-1才對
            this.semaphoreAccept = new Semaphore(maxCount-1, maxCount-1);
            InitPool();
        }

        /// <summary>
        /// 初始化SocketAsyncEventArgs池
        /// 這里主要是給空閑棧填充足夠的實例
        /// </summary>
        private void InitPool()
        {
            ConnectionUnit unit = null;
            for (int i = 0; i < maxConnect; i++)
            {
                unit = new ConnectionUnit();
                unit.Uid = "-1";
                unit.RecArg = new SocketAsyncEventArgs();
                unit.RecArg.Completed += new EventHandler<SocketAsyncEventArgs>(RecArg_Completed);
                unit.SendArg = new SocketAsyncEventArgs();
                unit.SendArg.Completed += new EventHandler<SocketAsyncEventArgs>(SendArg_Completed);
                this.pool.Push(unit);
            }
        }

        /// <summary>
        /// 啟動池
        /// </summary>
        /// <param name="ipAddress">服務端的IP</param>
        /// <param name="port">端口</param>
        public void RunPool(string ipAddress, int port)
        {
            IPEndPoint endpoint = new IPEndPoint(IPAddress.Parse(ipAddress), port);
            server = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
            server.Bind(endpoint);
            server.Listen(100);

            //調用方法異步Accept客戶端的連接
            MyAsyncAccept();
            //設置信號，防止再池在已經啟動的情況下再次啟動
            mutex.WaitOne();
        }

        /// <summary>
        /// 異步Accept客戶端的連接
        /// </summary>
        void MyAsyncAccept()
        {
            //這里使用Action的方式異步循環接受客戶端的連接
            //模仿同事的做法沒開線程，不知這種方式是好是壞
            Action callback = new Action(delegate()
            {
                while (true)
                {
                    //每次接受都要新開一個SocketAsyncEventArgs,否則會報錯
                    //其實我也想重復利用的
                    SocketAsyncEventArgs e = new SocketAsyncEventArgs();
                    e.Completed += new EventHandler<SocketAsyncEventArgs>(Accept_Completed);
                    
                    acceptLock.Reset();
                    server.AcceptAsync(e);
                    //在異步接受完成之前阻塞當前線程
                    acceptLock.WaitOne();
                }
            });
            callback.BeginInvoke(null, null);
        }


        /// <summary>
        /// 停止池
        /// </summary>
        public void StopPool()
        {
            //把服務端的socket關了
            if (server != null)
                server.Close();
            //釋放互斥信號，等待下次啟動
            mutex.ReleaseMutex();
            //釋放資源
            Dispose();
        }

        /// <summary>
        /// 發送消息
        /// </summary>
        /// <param name="uid"></param>
        /// <param name="message"></param>
        public void SendMessage(string uid, string message)
        {
            sendLock.Reset();
            ConnectionUnit unit=pool.GetConnectionUnitByUID(uid);
            //如果獲取不了連接單元就不發送了，
            if (unit == null)
            { 
                if(OnSend!=null) OnSend(uid,"100");
                sendLock.Set();
                return;
            }
            byte[] datas = Encoding.ASCII.GetBytes(message);
            unit.SendArg.SetBuffer(datas, 0, datas.Length);
            unit.client.SendAsync(unit.SendArg);
            //阻塞當前線程，等到發送完成才釋放
            sendLock.WaitOne();
        }

        void SendArg_Completed(object sender, SocketAsyncEventArgs e)
        {
            Socket client = sender as Socket;
            ConnectionUnit unit = e.UserToken as ConnectionUnit;
            //這里的消息碼有三個，2字頭的是成功的，1字頭是不成功的
            //101是未知錯誤，100是客戶端不在線
            if (e.SocketError == SocketError.Success)
                if (OnSend != null) OnSend(unit.Uid, "200");
            else if (OnSend != null) OnSend(unit.Uid, "101");
            //釋放信號，以便下次發送消息執行
            sendLock.Set();
        }

        void RecArg_Completed(object sender, SocketAsyncEventArgs e)
        {
            Socket client = sender as Socket;
            ConnectionUnit unit = e.UserToken as ConnectionUnit;
            //這里大致與上一篇異步通信的一樣，只是對緩沖區的處理有一點差異
            if (e.SocketError == SocketError.Success)
            {
                int rec = e.BytesTransferred;
                if (rec == 0)
                {
                    CloseSocket(unit);
                    return;
                }
                if (client.Available > 0)
                {
                    unit.tempArray.AddRange(e.Buffer);
                    buffer.FreeBuffer(unit.RecArg);
                    buffer.SetBuffer(unit.RecArg);
                    client.SendAsync(unit.RecArg);
                    return;
                }
                byte[] data = e.Buffer;
                int len = rec;
                if (unit.tempArray.Count != 0)
                {
                    foreach (byte item in data)
                    {
                        if (item == 0) break;
                        unit.tempArray.Add(item);
                    }
                    data = unit.tempArray.ToArray(typeof(byte)) as byte[];
                    rec = data.Length;
                    unit.tempArray.Clear();
                }

                string dataStr = Encoding.ASCII.GetString(data, 0, len);
                if (OnReceive != null)
                    OnReceive(unit.Uid, dataStr);

                if (!unit.State) return;
                buffer.FreeBuffer(e);
                buffer.SetBuffer(e);
                client.ReceiveAsync(e);
            }
            //這里還多個了一個關閉當前連接
            else
            {
                CloseSocket(unit);
            }
        }

        void Accept_Completed(object sender, SocketAsyncEventArgs e)
        {
            Socket client = e.AcceptSocket;
            try
            {
                if (client.Connected)
                {
                    IPEndPoint point = client.RemoteEndPoint as IPEndPoint;
                    string uid = point.Address + ":" + point.Port;
                    ConnectionUnit unit = pool.Pop(uid);
                    unit.client = client;
                    unit.State = true;
                    unit.Uid = uid;
                    unit.RecArg.UserToken = unit;
                    unit.SendArg.UserToken = unit;
                    buffer.SetBuffer(unit.RecArg);

                    //在接受成功之后就開始接收數據了
                    client.ReceiveAsync(unit.RecArg);
                    //設置並發限制信號和增加當前連接數
                    semaphoreAccept.WaitOne();
                    Interlocked.Increment(ref currentConnect);

                    if (OnAccept != null) OnAccept(uid);
                }
                else if (client != null)
                {
                    client.Close();
                    client.Dispose();
                    client = null;
                }
            }
            catch (Exception ex) { Console.WriteLine(ex.ToString()); }
            //設置Accept信號，以便下次Accept的執行
            acceptLock.Set();
            e.Dispose();
        }

        /// <summary>
        /// 關閉一個連接單元
        /// </summary>
        private void CloseSocket( ConnectionUnit unit )
        {
            //關閉並釋放客戶端socket的字眼
            if (unit.client != null)
            {
                unit.client.Shutdown(SocketShutdown.Both);
                unit.client.Dispose();
                unit.client = null;
            }
            //Console.WriteLine(unit.Uid+" disconnect ");
            //把連接放回連接池
            pool.Push(unit);
            //釋放並發信號
            semaphoreAccept.Release();
            //減少當前連接數
            Interlocked.Decrement(ref currentConnect);
        }
        
        public void Dispose()
        {
            if (pool != null)
            {
                pool.Dispose();
                pool = null;
            }
            if (buffer != null)
            {
                buffer.Dispose();
                buffer = null;
            }
            if (server != null)
            {
                server.Dispose();
                server = null;
            }

        }
    }