碼上歡樂
相關內容    簡體    繁體

【轉】C#高性能大容量SOCKET並發（二）：SocketAsyncEventArgs封裝

本文轉載自   查看原文   2016-12-07 00:00   2657 

http://blog.csdn.net/sqldebug_fan/article/details/17557341

1、SocketAsyncEventArgs介紹

SocketAsyncEventArgs是微軟提供的高性能異步Socket實現類，主要為高性能網絡服務器應用程序而設計，主要是為了避免在在異步套接字 I/O 量非常大時發生重復的對象分配和同步。使用此類執行異步套接字操作的模式包含以下步驟：
1.分配一個新的 SocketAsyncEventArgs 上下文對象，或者從應用程序池中獲取一個空閑的此類對象。
2.將該上下文對象的屬性設置為要執行的操作（例如，完成回調方法、數據緩沖區、緩沖區偏移量以及要傳輸的最大數據量）。
3.調用適當的套接字方法 (xxxAsync) 以啟動異步操作。
4.如果異步套接字方法 (xxxAsync) 返回 true，則在回調中查詢上下文屬性來獲取完成狀態。
5.如果異步套接字方法 (xxxAsync) 返回 false，則說明操作是同步完成的。 可以查詢上下文屬性來獲取操作結果。
6.將該上下文重用於另一個操作，將它放回到應用程序池中，或者將它丟棄。

2、SocketAsyncEventArgs封裝

 

使用SocketAsyncEventArgs之前需要先建立一個Socket監聽對象，使用如下代碼：

 

[csharp]  view plain  copy
 
 在CODE上查看代碼片派生到我的代碼片
  1. public void Start(IPEndPoint localEndPoint)  
  2. {  
  3.     listenSocket = new Socket(localEndPoint.AddressFamily, SocketType.Stream, ProtocolType.Tcp);  
  4.     listenSocket.Bind(localEndPoint);  
  5.     listenSocket.Listen(m_numConnections);  
  6.     Program.Logger.InfoFormat("Start listen socket {0} success", localEndPoint.ToString());  
  7.     //for (int i = 0; i < 64; i++) //不能循環投遞多次AcceptAsync，會造成只接收8000連接后不接收連接了  
  8.     StartAccept(null);  
  9.     m_daemonThread = new DaemonThread(this);  
  10. }  
然后開始接受連接，SocketAsyncEventArgs有連接時會通過Completed事件通知外面，所以接受連接的代碼如下：

 

 

[csharp]  view plain  copy
 
 在CODE上查看代碼片派生到我的代碼片
  1. public void StartAccept(SocketAsyncEventArgs acceptEventArgs)  
  2. {  
  3.     if (acceptEventArgs == null)  
  4.     {  
  5.         acceptEventArgs = new SocketAsyncEventArgs();  
  6.         acceptEventArgs.Completed += new EventHandler<SocketAsyncEventArgs>(AcceptEventArg_Completed);  
  7.     }  
  8.     else  
  9.     {  
  10.         acceptEventArgs.AcceptSocket = null; //釋放上次綁定的Socket，等待下一個Socket連接  
  11.     }  
  12.   
  13.     m_maxNumberAcceptedClients.WaitOne(); //獲取信號量  
  14.     bool willRaiseEvent = listenSocket.AcceptAsync(acceptEventArgs);  
  15.     if (!willRaiseEvent)  
  16.     {  
  17.         ProcessAccept(acceptEventArgs);  
  18.     }  
  19. }  
接受連接響應事件代碼：

 

 

[csharp]  view plain  copy
 
 在CODE上查看代碼片派生到我的代碼片
  1. void AcceptEventArg_Completed(object sender, SocketAsyncEventArgs acceptEventArgs)  
  2. {  
  3.     try  
  4.     {  
  5.         ProcessAccept(acceptEventArgs);  
  6.     }  
  7.     catch (Exception E)  
  8.     {  
  9.         Program.Logger.ErrorFormat("Accept client {0} error, message: {1}", acceptEventArgs.AcceptSocket, E.Message);  
  10.         Program.Logger.Error(E.StackTrace);    
  11.     }              
  12. }  
