因為學(xué)習(xí)的需要，要求一個高性能的Socket服務(wù)器來提供多而繁雜的客戶端連接請求，參考了許多資料和各位的思想，自己琢磨出了一套方案，覺的可行，于是拿出來曬曬，希望大家一起學(xué)習(xí)改進。（這個方案的1.0版本已經(jīng)貼出來了，但是由于本人覺的1.0不太完美，做了下改進，本篇講的主要是2.0）
1.0的文章參考：http://www.cnblogs.com/niuchenglei/archive/2009/07/23/1529462.html
1.0和2.0性能上基本沒有變化，只是針對某些地方做了改進性的修改，本篇主要介紹原理，并貼出部分代碼，上一篇是一個Overview。
設(shè)計原則：使用.net的SocketAsyncEventArgs（原因是這個比較簡單，而且性能也很好，當(dāng)然要是c++的話就用IOCP了）?？紤]到能快速的反應(yīng)用戶的連接請求我采用了連接池的技術(shù)，類似于sqlserver的連接池，當(dāng)然我的“池”還不夠好，為了能快速的處理接受的數(shù)據(jù)我又加入了一個緩沖區(qū)池，說白了就是給每一個連接對象事先開辟好了空間。在傳輸方面，為了保證數(shù)據(jù)的有效性我們采用客戶端和服務(wù)器端的驗證（當(dāng)然也不是太復(fù)雜）。
具體分析：分析的順序是自底向上的
1.MySocketAsyncEventArgs類：這個類是一個繼承自System.Net.Socket.SocketAsyncEventArgs類，是由于特定情況需要而添加了一些外加屬性的類。
1internalsealedclassMySocketAsyncEventArgs : SocketAsyncEventArgs
2{
3internalstringUID;
4privatestringProperty;
5internalMySocketAsyncEventArgs(stringproperty){
6this.Property = property;
7}
8}
UID:用戶標(biāo)識符，用來標(biāo)識這個連接是那個用戶的。
Property:標(biāo)識該連接是用來發(fā)送信息還是監(jiān)聽接收信息的。param:Receive/Send，MySocketAsyncEventArgs類只帶有一個參數(shù)的構(gòu)造函數(shù)，說明類在實例化時就被說明是用來完成接收還是發(fā)送任務(wù)的。
2.SocketAsyncEventArgsWithId類：該類是一個用戶的連接的最小單元，也就是說對一個用戶來說有兩個SocketAsyncEventArgs對象，這兩個對象是一樣的，但是有一個用來發(fā)送消息，一個接收消息，這樣做的目的是為了實現(xiàn)雙工通訊，提高用戶體驗。默認(rèn)的用戶標(biāo)識是"-1”，狀態(tài)是false表示不可用
01 | internal sealed class SocketAsyncEventArgsWithId:IDisposable |
02 | { |
03 | private string uid = "-1"; |
04 | private bool state = false; |
05 | private MySocketAsyncEventArgs receivesaea; |
06 | private MySocketAsyncEventArgs sendsaea; |
07 | internal string UID |
08 | { |
09 | get { return uid; } |
10 | set |
11 | { |
12 | uid = value; |
13 | ReceiveSAEA.UID = value; |
14 | SendSAEA.UID = value; |
15 | } |
16 | } |
17 | } |
UID：用戶標(biāo)識，跟MySocketAsyncEventArgs的UID是一樣的，在對SocketAsycnEventArgsWithId的UID屬性賦值的時候也對MySocketAsyncEventArgs的UID屬性賦值。
State:表示連接的可用與否，一旦連接被實例化放入連接池后State即變?yōu)門rue
3.SocketAsyncEventArgsPool類：這個類才是真正的連接池類，這個類真正的為server提供一個可用的用戶連接，并且維持這個連接直到用戶斷開，并把不用的連接放回連接池中供下一用戶連接。
這個類是最核心的東西了，當(dāng)然它設(shè)計的好壞影響著總體性能的好壞，它的各項操作也可能成為整個服務(wù)器性能的瓶頸。Pool包含有幾個成員：
01internal sealed class SocketAsyncEventArgsPool:IDisposable
02{
03internal Stack<SocketAsyncEventArgsWithId> pool;
04internal IDictionary<string, SocketAsyncEventArgsWithId> busypool;
05private string[] keys;
06internal Int32 Count
07{
08get
09{
10lock (this.pool)
11{
12return this.pool.Count;
13}
14}
15}
16internal string[] OnlineUID
17{
18get
19{
20lock (this.busypool)
21{
22busypool.Keys.CopyTo(keys, 0);
23}
24return keys;
25}
26}
27internal SocketAsyncEventArgsPool(Int32 capacity)
28{
29keys = new string[capacity];
30this.pool = new Stack<SocketAsyncEventArgsWithId>(capacity);
31this.busypool = new Dictionary<string, SocketAsyncEventArgsWithId>(capacity);
32}
33internal SocketAsyncEventArgsWithId Pop(string uid)
34{
35if (uid == string.Empty || uid == "")
36return null;
37SocketAsyncEventArgsWithId si = null;
38lock (this.pool)
39{
40si = this.pool.Pop();
41}
42si.UID = uid;
43si.State = true; //mark the state of pool is not the initial step
44busypool.Add(uid, si);
45return si;
46}
47internal void Push(SocketAsyncEventArgsWithId item)
48{
49if (item == null)
50throw new ArgumentNullException("SocketAsyncEventArgsWithId對象為空");
51if (item.State == true)
52{
53if (busypool.Keys.Count != 0)
54{
55if (busypool.Keys.Contains(item.UID))
56busypool.Remove(item.UID);
57else
58throw new ArgumentException("SocketAsyncEventWithId不在忙碌隊列中");
59}
60else
61throw new ArgumentException("忙碌隊列為空");
62}
63item.UID = "-1";
64item.State = false;
65lock (this.pool)
66{
67this.pool.Push(item);
68}
69}
70internal SocketAsyncEventArgsWithId FindByUID(string uid)
71{
72if (uid == string.Empty || uid == "")
73return null;
74SocketAsyncEventArgsWithId si = null;
75foreach (string key in this.OnlineUID)
76{
77if (key == uid)
78{
79si = busypool[uid];
80break;
81}
82}
83return si;
84}
85internal bool BusyPoolContains(string uid)
86{
87lock (this.busypool)
88{
89return busypool.Keys.Contains(uid);
90}
91}
92}
Note:這個類的設(shè)計缺陷是使用了太多的lock語句，對對象做了太多的互斥操作，所以我盡量的把lock內(nèi)的語句化簡或挪到lock外部執(zhí)行。4.BufferManager類：該類是一個管理連接緩沖區(qū)的類，職責(zé)是為每一個連接維持一個接收數(shù)據(jù)的區(qū)域。它的設(shè)計也采用了類似與池的技術(shù)，先實例化好多內(nèi)存區(qū)域，并把每一塊的地址放入棧中，每執(zhí)行依次pop時拿出一塊區(qū)域來給SocketAsyncEventArgs對象作為Buffer.
01 | internal sealed class BufferManager:IDisposable |
02 | { |
03 | private Byte[] buffer; |
04 | private Int32 bufferSize; |
05 | private Int32 numSize; |
06 | private Int32 currentIndex; |
07 | private Stack<Int32> freeIndexPool; |
08 | internal Boolean SetBuffer(SocketAsyncEventArgs args) |
09 | { |
10 | if (this.freeIndexPool.Count > 0) |
11 | { |
12 | args.SetBuffer(this.buffer, this.freeIndexPool.Pop(), this.bufferSize); |
13 | } |
14 | else |
15 | { |
16 | if ((this.numSize - this.bufferSize) < this.currentIndex) |
17 | { |
18 | return false; |
19 | } |
20 | args.SetBuffer(this.buffer, this.currentIndex, this.bufferSize); |
21 | this.currentIndex += this.bufferSize; |
22 | } |
23 | return true; |
24 | } |
25 | } |
5.RequestHandler類：這里代碼就不貼了，這個類也比較簡單。比如發(fā)送方要發(fā)送的內(nèi)容為:hello,nice to meet you那么真正發(fā)送的內(nèi)容是:[length=22]hello,nice to meet you，length后的數(shù)字是字符串的長度，接收方接收到消息后根據(jù)長度檢驗和獲取信息。
強烈推薦這篇文章：http://www.cnblogs.com/JimmyZhang/archive/2008/09/16/1291854.html
6.SocketListener類：終于到了最重要的部分了，也是一個對外部真正有意義的類，這個類監(jiān)聽用戶的連接請求并從連接池取出一個可用連接給用戶，并且時刻監(jiān)聽用戶發(fā)來的數(shù)據(jù)并處理。在設(shè)計這個類時為了迎合雙工通信我把監(jiān)聽的任務(wù)放到另一個線程中去，這也是我為什么要給每個用戶兩個SocketAsyncEventArgs的原因。當(dāng)然兩個線程是不夠的還要異步。比較重要的語句我都用粗體標(biāo)注了。socket的方法都是成對出現(xiàn)的，ReceiveAsync對應(yīng)OnReceiveCompleted，SendAsync對應(yīng)OnSendCompleted，所以理解起來也不算太難，只是要注意一點：就是接收和發(fā)送消息是在兩個線程里的。
001public sealed class SocketListener:IDisposable
002{
003/// <summary>
004/// 緩沖區(qū)
005/// </summary>
006private BufferManager bufferManager;
007/// <summary>
008/// 服務(wù)器端Socket
009/// </summary>
010private Socket listenSocket;
011/// <summary>
012/// 服務(wù)同步鎖
013/// </summary>
014private static Mutex mutex = new Mutex();
015/// <summary>
016/// 當(dāng)前連接數(shù)
017/// </summary>
018private Int32 numConnections;
019/// <summary>
020/// 最大并發(fā)量
021/// </summary>
022private Int32 numConcurrence;
023/// <summary>
024/// 服務(wù)器狀態(tài)
025/// </summary>
026private ServerState serverstate;
027/// <summary>
028/// 讀取寫入字節(jié)
029/// </summary>
030private const Int32 opsToPreAlloc = 1;
031/// <summary>
032/// Socket連接池
033/// </summary>
034private SocketAsyncEventArgsPool readWritePool;
035/// <summary>
036/// 并發(fā)控制信號量
037/// </summary>
038private Semaphore semaphoreAcceptedClients;
039/// <summary>
040/// 通信協(xié)議
041/// </summary>
042private RequestHandler handler;
043/// <summary>
044/// 回調(diào)委托
045/// </summary>
046/// <param name="IP"></param>
047/// <returns></returns>
048public delegate string GetIDByIPFun(string IP);
049/// <summary>
050/// 回調(diào)方法實例
051/// </summary>
052private GetIDByIPFun GetIDByIP;
053/// <summary>
054/// 接收到信息時的事件委托
055/// </summary>
056/// <param name="info"></param>
057public delegate void ReceiveMsgHandler(string uid, string info);
058/// <summary>
059/// 接收到信息時的事件
060/// </summary>
061public event ReceiveMsgHandler OnMsgReceived;
062/// <summary>
063/// 開始監(jiān)聽數(shù)據(jù)的委托
064/// </summary>
065public delegate void StartListenHandler();
066/// <summary>
067/// 開始監(jiān)聽數(shù)據(jù)的事件
068/// </summary>
069public event StartListenHandler StartListenThread;
070/// <summary>
071/// 發(fā)送信息完成后的委托
072/// </summary>
073/// <param name="successorfalse"></param>
074public delegate void SendCompletedHandler(string uid,string exception);
075/// <summary>
076/// 發(fā)送信息完成后的事件
077/// </summary>
078public event SendCompletedHandler OnSended;
079/// <summary>
080/// 獲取當(dāng)前的并發(fā)數(shù)
081/// </summary>
082public Int32 NumConnections
083{
084get { return this.numConnections; }
085}
086/// <summary>
087/// 最大并發(fā)數(shù)
088/// </summary>
089public Int32 MaxConcurrence
090{
091get { return this.numConcurrence; }
092}
093/// <summary>
094/// 返回服務(wù)器狀態(tài)
095/// </summary>
096public ServerState State
097{
098get
099{
100return serverstate;
101}
102}
103/// <summary>
104/// 獲取當(dāng)前在線用戶的UID
105/// </summary>
106public string[] OnlineUID
107{
108get { return readWritePool.OnlineUID; }
109}
110/// <summary>
111/// 初始化服務(wù)器端
112/// </summary>
113/// <param name="numConcurrence">并發(fā)的連接數(shù)量(1000以上)</param>
114/// <param name="receiveBufferSize">每一個收發(fā)緩沖區(qū)的大小(32768)</param>
115public SocketListener(Int32 numConcurrence, Int32 receiveBufferSize, GetIDByIPFun GetIDByIP)
116{
117serverstate = ServerState.Initialing;
118this.numConnections = 0;
119this.numConcurrence = numConcurrence;
120this.bufferManager = new BufferManager(receiveBufferSize * numConcurrence * opsToPreAlloc, receiveBufferSize);
121this.readWritePool = new SocketAsyncEventArgsPool(numConcurrence);
122this.semaphoreAcceptedClients = new Semaphore(numConcurrence, numConcurrence);
123handler = new RequestHandler();
124this.GetIDByIP = GetIDByIP;
125}
126/// <summary>
127/// 服務(wù)端初始化
128/// </summary>
129public void Init()
130{
131this.bufferManager.InitBuffer();
132SocketAsyncEventArgsWithId readWriteEventArgWithId;
133for (Int32 i = 0; i < this.numConcurrence; i++)
134{
135readWriteEventArgWithId = new SocketAsyncEventArgsWithId();
136readWriteEventArgWithId.ReceiveSAEA.Completed += new EventHandler<SocketAsyncEventArgs>(OnReceiveCompleted);
137readWriteEventArgWithId.SendSAEA.Completed += new EventHandler<SocketAsyncEventArgs>(OnSendCompleted);
138//只給接收的SocketAsyncEventArgs設(shè)置緩沖區(qū)
139this.bufferManager.SetBuffer(readWriteEventArgWithId.ReceiveSAEA);
140this.readWritePool.Push(readWriteEventArgWithId);
141}
142serverstate = ServerState.Inited;
143}
144/// <summary>
145/// 啟動服務(wù)器
146/// </summary>
147/// <param name="data">端口號</param>
148public void Start(Object data)
149{
150Int32 port = (Int32)data;
151IPAddress[] addresslist = Dns.GetHostEntry(Environment.MachineName).AddressList;
152IPEndPoint localEndPoint = new IPEndPoint(addresslist[addresslist.Length - 1], port);
153this.listenSocket = new Socket(localEndPoint.AddressFamily, SocketType.Stream, ProtocolType.Tcp);
154if (localEndPoint.AddressFamily == AddressFamily.InterNetworkV6)
155{
156this.listenSocket.SetSocketOption(SocketOptionLevel.IPv6, (SocketOptionName)27, false);
157this.listenSocket.Bind(new IPEndPoint(IPAddress.IPv6Any, localEndPoint.Port));
158}
159else
160{
161this.listenSocket.Bind(localEndPoint);
162}
163this.listenSocket.Listen(100);
164this.StartAccept(null);
165//開始監(jiān)聽已連接用戶的發(fā)送數(shù)據(jù)
166StartListenThread();
167serverstate = ServerState.Running;
168mutex.WaitOne();
169}
170/// <summary>
171/// 開始監(jiān)聽線程的入口函數(shù)
172/// </summary>
173public void Listen()
174{
175while (true)
176{
177string[] keys = readWritePool.OnlineUID;
178foreach (string uid in keys)
179{
180if (uid != null && readWritePool.busypool[uid].ReceiveSAEA.LastOperation != SocketAsyncOperation.Receive)
181{
182Boolean willRaiseEvent = (readWritePool.busypool[uid].ReceiveSAEA.UserToken as Socket).ReceiveAsync(readWritePool.busypool[uid].ReceiveSAEA);
183if (!willRaiseEvent)
184ProcessReceive(readWritePool.busypool[uid].ReceiveSAEA);
185}
186}
187}
188}
189/// <summary>
190/// 發(fā)送信息
191/// </summary>
192/// <param name="uid">要發(fā)送的用戶的uid</param>
193/// <param name="msg">消息體</param>
194public void Send(string uid, string msg)
195{
196if (uid == string.Empty || uid == "" || msg == string.Empty || msg == "")
197return;
198SocketAsyncEventArgsWithId socketWithId = readWritePool.FindByUID(uid);
199if (socketWithId == null)
200//說明用戶已經(jīng)斷開
201//100 發(fā)送成功
202//200 發(fā)送失敗
203//300 用戶不在線
204//其它 表示異常的信息
205OnSended(uid, "300");
206else
207{
208MySocketAsyncEventArgs e = socketWithId.SendSAEA;
209if (e.SocketError == SocketError.Success)
210{
211int i = 0;
212try
213{
214string message = @"[lenght=" + msg.Length + @"]" + msg;
215byte[] sendbuffer = Encoding.Unicode.GetBytes(message);
216e.SetBuffer(sendbuffer, 0, sendbuffer.Length);
217Boolean willRaiseEvent = (e.UserToken as Socket).SendAsync(e);
218if (!willRaiseEvent)
219{
220this.ProcessSend(e);
221}
222}
223catch (Exception ex)
224{
225if (i <= 5)
226{
227i++;
228//如果發(fā)送出現(xiàn)異常就延遲0.01秒再發(fā)
229Thread.Sleep(10);
230Send(uid, msg);
231}
232else
233{
234OnSended(uid, ex.ToString());
235}
236}
237}
238else
239{
240OnSended(uid, "200");
241this.CloseClientSocket(((MySocketAsyncEventArgs)e).UID);
242}
243}
244}
245/// <summary>
246/// 停止服務(wù)器
247/// </summary>
248public void Stop()
249{
250if(listenSocket!=null)
251listenSocket.Close();
252listenSocket = null;
253Dispose();
254mutex.ReleaseMutex();
255serverstate = ServerState.Stoped;
256}
257private void StartAccept(SocketAsyncEventArgs acceptEventArg)
258{
259if (acceptEventArg == null)
260{
261acceptEventArg = new SocketAsyncEventArgs();
262acceptEventArg.Completed += new EventHandler<SocketAsyncEventArgs>(OnAcceptCompleted);
263}
264else
265acceptEventArg.AcceptSocket = null;
266this.semaphoreAcceptedClients.WaitOne();
267Boolean willRaiseEvent = this.listenSocket.AcceptAsync(acceptEventArg);
268if (!willRaiseEvent)
269{
270this.ProcessAccept(acceptEventArg);
271}
272}
273private void OnAcceptCompleted(object sender, SocketAsyncEventArgs e)
274{
275this.ProcessAccept(e);
276}
277private void ProcessAccept(SocketAsyncEventArgs e)
278{
279if (e.LastOperation != SocketAsyncOperation.Accept) //檢查上一次操作是否是Accept，不是就返回
280return;
281if (e.BytesTransferred <= 0) //檢查發(fā)送的長度是否大于0,不是就返回
282return;
283string UID = GetIDByIP((e.AcceptSocket.RemoteEndPoint as IPEndPoint).Address.ToString()); //根據(jù)IP獲取用戶的UID
284if (UID == string.Empty || UID == null || UID == "")
285return;
286if (readWritePool.BusyPoolContains(UID)) //判斷現(xiàn)在的用戶是否已經(jīng)連接，避免同一用戶開兩個連接
287return;
288SocketAsyncEventArgsWithId readEventArgsWithId = this.readWritePool.Pop(UID);
289readEventArgsWithId.ReceiveSAEA.UserToken = e.AcceptSocket;
290readEventArgsWithId.SendSAEA.UserToken = e.AcceptSocket;
291Interlocked.Increment(ref this.numConnections);
292this.StartAccept(e);
293}
294private void OnReceiveCompleted(object sender, SocketAsyncEventArgs e)
295{
296ProcessReceive(e);
297}
298private void OnSendCompleted(object sender, SocketAsyncEventArgs e)
299{
300ProcessSend(e);
301}
302private void ProcessReceive(SocketAsyncEventArgs e)
303{
304if (e.LastOperation != SocketAsyncOperation.Receive)
305return;
306if (e.BytesTransferred > 0)
307{
308if (e.SocketError == SocketError.Success)
309{
310Int32 byteTransferred = e.BytesTransferred;
311string received = Encoding.Unicode.GetString(e.Buffer, e.Offset, byteTransferred);
312//檢查消息的準(zhǔn)確性
313string[] msg = handler.GetActualString(received);
314foreach (string m in msg)
315OnMsgReceived(((MySocketAsyncEventArgs)e).UID, m);
316//可以在這里設(shè)一個停頓來實現(xiàn)間隔時間段監(jiān)聽，這里的停頓是單個用戶間的監(jiān)聽間隔
317//發(fā)送一個異步接受請求，并獲取請求是否為成功
318Boolean willRaiseEvent = (e.UserToken as Socket).ReceiveAsync(e);
319if (!willRaiseEvent)
320ProcessReceive(e);
321}
322}
323else
324this.CloseClientSocket(((MySocketAsyncEventArgs)e).UID);
325}
326private void ProcessSend(SocketAsyncEventArgs e)
327{
328if (e.LastOperation != SocketAsyncOperation.Send)
329return;
330if (e.BytesTransferred > 0)
331{
332if (e.SocketError == SocketError.Success)
333OnSended(((MySocketAsyncEventArgs)e).UID, "100");
334else
335OnSended(((MySocketAsyncEventArgs)e).UID, "200");
336}
337else
338this.CloseClientSocket(((MySocketAsyncEventArgs)e).UID);
339}
340private void CloseClientSocket(string uid)
341{
342if (uid == string.Empty || uid == "")
343return;
344SocketAsyncEventArgsWithId saeaw = readWritePool.FindByUID(uid);
345if (saeaw == null)
346return;
347Socket s = saeaw.ReceiveSAEA.UserToken as Socket;
348try
349{
350s.Shutdown(SocketShutdown.Both);
351}
352catch (Exception)
353{
354//客戶端已經(jīng)關(guān)閉
355}
356this.semaphoreAcceptedClients.Release();
357Interlocked.Decrement(ref this.numConnections);
358this.readWritePool.Push(saeaw);
359}
360#region IDisposable Members
361public void Dispose()
362{
363bufferManager.Dispose();
364bufferManager = null;
365readWritePool.Dispose();
366readWritePool = null;
367}
368#endregion
369}
關(guān)于所有的類已經(jīng)介紹完了，相信各位都已經(jīng)很明白了，如果不是太清楚就看源代碼，別忘了源代碼是最好的文檔！
當(dāng)然這個2.0仍然還有很多缺陷，比如職責(zé)劃分不太OO，運行不太穩(wěn)定，處理異常能力較差，處理超負(fù)載的連接能力較差，主動拒絕，可測試性差等，希望大家多給點建議，改進才對啊。
源代碼下載：http://files.cnblogs.com/niuchenglei/socketlib.rar
