引言
我
一直在探尋一個(gè)高性能的Socket客戶端代碼,。以前，我使用Socket類寫了一些基于傳統(tǒng)異步編程模型的代碼（BeginSend,、
BeginReceive,，等等）。但它沒有滿足我所要的性能需求,。終于,，我找到了基于事件的異步操作新模式（參見2007年9月MSDN雜志上的“
連接.NET框架3.5”）（部分內(nèi)容見文后的
翻譯附注——譯者注）。
背景
由
于減少了阻塞線程,，高性能I/O限制應(yīng)用中廣泛使用異步編程模型（AMP,，Asynchronous Programming Model）。.NET
Framework第一個(gè)版本就實(shí)現(xiàn)了APM,，現(xiàn)在使用諸如lambda表達(dá)式等新的技術(shù)C#3.0一直在改進(jìn)其性能,。針對(duì)Socket編程，不僅性能上
提升了不少,，而且新APM模型發(fā)布了一個(gè)更簡(jiǎn)易的編程方法,，該方法使用SocketAsyncEventArgs類來保持I/O操作之間的上下文（見文后
的
翻譯附注——譯者注），從而降低對(duì)象分配和垃圾收集工作,。 在.NET 2.0 SP1上可以使用SocketAsyncEventArgs類,，本文的代碼就是用Microsoft Visual Studio .NET 2005編寫的。
使用代碼
從SocketAsyncEventArgs類開始,，我學(xué)習(xí)了
MSDN上
的樣例程序,，但該文缺少一些內(nèi)容：AsyncUserToken類。我認(rèn)為這個(gè)類應(yīng)該公開一個(gè)Socket屬性,，它對(duì)應(yīng)執(zhí)行I/O操作的Socket,。一
段時(shí)間后，我認(rèn)識(shí)到這個(gè)類不是必要的,，因?yàn)閷傩訳serToken是一個(gè)Object,，它可以接受任何東西。下面的修改方法中直接使用一個(gè)Socket實(shí)
例當(dāng)作UserToken,。
// 處理Socket偵聽者接收,。
private void ProcessAccept(SocketAsyncEventArgs e)
{
if (e.BytesTransferred > 0)
{
Interlocked.Increment(ref numConnectedSockets);
Console.WriteLine( "Client connection accepted. "
"There are {0} clients connected to the server",
numConnectedSockets);
}
// 獲取接受的客戶端連接，賦給ReadEventArg對(duì)象的UserToken,。
SocketAsyncEventArgs readEventArgs = readWritePool.Pop();
readEventArgs.UserToken = e.AcceptSocket;
// 一旦客戶端連接,，提交一個(gè)連接接收。
Boolean willRaiseEvent = e.AcceptSocket.ReceiveAsync(readEventArgs);
if (!willRaiseEvent)
{
ProcessReceive(readEventArgs);
}
// 接受下一個(gè)連接請(qǐng)求,。
StartAccept(e);
}
// 當(dāng)一個(gè)異步接收操作完成時(shí)調(diào)用該方法,。
// 如果遠(yuǎn)程主機(jī)關(guān)閉了連接,，該Socket也關(guān)閉。
// 如果收到數(shù)據(jù),，則回返到客戶端,。
private void ProcessReceive(SocketAsyncEventArgs e)
{
// 檢查遠(yuǎn)程主機(jī)是否關(guān)閉了連接。
if (e.BytesTransferred > 0)
{
if (e.SocketError == SocketError.Success)
{
Socket s = e.UserToken as Socket;
Int32 bytesTransferred = e.BytesTransferred;
// 從偵聽者獲取接收到的消息,。
String received = Encoding.ASCII.GetString(e.Buffer,
e.Offset, bytesTransferred);
// 增加服務(wù)器接收的總字節(jié)數(shù),。
Interlocked.Add(ref totalBytesRead, bytesTransferred);
Console.WriteLine("Received: /"{0}/". The server has read" +
" a total of {1} bytes.", received,
totalBytesRead);
// 格式化數(shù)據(jù)后發(fā)回客戶端。
Byte [] sendBuffer =
Encoding.ASCII.GetBytes("Returning " + received);
// 設(shè)置傳回客戶端的緩沖區(qū),。
e.SetBuffer(sendBuffer, 0, sendBuffer.Length);
Boolean willRaiseEvent = s.SendAsync(e);
if (!willRaiseEvent)
{
ProcessSend(e);
}
}
else
{
CloseClientSocket(e);
}
}
}
// 當(dāng)異步發(fā)送操作完成時(shí)調(diào)用該方法。
// 當(dāng)Socket讀客戶端的任何附加數(shù)據(jù)時(shí),，該方法啟動(dòng)另一個(gè)接收操作,。
private void ProcessSend(SocketAsyncEventArgs e)
{
if (e.SocketError == SocketError.Success)
{
// 完成回發(fā)數(shù)據(jù)到客戶端。
Socket s = e.UserToken as Socket;
// 讀取從發(fā)送客戶端發(fā)送的下一個(gè)數(shù)據(jù)塊,。
Boolean willRaiseEvent = s.ReceiveAsync(e);
if (!willRaiseEvent)
{
ProcessReceive(e);
}
}
else
{
CloseClientSocket(e);
}
}
我修改了如何操作偵聽者收到消息的代碼——不是簡(jiǎn)單地回發(fā)給客戶端（參見ProcessReceive方法）,。在樣例程序中，我使用屬性Buffer,、Offset與BytesTransfered來接收消息,，SetBuffer方法把修改后的消息回返給客戶端。 為了控制偵聽者生存期時(shí)間,，使用了一個(gè)Mutex類的實(shí)例,。基于原Init方法的Start方法創(chuàng)建Mutex對(duì)象,，相應(yīng)的Stop方法釋放Mutex對(duì)象,。這些方法適用于實(shí)現(xiàn)作為Windows服務(wù)的Socket服務(wù)器。
// 啟動(dòng)服務(wù)器并開始偵聽傳入連接請(qǐng)求,。
internal void Start(Object data)
{
Int32 port = (Int32)data;
// 獲取主機(jī)相關(guān)信息,。
IPAddress[] addressList =
Dns.GetHostEntry(Environment.MachineName).AddressList;
// 獲取偵聽者所需的端點(diǎn)（endpoint）。
IPEndPoint localEndPoint =
new IPEndPoint(addressList[addressList.Length - 1], port);
// 創(chuàng)建偵聽傳入連接的Socket,。
this.listenSocket = new Socket(localEndPoint.AddressFamily,
SocketType.Stream, ProtocolType.Tcp);
if (localEndPoint.AddressFamily == AddressFamily.InterNetworkV6)
{
// 設(shè)置Socket偵聽者的雙模式（IPv4與IPv6）,。
// 27等價(jià)于IPV6_V6ONLY Socket
// Winsock片段中的如下選項(xiàng)，
// 根據(jù) Creating IP Agnostic Applications - Part 2 (Dual Mode Sockets)
// 創(chuàng)建IP的不可知應(yīng)用——第2部分（雙模式 Sockets）
this.listenSocket.SetSocketOption(SocketOptionLevel.IPv6,
(SocketOptionName)27, false);
this.listenSocket.Bind(new IPEndPoint(IPAddress.IPv6Any,
localEndPoint.Port));
}
else
{
// Socket與本地端點(diǎn)關(guān)聯(lián),。
this.listenSocket.Bind(localEndPoint);
}
// 啟動(dòng)偵聽隊(duì)列最大等待數(shù)為100個(gè)連接的服務(wù)器,。
this.listenSocket.Listen(100);
// 提交一個(gè)偵聽Socket的接收任務(wù)。
this.StartAccept(null);
mutex.WaitOne();
}
// 停止服務(wù)器,。
internal void Stop()
{
mutex.ReleaseMutex();
}
現(xiàn)在,，我們有了一個(gè)Socket服務(wù)器，下一步使用SocketAsyncEventArgs類建立一個(gè)Socket客戶端,。雖然MSDN說這個(gè)類特別設(shè)計(jì)給網(wǎng)絡(luò)服務(wù)器應(yīng)用,，但也沒有限制在客戶端代碼中使用APM,。下面給出了SocketClient類的樣例代碼：
using System;
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Threading;
namespace SocketAsyncClient
{
// 實(shí)現(xiàn)Socket客戶端的連接邏輯。
internal sealed class SocketClient: IDisposable
{
// Socket操作常數(shù),。
private const Int32 ReceiveOperation = 1, SendOperation = 0;
// 用于發(fā)送/接收消息的Socket,。
private Socket clientSocket;
// Socket連接標(biāo)志。
private Boolean connected = false;
// 偵聽者端點(diǎn),。
private IPEndPoint hostEndPoint;
// 觸發(fā)連接,。
private static AutoResetEvent autoConnectEvent =
new AutoResetEvent(false);
// 觸發(fā)發(fā)送/接收操作。
private static AutoResetEvent[]
autoSendReceiveEvents = new AutoResetEvent[]
{
new AutoResetEvent(false),
new AutoResetEvent(false)
};
// 創(chuàng)建一個(gè)未初始化的客戶端實(shí)例,。
// 啟動(dòng)傳送/接收處理將調(diào)用Connect方法,，然后是SendReceive方法。
internal SocketClient(String hostName, Int32 port)
{
// 獲取主機(jī)有關(guān)的信息,。
IPHostEntry host = Dns.GetHostEntry(hostName);
// 主機(jī)地址,。
IPAddress[] addressList = host.AddressList;
// 實(shí)例化端點(diǎn)和Socket。
hostEndPoint = new IPEndPoint(addressList[addressList.Length - 1], port);
clientSocket = new Socket(hostEndPoint.AddressFamily,
SocketType.Stream, ProtocolType.Tcp);
// 連接主機(jī),。
internal void Connect()
{
SocketAsyncEventArgs connectArgs = new SocketAsyncEventArgs();
connectArgs.UserToken = clientSocket;
connectArgs.RemoteEndPoint = hostEndPoint;
connectArgs.Completed +=
new EventHandler<socketasynceventargs>(OnConnect);
clientSocket.ConnectAsync(connectArgs);
autoConnectEvent.WaitOne();
SocketError errorCode = connectArgs.SocketError;
if (errorCode != SocketError.Success)
{
throw new SocketException((Int32)errorCode);
}
}
/// 與主機(jī)斷開連接,。
internal void isconnect()
{
clientSocket.Disconnect(false);
}
// 連接操作的回調(diào)方法
private void OnConnect(object sender, SocketAsyncEventArgs e)
{
// 發(fā)出連接完成信號(hào)。
autoConnectEvent.Set();
// 設(shè)置Socket已連接標(biāo)志,。
connected = (e.SocketError == SocketError.Success);
}
// 接收操作的回調(diào)方法
private void OnReceive(object sender, SocketAsyncEventArgs e)
{
// 發(fā)出接收完成信號(hào),。
autoSendReceiveEvents[SendOperation].Set();
}
// 發(fā)送操作的回調(diào)方法
private void OnSend(object sender, SocketAsyncEventArgs e)
{
// 發(fā)出發(fā)送完成信號(hào)。
autoSendReceiveEvents[ReceiveOperation].Set();
if (e.SocketError == SocketError.Success)
{
if (e.LastOperation == SocketAsyncOperation.Send)
{
// 準(zhǔn)備接收,。
Socket s = e.UserToken as Socket;
byte [] receiveBuffer = new byte [255];
e.SetBuffer(receiveBuffer, 0, receiveBuffer.Length);
e.Completed += new EventHandler<socketasynceventargs>(OnReceive);
s.ReceiveAsync(e);
}
}
else
{
ProcessError(e);
}
}
// 失敗時(shí)關(guān)閉Socket,，根據(jù)SocketError拋出異常。
private void ProcessError(SocketAsyncEventArgs e)
{
Socket s = e.UserToken as Socket;
if (s.Connected)
{
// 關(guān)閉與客戶端關(guān)聯(lián)的Socket
try
{
s.Shutdown(SocketShutdown.Both);
}
catch (Exception)
{
// 如果客戶端處理已經(jīng)關(guān)閉,，拋出異常
}
finally
{
if (s.Connected)
{
s.Close();
}
}
}
// 拋出SocketException
throw new SocketException((Int32)e.SocketError);
}
// 與主機(jī)交換消息,。
internal String SendReceive(String message)
{
if (connected)
{
// 創(chuàng)建一個(gè)發(fā)送緩沖區(qū)。
Byte [] sendBuffer = Encoding.ASCII.GetBytes(message);
// 準(zhǔn)備發(fā)送/接收操作的參數(shù),。
SocketAsyncEventArgs completeArgs = new SocketAsyncEventArgs();
completeArgs.SetBuffer(sendBuffer, 0, sendBuffer.Length);
completeArgs.UserToken = clientSocket;
completeArgs.RemoteEndPoint = hostEndPoint;
completeArgs.Completed +=
new EventHandler<socketasynceventargs>(OnSend);
// 開始異步發(fā)送,。
clientSocket.SendAsync(completeArgs);
// 等待發(fā)送/接收完成。
AutoResetEvent.WaitAll(autoSendReceiveEvents);
// 從SocketAsyncEventArgs緩沖區(qū)返回?cái)?shù)據(jù),。
return Encoding.ASCII.GetString(completeArgs.Buffer,
completeArgs.Offset, completeArgs.BytesTransferred);
}
else
{
throw new SocketException((Int32)SocketError.NotConnected);
}
}
#region IDisposable Members
// 釋放SocketClient實(shí)例,。
public void Dispose()
{
autoConnectEvent.Close();
autoSendReceiveEvents[SendOperation].Close();
autoSendReceiveEvents[ReceiveOperation].Close();
if (clientSocket.Connected)
{
clientSocket.Close();
}
}
#endregion
}
}
興趣點(diǎn)
我
有服務(wù)器群場(chǎng)景下的Socket服務(wù)器運(yùn)行的經(jīng)驗(yàn)。這種場(chǎng)景中,，不能使用主機(jī)地址列表的第一項(xiàng),，而要使用最后一項(xiàng)，在前面的Start方法中可以看到這一
點(diǎn),。另一個(gè)技巧就是如何為IP6地址族設(shè)置雙模式,，這對(duì)于那些想在Windows Vista和Windows Server
2008上運(yùn)行Socket服務(wù)器是有幫助的，它們默認(rèn)IP6。 本文的兩個(gè)程序都使用命令行參數(shù)運(yùn)行,。如果服務(wù)器和客戶端均運(yùn)行在一個(gè)Windows域之外的機(jī)器上,，客戶端代碼必須替換“l(fā)ocalhost”為主機(jī)名而不是機(jī)器名。
歷史
- 15 January, 2008 - 提交初版,。
作為IOCP關(guān)鍵類SocketAsyncEventArgs的補(bǔ)充知識(shí),，摘抄2007年9月MSDN雜志上的“
連接.NET框架3.5”的部分內(nèi)容如下： .NET
Framework中的APM也稱為Begin/End模式。這是因?yàn)闀?huì)調(diào)用Begin方法來啟動(dòng)異步操作,，然后返回一個(gè)IAsyncResult
對(duì)象,。可以選擇將一個(gè)代理作為參數(shù)提供給Begin方法,，異步操作完成時(shí)會(huì)調(diào)用該方法,。或者,，一個(gè)線程可以等待
IAsyncResult.AsyncWaitHandle,。當(dāng)回調(diào)被調(diào)用或發(fā)出等待信號(hào)時(shí)，就會(huì)調(diào)用End方法來獲取異步操作的結(jié)果,。這種模式很靈活，
使用相對(duì)簡(jiǎn)單,，在 .NET Framework 中非常常見,。 但是，您必須注意,，如果進(jìn)行大量異步套接字操作,，是要付出代價(jià)的。針
對(duì)每次操作,，都必須創(chuàng)建一個(gè)IAsyncResult對(duì)象,，而且該對(duì)象不能被重復(fù)使用。由于大量使用對(duì)象分配和垃圾收集,，這會(huì)影響性能,。為了解決這個(gè)問
題，新版本提供了另一個(gè)使用套接字上執(zhí)行異步I/O的方法模式,。這種新模式并不要求為每個(gè)套接字操作分配操作上下文對(duì)象,。 我們沒有創(chuàng)建全新的模式，而只是采用現(xiàn)有模式并做了一個(gè)基本更改?，F(xiàn)在,，在Socket類中有了一些方法，它們使用基于事件的完成模型的變體,。在 2.0 版本中,，您可以使用下列代碼在某個(gè)套接字上啟動(dòng)異步發(fā)送操作：
void OnSendCompletion(IAsyncResult ar) { }
IAsyncResult ar = socket.BeginSend(buffer, 0, buffer.Length,
SocketFlags.None, OnSendCompletion, state);
在新版本中，您還可以實(shí)現(xiàn)：
void OnSendCompletion(object src, SocketAsyncEventArgs sae) { }
SocketAsyncEventArgs sae = new SocketAsyncEventArgs();
sae.Completed += OnSendCompletion;
sae.SetBuffer(buffer, 0, buffer.Length);
socket.SendAsync(sae);
這
里有一些明顯的差別。封裝操作上下文的是一個(gè)SocketAsyncEventArgs對(duì)象,，而不是IAsyncResult對(duì)象,。該應(yīng)用程序創(chuàng)建并管理
（甚至可以重復(fù)使用）SocketAsyncEventArgs對(duì)象。套接字操作的所有參數(shù)都由SocketAsyncEventArgs對(duì)象的屬性和方
法指定,。完成狀態(tài)也由SocketAsyncEventArgs對(duì)象的屬性提供,。最后，需要使用事件處理程序回調(diào)完成方法,。
