最近有項(xiàng)目要做一個(gè)高性能網(wǎng)絡(luò)服務(wù)器,決定下功夫搞定完成端口(IOCP),最終花了一個(gè)星期終于把它弄清楚了,并用C++寫了一個(gè)版本,效率很不錯(cuò)。
但,從項(xiàng)目的總體需求來考慮,最終決定上.net平臺(tái),因此又花了一天一夜弄出了一個(gè)C#版,在這與大家分享。
一些心得體會(huì):
1、在C#中,不用去面對(duì)完成端口的操作系統(tǒng)內(nèi)核對(duì)象,Microsoft已經(jīng)為我們提供了SocketAsyncEventArgs類,它封裝了IOCP的使用。請(qǐng)參考:http://msdn.microsoft.com/zh-cn/library/system.net.sockets.socketasynceventargs.aspx?cs-save-lang=1&cs-lang=cpp#code-snippet-1。
2、我的SocketAsyncEventArgsPool類使用List對(duì)象來存儲(chǔ)對(duì)客戶端來通信的SocketAsyncEventArgs對(duì)象,它相當(dāng)于直接使用內(nèi)核對(duì)象時(shí)的IoContext。我這樣設(shè)計(jì)比用堆棧來實(shí)現(xiàn)的好處理是,我可以在SocketAsyncEventArgsPool池中找到任何一個(gè)與服務(wù)器連接的客戶,主動(dòng)向它發(fā)信息。而用堆棧來實(shí)現(xiàn)的話,要主動(dòng)給客戶發(fā)信息,則還要設(shè)計(jì)一個(gè)結(jié)構(gòu)來存儲(chǔ)已連接上服務(wù)器的客戶。
3、對(duì)每一個(gè)客戶端不管還發(fā)送還是接收,我使用同一個(gè)SocketAsyncEventArgs對(duì)象,對(duì)每一個(gè)客戶端來說,通信是同步進(jìn)行的,也就是說服務(wù)器高度保證同一個(gè)客戶連接上要么在投遞發(fā)送請(qǐng)求,并等待;或者是在投遞接收請(qǐng)求,等待中。本例只做echo服務(wù)器,還未考慮由服務(wù)器主動(dòng)向客戶發(fā)送信息。
4、SocketAsyncEventArgs的UserToken被直接設(shè)定為被接受的客戶端Socket。
5、沒有使用BufferManager 類,因?yàn)槲以诔跏蓟瘯r(shí)給每一個(gè)SocketAsyncEventArgsPool中的對(duì)象分配一個(gè)緩沖區(qū),發(fā)送時(shí)使用Arrary.Copy來進(jìn)行字符拷貝,不去改變緩沖區(qū)的位置,只改變使用的長(zhǎng)度,因此在下次投遞接收請(qǐng)求時(shí)恢復(fù)緩沖區(qū)長(zhǎng)度就可以了!如果要主動(dòng)給客戶發(fā)信息的話,可以new一個(gè)SocketAsyncEventArgs對(duì)象,或者在初始化中建立幾個(gè)來專門用于主動(dòng)發(fā)送信息,因?yàn)檫@種需求一般是進(jìn)行信息群發(fā),建立一個(gè)對(duì)象可以用于很多次信息發(fā)送,總體來看,這種花銷不大,還減去了字符拷貝和消耗。
6、測(cè)試結(jié)果:(在我的筆記本上時(shí)行的,我的本本是T420 I7 8G內(nèi)存)
100客戶 100,000(十萬次)不間斷的發(fā)送接收數(shù)據(jù)(發(fā)送和接收之間沒有Sleep,就一個(gè)一循環(huán),不斷的發(fā)送與接收)耗時(shí)3004.6325 秒完成總共 10,000,000 一千萬次訪問平均每分完成 199,691.6 次發(fā)送與接收平均每秒完成 3,328.2 次發(fā)送與接收
整個(gè)運(yùn)行過程中,內(nèi)存消耗在開始兩三分種后就保持穩(wěn)定不再增漲。
看了一下對(duì)每個(gè)客戶端的延遲最多不超過2毫秒,CPU占用在8%左右。
7、下載地址:http://download.csdn.net/detail/ztk12/4928644
8、源碼:
IoContextPool.csusing System;using System.Collections.Generic;using System.Text;using System.Net.Sockets;namespace IocpServer{ /// <summary> /// 與每個(gè)客戶Socket相關(guān)聯(lián),進(jìn)行Send和Receive投遞時(shí)所需要的參數(shù) /// </summary> internal sealed class IoContextPool { List<SocketAsyncEventArgs> pool; //為每一個(gè)Socket客戶端分配一個(gè)SocketAsyncEventArgs,用一個(gè)List管理,在程序啟動(dòng)時(shí)建立。 Int32 capacity; //pool對(duì)象池的容量 Int32 boundary; //已分配和未分配對(duì)象的邊界,大的是已經(jīng)分配的,小的是未分配的 internal IoContextPool(Int32 capacity) { this.pool = new List<SocketAsyncEventArgs>(capacity); this.boundary = 0; this.capacity = capacity; } /// <summary> /// 往pool對(duì)象池中增加新建立的對(duì)象,因?yàn)檫@個(gè)程序在啟動(dòng)時(shí)會(huì)建立好所有對(duì)象, /// 故這個(gè)方法只在初始化時(shí)會(huì)被調(diào)用,因此,沒有加鎖。 /// </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> /// 取出集合中指定對(duì)象,內(nèi)部使用 /// </summary> /// <param name="index"></param> /// <returns></returns> //internal SocketAsyncEventArgs Get(int index) //{ // if (index >= 0 && index < capacity) // return pool[index]; // else // return null; //} /// <summary> /// 從對(duì)象池中取出一個(gè)對(duì)象,交給一個(gè)socket來進(jìn)行投遞請(qǐng)求操作 /// </summary> /// <returns></returns> internal SocketAsyncEventArgs Pop() { lock (this.pool) { if (boundary > 0) { --boundary; return pool[boundary]; } else return null; } } /// <summary> /// 一個(gè)socket客戶斷開,與其相關(guān)的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.csusing System;using System.Collections.Generic;using System.Text;using System.Net.Sockets;using System.Threading;using System.Net;namespace IocpServer{ /// <summary> /// 基于SocketAsyncEventArgs 實(shí)現(xiàn) IOCP 服務(wù)器 /// </summary> internal sealed class IoServer { /// <summary> /// 監(jiān)聽Socket,用于接受客戶端的連接請(qǐng)求 /// </summary> PRivate Socket listenSocket; /// <summary> /// 用于服務(wù)器執(zhí)行的互斥同步對(duì)象 /// </summary> private static Mutex mutex = new Mutex(); /// <summary> /// 用于每個(gè)I/O Socket操作的緩沖區(qū)大小 /// </summary> private Int32 bufferSize; /// <summary> /// 服務(wù)器上連接的客戶端總數(shù) /// </summary> private Int32 numConnectedSockets; /// <summary> /// 服務(wù)器能接受的最大連接數(shù)量 /// </summary> private Int32 numConnections; /// <summary> /// 完成端口上進(jìn)行投遞所用的IoContext對(duì)象池 /// </summary> private IoContextPool ioContextPool; public MainForm mainForm; /// <summary> /// 構(gòu)造函數(shù),建立一個(gè)未初始化的服務(wù)器實(shí)例 /// </summary> /// <param name="numConnections">服務(wù)器的最大連接數(shù)據(jù)</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預(yù)分配SocketAsyncEventArgs對(duì)象 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); // 將預(yù)分配的對(duì)象加入SocketAsyncEventArgs對(duì)象池中 this.ioContextPool.Add(ioContext); } } /// <summary> /// 當(dāng)Socket上的發(fā)送或接收請(qǐng)求被完成時(shí),調(diào)用此函數(shù) /// </summary> /// <param name="sender">激發(fā)事件的對(duì)象</param> /// <param name="e">與發(fā)送或接收完成操作相關(guān)聯(lián)的SocketAsyncEventArg對(duì)象</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> ///接收完成時(shí)處理函數(shù) /// </summary> /// <param name="e">與接收完成操作相關(guān)聯(lián)的SocketAsyncEventArg對(duì)象</param> private void ProcessReceive(SocketAsyncEventArgs e) { // 檢查遠(yuǎn)程主機(jī)是否關(guān)閉連接 if (e.BytesTransferred > 0) { if (e.SocketError == SocketError.Success) { Socket s = (Socket)e.UserToken; //判斷所有需接收的數(shù)據(jù)是否已經(jīng)完成 if (s.Available == 0) { // 設(shè)置發(fā)送數(shù)據(jù) Array.Copy(e.Buffer, 0, e.Buffer, e.BytesTransferred, e.BytesTransferred); e.SetBuffer(e.Offset, e.BytesTransferred * 2); if (!s.SendAsync(e)) //投遞發(fā)送請(qǐng)求,這個(gè)函數(shù)有可能同步發(fā)送出去,這時(shí)返回false,并且不會(huì)引發(fā)SocketAsyncEventArgs.Completed事件 { // 同步發(fā)送時(shí)處理發(fā)送完成事件 this.ProcessSend(e); } } else if (!s.ReceiveAsync(e)) //為接收下一段數(shù)據(jù),投遞接收請(qǐng)求,這個(gè)函數(shù)有可能同步完成,這時(shí)返回false,并且不會(huì)引發(fā)SocketAsyncEventArgs.Completed事件 { // 同步接收時(shí)處理接收完成事件 this.ProcessReceive(e); } } else { this.ProcessError(e); } } else { this.CloseClientSocket(e); } } /// <summary> /// 發(fā)送完成時(shí)處理函數(shù) /// </summary> /// <param name="e">與發(fā)送完成操作相關(guān)聯(lián)的SocketAsyncEventArg對(duì)象</param> private void ProcessSend(SocketAsyncEventArgs e) { if (e.SocketError == SocketError.Success) { Socket s = (Socket)e.UserToken; //接收時(shí)根據(jù)接收的字節(jié)數(shù)收縮了緩沖區(qū)的大小,因此投遞接收請(qǐng)求時(shí),恢復(fù)緩沖區(qū)大小 e.SetBuffer(0, bufferSize); if (!s.ReceiveAsync(e)) //投遞接收請(qǐng)求 { // 同步接收時(shí)處理接收完成事件 this.ProcessReceive(e); } } else { this.ProcessError(e); } } /// <summary> /// 處理socket錯(cuò)誤 /// </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("套接字錯(cuò)誤 {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> /// 關(guān)閉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 對(duì)象被釋放,壓入可重用隊(duì)列。 this.ioContextPool.Push(e); string outStr = String.Format("客戶 {0} 斷開, 共有 {1} 個(gè)連接。", 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 操作完成時(shí)回調(diào)函數(shù) /// </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> /// 監(jiān)聽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) { // 從接受的客戶端連接中取數(shù)據(jù)配置ioContext ioContext.UserToken = s; Interlocked.Increment(ref this.numConnectedSockets); string outStr = String.Format("客戶 {0} 連入, 共有 {1} 個(gè)連接。", 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 //已經(jīng)達(dá)到最大客戶連接數(shù)量,在這接受連接,發(fā)送“連接已經(jīng)達(dá)到最大數(shù)”,然后斷開連接 { s.Send(Encoding.Default.GetBytes("連接已經(jīng)達(dá)到最大數(shù)!")); 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} 數(shù)據(jù)出錯(cuò), 異常信息: {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()); } // 投遞下一個(gè)接受請(qǐng)求 this.StartAccept(e); } } /// <summary> /// 從客戶端開始接受一個(gè)連接操作 /// </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 { // 重用前進(jìn)行對(duì)象清理 acceptEventArg.AcceptSocket = null; } if (!this.listenSocket.AcceptAsync(acceptEventArg)) { this.ProcessAccept(acceptEventArg); } } /// <summary> /// 啟動(dòng)服務(wù),開始監(jiān)聽 /// </summary> /// <param name="port">Port where the server will listen for connection requests.</param> internal void Start(Int32 port) { // 獲得主機(jī)相關(guān)信息 IPAddress[] addressList = Dns.GetHostEntry(Environment.MachineName).AddressList; IPEndPoint localEndPoint = new IPEndPoint(addressList[addressList.Length - 1], port); // 創(chuàng)建監(jiān)聽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) { // 配置監(jiān)聽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); } // 開始監(jiān)聽 this.listenSocket.Listen(this.numConnections); // 在監(jiān)聽Socket上投遞一個(gè)接受請(qǐng)求。 this.StartAccept(null); // Blocks the current thread to receive incoming messages. mutex.WaitOne(); } /// <summary> /// 停止服務(wù) /// </summary> internal void Stop() { this.listenSocket.Close(); mutex.ReleaseMutex(); } }}
新聞熱點(diǎn)
疑難解答
圖片精選