[csharp]  view plain  copy
 
 在CODE上查看代碼片派生到我的代碼片
  1. private void ProcessAccept(SocketAsyncEventArgs acceptEventArgs)  
  2. {  
  3.     Program.Logger.InfoFormat("Client connection accepted. Local Address: {0}, Remote Address: {1}",  
  4.         acceptEventArgs.AcceptSocket.LocalEndPoint, acceptEventArgs.AcceptSocket.RemoteEndPoint);  
  5.   
  6.     AsyncSocketUserToken userToken = m_asyncSocketUserTokenPool.Pop();  
  7.     m_asyncSocketUserTokenList.Add(userToken); //添加到正在連接列表  
  8.     userToken.ConnectSocket = acceptEventArgs.AcceptSocket;  
  9.     userToken.ConnectDateTime = DateTime.Now;  
  10.   
  11.     try  
  12.     {  
  13.         bool willRaiseEvent = userToken.ConnectSocket.ReceiveAsync(userToken.ReceiveEventArgs); //投遞接收請求  
  14.         if (!willRaiseEvent)  
  15.         {  
  16.             lock (userToken)  
  17.             {  
  18.                 ProcessReceive(userToken.ReceiveEventArgs);  
  19.             }  
  20.         }                      
  21.     }  
  22.     catch (Exception E)  
  23.     {  
  24.         Program.Logger.ErrorFormat("Accept client {0} error, message: {1}", userToken.ConnectSocket, E.Message);  
  25.         Program.Logger.Error(E.StackTrace);                  
  26.     }              
  27.   
  28.     StartAccept(acceptEventArgs); //把當前異步事件釋放，等待下次連接  
  29. }  
接受連接后，從當前Socket緩沖池AsyncSocketUserTokenPool中獲取一個用戶對象AsyncSocketUserToken，AsyncSocketUserToken包含一個接收異步事件m_receiveEventArgs，一個發送異步事件m_sendEventArgs，接收數據緩沖區m_receiveBuffer，發送數據緩沖區m_sendBuffer，協議邏輯調用對象m_asyncSocketInvokeElement，建立服務對象后，需要實現接收和發送的事件響應函數：

 

 

[csharp]  view plain  copy
 
 在CODE上查看代碼片派生到我的代碼片
  1. void IO_Completed(object sender, SocketAsyncEventArgs asyncEventArgs)  
  2. {  
  3.     AsyncSocketUserToken userToken = asyncEventArgs.UserToken as AsyncSocketUserToken;  
  4.     userToken.ActiveDateTime = DateTime.Now;  
  5.     try  
  6.     {                  
  7.         lock (userToken)  
  8.         {  
  9.             if (asyncEventArgs.LastOperation == SocketAsyncOperation.Receive)  
  10.                 ProcessReceive(asyncEventArgs);  
  11.             else if (asyncEventArgs.LastOperation == SocketAsyncOperation.Send)  
  12.                 ProcessSend(asyncEventArgs);  
  13.             else  
  14.                 throw new ArgumentException("The last operation completed on the socket was not a receive or send");  
  15.         }     
  16.     }  
  17.     catch (Exception E)  
  18.     {  
  19.         Program.Logger.ErrorFormat("IO_Completed {0} error, message: {1}", userToken.ConnectSocket, E.Message);  
  20.         Program.Logger.Error(E.StackTrace);  
  21.     }                       
  22. }  
在Completed事件中需要處理發送和接收的具體邏輯代碼，其中接收的邏輯實現如下：

 

 

