|
“池”這個概念好像最早是在操作系統(tǒng)的課上聽過的,，到后來出來工作的第二天組長也跟我提起“池”這個東東,。它給我的感覺是某種對象的集合,，如果要用的話就取出，不用的話就放回,。在學(xué)多線程的時候有接觸過線程池,，在寫《Socket 一對多通信》的時候想到了Socket連接池這回事，不過在網(wǎng)上谷歌了一下,，發(fā)現(xiàn)這類的文章貌似不多,，看了一下園友的博文《高性能Socket設(shè)計實(shí)現(xiàn)》，獲益良多,，下了份源碼來看,，雖然有一部分看不明白,，而且由于個人水平跑不了那份代碼，但是從中我學(xué)到了不少,，至少我寫的“池”有一部分是用了這位田大哥的思想,。
先來分析各個類之間的結(jié)構(gòu)，整個連接池里面實(shí)際上是有兩個池,，一個是在異步通信中可以重復(fù)利用的SocketAsyncEventArgs池(當(dāng)然處于別的方面考慮,，池里面并不是單純放SocketAsyncEventArgs的實(shí)例集合)，另一個是接收數(shù)據(jù)時用到的byte[]緩沖池,。而這兩個池是外部不能訪問的,，外部通過一個控制(或者叫管理)的類進(jìn)行操作。繼承了前兩篇博文中的異步通信的思想,，最底的一個類存放了一次連接的信息,，包括兩個SocketAsyncEventArgs實(shí)例和通信另一端的Socket實(shí)例。下面將逐部分介紹,。
 1 /// <summary> 2 /// 連接單元 3 /// </summary> 4 class ConnectionUnit:IDisposable 5 { 6 private string _uid;//單元的編號,，默認(rèn)為-1 7 private bool _state;//單元的狀態(tài)，true表示使用,，false表示空閑 8 private SocketAsyncEventArgs _sendArg;//專用于發(fā)送 9 private SocketAsyncEventArgs _recArg;//專用于接收 10 internal Socket client { get; set; }//客戶端的socket實(shí)例 11 internal ArrayList tempArray { get; set; }//暫存已接收的數(shù)據(jù),，避免粘包用的 12 13 14 public ConnectionUnit(string UID) 15 { 16 _uid = UID; 17 tempArray = new ArrayList(); 18 } 19 20 public ConnectionUnit() : this("-1") { } 21 22 public void Dispose() 23 { 24 if (_sendArg != null) 25 _sendArg.Dispose(); 26 if (_recArg != null) 27 _recArg.Dispose(); 28 29 _sendArg = null; 30 _recArg = null; 31 } 32 }
這個與之前一篇講異步通信的博文一樣，一個Socket兩個SocketAsyncEventArgs,。為了取客戶端的Socket方便一點(diǎn)，兩個SocketAsyncEventArgs一個用于收,，一個用于發(fā),，田大哥說這樣可以實(shí)現(xiàn)雙工。的確是這樣,，當(dāng)初把類定成這樣也是為了在收的同時也能發(fā),。以上代碼刪掉了把字段裝成屬性的那部分。 SocketAsyncEventArgsPool這個類就是上面鏈接單元的一個池,，這個池才是整個連接池里最核心的池,，也是真正意義上的池。池里面用了兩個集合來存放這一堆連接單元,。分別是空閑棧(freeCollecton)和在線字典集(busyCollection),。對于這樣的一個池，我就認(rèn)為,，只要在我需要用的時候把對象取出來,，頂多在去的時候給一些參數(shù)，讓池幫忙把這個對象配置好,，我就拿來使用,，等到我用完了,，把他放回池里，池就把對象還原,，等待下次再使用,。正體感覺這個池對外有點(diǎn)像個棧，取的時候就Pop,，放的時候就Push,，此外還提供了兩個與棧不相干的的方法，根據(jù)編號獲取在線的連接單元和獲取所有在線連接單元的編號,。 1 class SocketAsyncEventArgsPool:IDisposable 2 { 3 private Dictionary<string, ConnectionUnit> busyCollection; 4 private Stack<ConnectionUnit> freeCollecton; 5 6 internal SocketAsyncEventArgsPool() 7 { 8 busyCollection = new Dictionary<string, ConnectionUnit>(); 9 freeCollecton = new Stack<ConnectionUnit>(); 10 } 11 12 /// <summary> 13 /// 取出 14 /// </summary> 15 internal ConnectionUnit Pop(string uid) 16 { 17 ConnectionUnit unit = freeCollecton.Pop(); 18 unit.State = true; 19 unit.Uid = uid; 20 busyCollection.Add(uid, unit); 21 return unit; 22 } 23 24 /// <summary> 25 /// 放回 26 /// </summary> 27 internal void Push(ConnectionUnit unit) 28 { 29 if (!string.IsNullOrEmpty(unit.Uid) && unit.Uid != "-1") 30 busyCollection.Remove(unit.Uid); 31 unit.Uid = "-1"; 32 unit.State = false; 33 freeCollecton.Push(unit); 34 } 35 36 /// <summary> 37 /// 獲取 38 /// </summary> 39 internal ConnectionUnit GetConnectionUnitByUID(string uid) 40 { 41 if (busyCollection.ContainsKey(uid)) 42 return busyCollection[uid]; 43 return null; 44 } 45 46 /// <summary> 47 /// 48 /// </summary> 49 internal string[] GetOnLineList() 50 { 51 return busyCollection.Keys.ToArray(); 52 } 53 54 public void Dispose() 55 { 56 foreach (KeyValuePair<string,ConnectionUnit> item in busyCollection) 57 item.Value.Dispose(); 58 59 busyCollection.Clear(); 60 61 while (freeCollecton.Count > 0) 62 freeCollecton.Pop().Dispose(); 63 }
BufferManager這個是專給接收的SocketAsyncEventArgs用的緩沖池,，整一個連接池里面所有接收用的緩沖區(qū)都用這個BufferManager，參照田大哥的思想,，現(xiàn)在內(nèi)存里開辟一大片區(qū)域存放byte,，然后給每一個接收用的SocketAsyncEventArgs分配一塊。 1 class BufferManager:IDisposable 2 { 3 private byte[] buffers;//緩沖池 4 private int bufferSize;//每個單元使用的大小 5 private int allSize;//池的總大小 6 private int currentIndex;//當(dāng)前可用的索引 7 private Stack<int> freeIndexs;//已使用過的空閑索引 8 9 /// <summary> 10 /// 構(gòu)造緩存池 11 /// </summary> 12 /// <param name="buffersSize">池總大小</param> 13 /// <param name="defaultSize">默認(rèn)單元大小</param> 14 internal BufferManager(int buffersSize, int defaultSize) 15 { 16 this.bufferSize=defaultSize; 17 this.allSize=buffersSize; 18 currentIndex=0; 19 this.buffers = new byte[allSize]; 20 freeIndexs = new Stack<int>(); 21 } 22 23 /// <summary> 24 /// 給SocketAsyncEventArgs設(shè)置緩沖區(qū) 25 /// </summary> 26 internal bool SetBuffer(SocketAsyncEventArgs e) 27 { 28 //首先看看空閑棧里有沒有空閑的區(qū)域,，有就使用 29 if (freeIndexs.Count > 0) 30 { 31 e.SetBuffer(buffers, freeIndexs.Pop(), bufferSize); 32 } 33 else 34 { 35 //沒有就得從buffers里取,，如果buffers用光了當(dāng)然取不了 36 if ((allSize - currentIndex) < bufferSize) return false; 37 e.SetBuffer(buffers, currentIndex, bufferSize); 38 currentIndex += bufferSize; 39 } 40 return true; 41 } 42 43 /// <summary> 44 /// 釋放SocketAsyncEventArgs的緩沖區(qū) 45 /// </summary> 46 /// <param name="e"></param> 47 internal void FreeBuffer(SocketAsyncEventArgs e) 48 { 49 //把索引放到空閑索引棧里面，供下次取的時候重復(fù)利用 50 freeIndexs.Push(e.Offset); 51 //同時清空這部分區(qū)域的數(shù)據(jù)免得上次使用時遺留的數(shù)據(jù)會摻 52 //和到下次讀取的數(shù)據(jù)中 53 for (int i = e.Offset; i < e.Offset + bufferSize; i++) 54 { 55 if (buffers[i] == 0) break; 56 buffers[i] = 0; 57 } 58 e.SetBuffer(null, 0, 0); 59 } 60 61 public void Dispose() 62 { 63 buffers = null; 64 freeIndexs = null; 65 } 66 }
其實(shí)上面兩個池都是很大程度參照了《高性能Socket設(shè)計實(shí)現(xiàn)》中的內(nèi)容,。下面這個SocketPoolController是對外的類,，這個類的設(shè)計參照的就沒那么多了。而對于一個Socket通信(服務(wù)端的)來說,，無非都是三件事,，接受連接，接收數(shù)據(jù),，發(fā)送數(shù)據(jù),。這三件事我再操作類里面是這樣做的 接受連接：接受再運(yùn)行池的時候就開始了，異步循環(huán)地接受,，執(zhí)行一次異步接受就阻塞,，等到接受完成才被喚醒。 1 /// <summary> 2 /// 異步Accept客戶端的連接 3 /// </summary> 4 void MyAsyncAccept() 5 { 6 //這里使用Action的方式異步循環(huán)接受客戶端的連接 7 //模仿同事的做法沒開線程,，不知這種方式是好是壞 8 Action callback = new Action(delegate() 9 { 10 while (true) 11 { 12 //每次接受都要新開一個SocketAsyncEventArgs,否則會報錯 13 //其實(shí)我也想重復(fù)利用的 14 SocketAsyncEventArgs e = new SocketAsyncEventArgs(); 15 e.Completed += new EventHandler<SocketAsyncEventArgs>(Accept_Completed); 16 17 acceptLock.Reset(); 18 server.AcceptAsync(e); 19 //在異步接受完成之前阻塞當(dāng)前線程 20 acceptLock.WaitOne(); 21 } 22 }); 23 callback.BeginInvoke(null, null); 24 } 25 26 void Accept_Completed(object sender, SocketAsyncEventArgs e) 27 { 28 Socket client = e.AcceptSocket; 29 try 30 { 31 if (client.Connected) 32 { 33 IPEndPoint point = client.RemoteEndPoint as IPEndPoint; 34 string uid = point.Address + ":" + point.Port; 35 ConnectionUnit unit = pool.Pop(uid); 36 unit.client = client; 37 unit.State = true; 38 unit.Uid = uid; 39 unit.RecArg.UserToken = unit; 40 unit.SendArg.UserToken = unit; 41 buffer.SetBuffer(unit.RecArg); 42 43 //在接受成功之后就開始接收數(shù)據(jù)了 44 client.ReceiveAsync(unit.RecArg); 45 //設(shè)置并發(fā)限制信號和增加當(dāng)前連接數(shù) 46 semaphoreAccept.WaitOne(); 47 Interlocked.Increment(ref currentConnect); 48 49 if (OnAccept != null) OnAccept(uid); 50 } 51 else if (client != null) 52 { 53 client.Close(); 54 client.Dispose(); 55 client = null; 56 } 57 } 58 catch (Exception ex) { Console.WriteLine(ex.ToString()); } 59 //設(shè)置Accept信號,，以便下次Accept的執(zhí)行 60 acceptLock.Set(); 61 e.Dispose(); 62 }
接收消息：在異步接受成功的時候開始接收，每次接收完成之后就進(jìn)行下一次接收,，直到客戶端斷開連接才終止,。 1 void RecArg_Completed(object sender, SocketAsyncEventArgs e) 2 { 3 Socket client = sender as Socket; 4 ConnectionUnit unit = e.UserToken as ConnectionUnit; 5 //這里大致與上一篇異步通信的一樣，只是對緩沖區(qū)的處理有一點(diǎn)差異 6 if (e.SocketError == SocketError.Success) 7 { 8 int rec = e.BytesTransferred; 9 if (rec == 0) 10 { 11 CloseSocket(unit); 12 return; 13 } 14 if (client.Available > 0) 15 { 16 unit.tempArray.AddRange(e.Buffer); 17 buffer.FreeBuffer(unit.RecArg); 18 buffer.SetBuffer(unit.RecArg); 19 client.SendAsync(unit.RecArg); 20 return; 21 } 22 byte[] data = e.Buffer; 23 int len = rec; 24 int offset = e.Offset; 25 if (unit.tempArray.Count != 0) 26 { 27 foreach (byte item in data) 28 { 29 if (item == 0) break; 30 unit.tempArray.Add(item); 31 } 32 data = unit.tempArray.ToArray(typeof(byte)) as byte[]; 33 rec = data.Length; 34 offset = 0; 35 unit.tempArray.Clear(); 36 } 37 38 string dataStr = Encoding.ASCII.GetString(data,offset,len); 39 if (OnReceive != null) 40 OnReceive(unit.Uid, dataStr); 41 42 if (!unit.State) return; 43 buffer.FreeBuffer(e); 44 buffer.SetBuffer(e); 45 client.ReceiveAsync(e); 46 } 47 //這里還多個了一個關(guān)閉當(dāng)前連接 48 else 49 { 50 CloseSocket(unit); 51 } 52 } 53 54 /// <summary> 55 /// 關(guān)閉一個連接單元 56 /// </summary> 57 private void CloseSocket( ConnectionUnit unit ) 58 { 59 //關(guān)閉并釋放客戶端socket的字眼 60 if (unit.client != null) 61 { 62 unit.client.Shutdown(SocketShutdown.Both); 63 unit.client.Dispose(); 64 unit.client = null; 65 } 66 //Console.WriteLine(unit.Uid+" disconnect "); 67 //把連接放回連接池 68 pool.Push(unit); 69 //釋放并發(fā)信號 70 semaphoreAccept.Release(); 71 //減少當(dāng)前連接數(shù) 72 Interlocked.Decrement(ref currentConnect); 73 } 
發(fā)送消息：外放方法,，在需要的時候自行調(diào)用方法發(fā)送,。 1 /// <summary> 2 /// 發(fā)送消息 3 /// </summary> 4 /// <param name="uid"></param> 5 /// <param name="message"></param> 6 public void SendMessage(string uid, string message) 7 { 8 sendLock.Reset(); 9 ConnectionUnit unit=pool.GetConnectionUnitByUID(uid); 10 //如果獲取不了連接單元就不發(fā)送了， 11 if (unit == null) 12 { 13 if(OnSend!=null) OnSend(uid,"100"); 14 sendLock.Set(); 15 return; 16 } 17 byte[] datas = Encoding.ASCII.GetBytes(message); 18 unit.SendArg.SetBuffer(datas, 0, datas.Length); 19 unit.client.SendAsync(unit.SendArg); 20 //阻塞當(dāng)前線程，等到發(fā)送完成才釋放 21 sendLock.WaitOne(); 22 } 23 24 void SendArg_Completed(object sender, SocketAsyncEventArgs e) 25 { 26 Socket client = sender as Socket; 27 ConnectionUnit unit = e.UserToken as ConnectionUnit; 28 //這里的消息碼有三個,，2字頭的是成功的,，1字頭是不成功的 29 //101是未知錯誤，100是客戶端不在線 30 if (e.SocketError == SocketError.Success) 31 if (OnSend != null) OnSend(unit.Uid, "200"); 32 else if (OnSend != null) OnSend(unit.Uid, "101"); 33 //釋放信號,，以便下次發(fā)送消息執(zhí)行 34 sendLock.Set(); 35 }
下面則是類里面的一些字段信息和構(gòu)造函數(shù) 1 /// <summary> 2 /// 初始化池的互斥體 3 /// </summary> 4 private Mutex mutex = new Mutex(); 5 6 /// <summary> 7 /// Accept限制信號 8 /// </summary> 9 private Semaphore semaphoreAccept; 10 11 /// <summary> 12 /// Accept信號 13 /// </summary> 14 private static ManualResetEvent acceptLock = new ManualResetEvent(false); 15 16 /// <summary> 17 /// Send信號 18 /// </summary> 19 private static ManualResetEvent sendLock = new ManualResetEvent(false); 20 21 /// <summary> 22 /// 最大并發(fā)數(shù)(連接數(shù)) 23 /// </summary> 24 private int maxConnect; 25 26 /// <summary> 27 /// 當(dāng)前連接數(shù)(并發(fā)數(shù)) 28 /// </summary> 29 private int currentConnect; 30 31 /// <summary> 32 /// 緩沖池 33 /// </summary> 34 private BufferManager buffer; 35 36 /// <summary> 37 /// SocketasyncEventArgs池 38 /// </summary> 39 private SocketAsyncEventArgsPool pool; 40 41 /// <summary> 42 /// 服務(wù)端Socket 43 /// </summary> 44 private Socket server; 45 46 /// <summary> 47 /// 完成接受的委托 48 /// </summary> 49 public delegate void AcceptHandler(string uid); 50 51 /// <summary> 52 /// 完成發(fā)送的委托 53 /// </summary> 54 public delegate void SendHandler(string uid, string result); 55 56 /// <summary> 57 /// 完成接收的委托 58 /// </summary> 59 public delegate void RecevieHandler(string uid, string data); 60 61 /// <summary> 62 /// 完成接受事件 63 /// </summary> 64 public event AcceptHandler OnAccept; 65 66 /// <summary> 67 /// 完成發(fā)送事件 68 /// </summary> 69 public event SendHandler OnSend; 70 71 /// <summary> 72 /// 完成接收事件 73 /// </summary> 74 public event RecevieHandler OnReceive; 75 76 /// <summary> 77 /// 構(gòu)造函數(shù) 78 /// </summary> 79 /// <param name="buffersize">單元緩沖區(qū)大小</param> 80 /// <param name="maxCount">并發(fā)總數(shù)</param> 81 public SocketPoolController(int buffersize, int maxCount) 82 { 83 buffer = new BufferManager(buffersize * maxCount,buffersize); 84 this.currentConnect = 0; 85 this.maxConnect = maxCount; 86 this.currentConnect = 0; 87 this.pool = new SocketAsyncEventArgsPool(); 88 //設(shè)置并發(fā)數(shù)信號,，經(jīng)試驗(yàn)過是并發(fā)數(shù)-1才對 89 this.semaphoreAccept = new Semaphore(maxCount-1, maxCount-1); 90 InitPool(); 91 }
構(gòu)造函數(shù)里用到的方法 1 /// <summary> 2 /// 初始化SocketAsyncEventArgs池 3 /// 這里主要是給空閑棧填充足夠的實(shí)例 4 /// </summary> 5 private void InitPool() 6 { 7 ConnectionUnit unit = null; 8 for (int i = 0; i < maxConnect; i++) 9 { 10 unit = new ConnectionUnit(); 11 unit.Uid = "-1"; 12 unit.RecArg = new SocketAsyncEventArgs(); 13 unit.RecArg.Completed += new EventHandler<SocketAsyncEventArgs>(RecArg_Completed); 14 unit.SendArg = new SocketAsyncEventArgs(); 15 unit.SendArg.Completed += new EventHandler<SocketAsyncEventArgs>(SendArg_Completed); 16 this.pool.Push(unit); 17 } 18 }
其他外放專門控制池的方法 1 /// <summary> 2 /// 啟動池 3 /// </summary> 4 /// <param name="ipAddress">服務(wù)端的IP</param> 5 /// <param name="port">端口</param> 6 public void RunPool(string ipAddress, int port) 7 { 8 IPEndPoint endpoint = new IPEndPoint(IPAddress.Parse(ipAddress), port); 9 server = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp); 10 server.Bind(endpoint); 11 server.Listen(100); 12 13 //調(diào)用方法異步Accept客戶端的連接 14 MyAsyncAccept(); 15 //設(shè)置信號，防止再池在已經(jīng)啟動的情況下再次啟動 16 mutex.WaitOne(); 17 } 18 19 /// <summary> 20 /// 停止池 21 /// </summary> 22 public void StopPool() 23 { 24 //把服務(wù)端的socket關(guān)了 25 if (server != null) 26 server.Close(); 27 //釋放互斥信號,，等待下次啟動 28 mutex.ReleaseMutex(); 29 //釋放資源 30 Dispose(); 31 }
要把這個操作類與現(xiàn)實(shí)的事物類比的話,，這個與我們平常跟某個或一堆人聊天時差不多，別人說的東西,，不用你自己控制你都會聽得到(排除帶上耳機(jī),，塞住耳朵這種極端的情況)，所以接收消息那個方法就不需要了,，而你要說什么這個要靠個人控制,，而事件那些也好類比，OnAccept就相當(dāng)于某人加入到這次聊天中你會做些什么(say hello 或者無視),，OnRecevie就聽到別人說什么自己有什么反應(yīng),，OnSend就自己說完話之后有什么反應(yīng)(有時候發(fā)現(xiàn)說錯了會糾正，有時覺得自己說得好笑的也笑一場諸如此類),。 在使用的時候可以這樣子 1 SocketPoolController pool; 2 pool = new SocketPoolController(32 * 1024, 1000); 3 pool.OnReceive += new SocketPoolController.RecevieHandler(pool_OnReceive); 4 pool.OnSend += new SocketPoolController.SendHandler(pool_OnSend); 5 pool.OnAccept += new SocketPoolController.AcceptHandler(pool_OnAccept); 6 pool.RunPool("127.0.0.1", 8081); 7 Console.WriteLine("Pool has run\r\npress any key to stop..."); 8 Console.ReadKey(); 9 pool.StopPool(); 10 Console.WriteLine("Pool has stop\r\npress any key to exit..."); 11 Console.ReadLine();
在池里面有一部分地方看似跟用同步的差不多,，像接受客戶端的連接，發(fā)送消息這些地方,?？墒怯眠@種異步，萬一客戶端突然斷開連接也不會有同步那樣馬上拋異常,。還有一點(diǎn)的是在這份代碼里面缺少了對異常的捕捉,，有一部分錯誤我在測試的過程中設(shè)了判斷避開了。以前跟某只貓?zhí)徇^Socket編程會與多線程一起使用,，我也覺得是這樣，在之前一篇博文 《Socket一對多通信》里我也用到線程,，后來的異步通信也是有線程的,，不過不是.net framework自行創(chuàng)建的?？戳颂锎蟾绲牟┪慕o我的另一個收獲是信號量的使用,，在以前不懂得使用信號量，只會設(shè)置一大堆標(biāo)識狀態(tài)的布爾值或整形的變量來計數(shù)判斷,。田大哥的博文介紹的時高性能的Socket,，而我的這個應(yīng)該性能不會高到哪里去。上面有什么說錯的請各位指出,，有什么說漏的,，請各位提點(diǎn),，多多指導(dǎo)。謝謝,！ 整份源碼
|
|
|