[csharp]  view plain  copy
 
 在CODE上查看代碼片派生到我的代碼片
  1. private void ProcessReceive(SocketAsyncEventArgs receiveEventArgs)  
  2. {  
  3.     AsyncSocketUserToken userToken = receiveEventArgs.UserToken as AsyncSocketUserToken;  
  4.     if (userToken.ConnectSocket == null)  
  5.         return;  
  6.     userToken.ActiveDateTime = DateTime.Now;  
  7.     if (userToken.ReceiveEventArgs.BytesTransferred > 0 && userToken.ReceiveEventArgs.SocketError == SocketError.Success)  
  8.     {  
  9.         int offset = userToken.ReceiveEventArgs.Offset;  
  10.         int count = userToken.ReceiveEventArgs.BytesTransferred;  
  11.         if ((userToken.AsyncSocketInvokeElement == null) & (userToken.ConnectSocket != null)) //存在Socket對象，並且沒有綁定協議對象，則進行協議對象綁定  
  12.         {  
  13.             BuildingSocketInvokeElement(userToken);  
  14.             offset = offset + 1;  
  15.             count = count - 1;  
  16.         }  
  17.         if (userToken.AsyncSocketInvokeElement == null) //如果沒有解析對象，提示非法連接並關閉連接  
  18.         {  
  19.             Program.Logger.WarnFormat("Illegal client connection. Local Address: {0}, Remote Address: {1}", userToken.ConnectSocket.LocalEndPoint,   
  20.                 userToken.ConnectSocket.RemoteEndPoint);  
  21.             CloseClientSocket(userToken);  
  22.         }  
  23.         else  
  24.         {  
  25.             if (count > 0) //處理接收數據  
  26.             {  
  27.                 if (!userToken.AsyncSocketInvokeElement.ProcessReceive(userToken.ReceiveEventArgs.Buffer, offset, count))  
  28.                 { //如果處理數據返回失敗，則斷開連接  
  29.                     CloseClientSocket(userToken);  
  30.                 }  
  31.                 else //否則投遞下次介紹數據請求  
  32.                 {  
  33.                     bool willRaiseEvent = userToken.ConnectSocket.ReceiveAsync(userToken.ReceiveEventArgs); //投遞接收請求  
  34.                     if (!willRaiseEvent)  
  35.                         ProcessReceive(userToken.ReceiveEventArgs);  
  36.                 }  
  37.             }  
  38.             else  
  39.             {  
  40.                 bool willRaiseEvent = userToken.ConnectSocket.ReceiveAsync(userToken.ReceiveEventArgs); //投遞接收請求  
  41.                 if (!willRaiseEvent)  
  42.                     ProcessReceive(userToken.ReceiveEventArgs);  
  43.             }  
  44.         }  
  45.     }  
  46.     else  
  47.     {  
  48.         CloseClientSocket(userToken);  
  49.     }  
  50. }  
由於我們制定的協議第一個字節是協議標識，因此在接收到第一個字節的時候需要綁定協議解析對象，具體代碼實現如下：

 

 

[csharp]  view plain  copy
 
 在CODE上查看代碼片派生到我的代碼片
  1. private void BuildingSocketInvokeElement(AsyncSocketUserToken userToken)  
  2. {  
  3.     byte flag = userToken.ReceiveEventArgs.Buffer[userToken.ReceiveEventArgs.Offset];  
  4.     if (flag == (byte)SocketFlag.Upload)  
  5.         userToken.AsyncSocketInvokeElement = new UploadSocketProtocol(this, userToken);  
  6.     else if (flag == (byte)SocketFlag.Download)  
  7.         userToken.AsyncSocketInvokeElement = new DownloadSocketProtocol(this, userToken);  
  8.     else if (flag == (byte)SocketFlag.RemoteStream)  
  9.         userToken.AsyncSocketInvokeElement = new RemoteStreamSocketProtocol(this, userToken);  
  10.     else if (flag == (byte)SocketFlag.Throughput)  
  11.         userToken.AsyncSocketInvokeElement = new ThroughputSocketProtocol(this, userToken);  
  12.     else if (flag == (byte)SocketFlag.Control)  
  13.         userToken.AsyncSocketInvokeElement = new ControlSocketProtocol(this, userToken);  
  14.     else if (flag == (byte)SocketFlag.LogOutput)  
  15.         userToken.AsyncSocketInvokeElement = new LogOutputSocketProtocol(this, userToken);  
  16.     if (userToken.AsyncSocketInvokeElement != null)  
  17.     {  
  18.         Program.Logger.InfoFormat("Building socket invoke element {0}.Local Address: {1}, Remote Address: {2}",  
  19.             userToken.AsyncSocketInvokeElement, userToken.ConnectSocket.LocalEndPoint, userToken.ConnectSocket.RemoteEndPoint);  
  20.     }   
  21. }  
發送響應函數實現需要注意，我們是把發送數據放到一個列表中，當上一個發送事件完成響應Completed事件，這時我們需要檢測發送隊列中是否存在未發送的數據，如果存在則繼續發送。

 

 

[csharp]  view plain  copy
 
 在CODE上查看代碼片派生到我的代碼片
  1. private bool ProcessSend(SocketAsyncEventArgs sendEventArgs)  
  2. {  
  3.     AsyncSocketUserToken userToken = sendEventArgs.UserToken as AsyncSocketUserToken;  
  4.     if (userToken.AsyncSocketInvokeElement == null)  
  5.         return false;  
  6.     userToken.ActiveDateTime = DateTime.Now;  
  7.     if (sendEventArgs.SocketError == SocketError.Success)  
  8.         return userToken.AsyncSocketInvokeElement.SendCompleted(); //調用子類回調函數  
  9.     else  
  10.     {  
  11.         CloseClientSocket(userToken);  
  12.         return false;  
  13.     }  
  14. }  
SendCompleted用於回調下次需要發送的數據，具體實現過程如下：

 

 

[csharp]  view plain  copy
 
 在CODE上查看代碼片派生到我的代碼片
  1. public virtual bool SendCompleted()  
  2. {  
  3.     m_activeDT = DateTime.UtcNow;  
  4.     m_sendAsync = false;  
  5.     AsyncSendBufferManager asyncSendBufferManager = m_asyncSocketUserToken.SendBuffer;  
  6.     asyncSendBufferManager.ClearFirstPacket(); //清除已發送的包  
  7.     int offset = 0;  
  8.     int count = 0;  
  9.     if (asyncSendBufferManager.GetFirstPacket(ref offset, ref count))  
  10.     {  
  11.         m_sendAsync = true;  
  12.         return m_asyncSocketServer.SendAsyncEvent(m_asyncSocketUserToken.ConnectSocket, m_asyncSocketUserToken.SendEventArgs,  
  13.             asyncSendBufferManager.DynamicBufferManager.Buffer, offset, count);  
  14.     }  
  15.     else  
  16.         return SendCallback();  
  17. }  
  18.   
  19. //發送回調函數，用於連續下發數據  
  20. public virtual bool SendCallback()  
  21. {  
  22.     return true;  
  23. }  
當一個SocketAsyncEventArgs斷開后，我們需要斷開對應的Socket連接，並釋放對應資源，具體實現函數如下：

 

 

[csharp]  view plain  copy
 
 在CODE上查看代碼片派生到我的代碼片
  1. public void CloseClientSocket(AsyncSocketUserToken userToken)  
  2. {  
  3.     if (userToken.ConnectSocket == null)  
  4.         return;  
  5.     string socketInfo = string.Format("Local Address: {0} Remote Address: {1}", userToken.ConnectSocket.LocalEndPoint,  
  6.         userToken.ConnectSocket.RemoteEndPoint);  
  7.     Program.Logger.InfoFormat("Client connection disconnected. {0}", socketInfo);  
  8.     try  
  9.     {  
  10.         userToken.ConnectSocket.Shutdown(SocketShutdown.Both);  
  11.     }  
  12.     catch (Exception E)   
  13.     {  
  14.         Program.Logger.ErrorFormat("CloseClientSocket Disconnect client {0} error, message: {1}", socketInfo, E.Message);  
  15.     }  
  16.     userToken.ConnectSocket.Close();  
  17.     userToken.ConnectSocket = null; //釋放引用，並清理緩存，包括釋放協議對象等資源  
  18.   
  19.     m_maxNumberAcceptedClients.Release();  
  20.     m_asyncSocketUserTokenPool.Push(userToken);  
  21.     m_asyncSocketUserTokenList.Remove(userToken);  
  22. }  

 

3、SocketAsyncEventArgs封裝和MSDN的不同點

MSDN在http://msdn.microsoft.com/zh-cn/library/system.NET.sockets.socketasynceventargs(v=vs.110).aspx實現了示例代碼，並實現了初步的池化處理，我們是在它的基礎上擴展實現了接收數據緩沖，發送數據隊列，並把發送SocketAsyncEventArgs和接收SocketAsyncEventArgs分開，並實現了協議解析單元，這樣做的好處是方便后續邏輯實現文件的上傳，下載和日志輸出。

 

DEMO下載地址：http://download.csdn.Net/detail/sqldebug_fan/7467745
免責聲明：此代碼只是為了演示C#完成端口編程，僅用於學習和研究，切勿用於商業用途。水平有限，C#也屬於初學，錯誤在所難免，歡迎指正和指導。郵箱地址：fansheng_hx@163.com。


免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。



猜您在找 C#高性能大容量SOCKET並發(轉) C#高性能大容量SOCKET並發（一）：IOCP完成端口例子介紹（轉） 高性能TcpServer(C#) - 2.創建高性能Socket服務器SocketAsyncEventArgs的實現(IOCP) C#高性能大容量SOCKET並發（一）：IOCP完成端口例子介紹 C#高性能Socket服務器SocketAsyncEventArgs的實現(IOCP) 高並發的socket的高性能設計【轉】 一個高性能異步socket封裝庫的實現思路 （c#） C#實現高性能高並發Socket服務器 C#高性能Socket服務器IOCP實現 C#高性能Socket服務器IOCP實現
 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM