- 原文&＃xff1a;How to use the SocketAsyncEventArgs class. by Marcos Hidalgo Nunes
Download client - 4.09 KB - Download server - 7.5 KB
引言
我一直在探寻一个高性能的Socket客户端代码。以前&＃xff0c;我使用Socket类写了一些基于传统异步编程模型的代码&＃xff08;BeginSend、BeginReceive&＃xff0c;等等&＃xff09;。但它没有满足我所要的性能需求。终于&＃xff0c;我找到了基于事件的异步操作新模式&＃xff08;参见2007年9月MSDN杂志上的“连接.NET框架3.5”&＃xff09;&＃xff08;部分内容见文后的翻译附注——译者注&＃xff09;。
背景
由于减少了阻塞线程&＃xff0c;高性能I/O限制应用中广泛使用异步编程模型&＃xff08;AMP&＃xff0c;Asynchronous Programming Model&＃xff09;。.NET Framework第一个版本就实现了APM&＃xff0c;现在使用诸如lambda表达式等新的技术C#3.0一直在改进其性能。针对Socket编程&＃xff0c;不仅性能上提升了不少&＃xff0c;而且新APM模型发布了一个更简易的编程方法&＃xff0c;该方法使用SocketAsyncEventArgs类来保持I/O操作之间的上下文&＃xff08;见文后的翻译附注——译者注&＃xff09;&＃xff0c;从而降低对象分配和垃圾收集工作。
在.NET 2.0 SP1上可以使用SocketAsyncEventArgs类&＃xff0c;本文的代码就是用Microsoft Visual Studio .NET 2005编写的。
使用代码
从SocketAsyncEventArgs类开始&＃xff0c;我学习了MSDN上的样例程序&＃xff0c;但该文缺少一些内容&＃xff1a;AsyncUserToken类。我认为这个类应该公开一个Socket属性&＃xff0c;它对应执行I/O操作的Socket。一段时间后&＃xff0c;我认识到这个类不是必要的&＃xff0c;因为属性UserToken是一个Object&＃xff0c;它可以接受任何东西。下面的修改方法中直接使用一个Socket实例当作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);
}
// 获取接受的客户端连接&＃xff0c;赋给ReadEventArg对象的UserToken。
SocketAsyncEventArgs readEventArgs &＃61; readWritePool.Pop();
readEventArgs.UserToken &＃61; e.AcceptSocket;
// 一旦客户端连接&＃xff0c;提交一个连接接收。
Boolean willRaiseEvent &＃61; e.AcceptSocket.ReceiveAsync(readEventArgs);
if (!willRaiseEvent)
{
ProcessReceive(readEventArgs);
}
// 接受下一个连接请求。
StartAccept(e);
}
// 当一个异步接收操作完成时调用该方法。
// 如果远程主机关闭了连接&＃xff0c;该Socket也关闭。
// 如果收到数据&＃xff0c;则回返到客户端。
private void ProcessReceive(SocketAsyncEventArgs e)
{
// 检查远程主机是否关闭了连接。
if (e.BytesTransferred > 0)
{
if (e.SocketError &＃61;&＃61; SocketError.Success)
{
Socket s &＃61; e.UserToken as Socket;
Int32 bytesTransferred &＃61; e.BytesTransferred;
// 从侦听者获取接收到的消息。
String received &＃61; Encoding.ASCII.GetString(e.Buffer,
e.Offset, bytesTransferred);
// 增加服务器接收的总字节数。
Interlocked.Add(ref totalBytesRead, bytesTransferred);
Console.WriteLine("Received: /"{0}/". The server has read" &＃43;
" a total of {1} bytes.", received,
totalBytesRead);
// 格式化数据后发回客户端。
Byte [] sendBuffer &＃61;
Encoding.ASCII.GetBytes("Returning " &＃43; received);
// 设置传回客户端的缓冲区。
e.SetBuffer(sendBuffer, 0, sendBuffer.Length);
Boolean willRaiseEvent &＃61; s.SendAsync(e);
if (!willRaiseEvent)
{
ProcessSend(e);
}
}
else
{
CloseClientSocket(e);
}
}
}
// 当异步发送操作完成时调用该方法。
// 当Socket读客户端的任何附加数据时&＃xff0c;该方法启动另一个接收操作。
private void ProcessSend(SocketAsyncEventArgs e)
{
if (e.SocketError &＃61;&＃61; SocketError.Success)
{
// 完成回发数据到客户端。
Socket s &＃61; e.UserToken as Socket;
// 读取从发送客户端发送的下一个数据块。
Boolean willRaiseEvent &＃61; s.ReceiveAsync(e);
if (!willRaiseEvent)
{
ProcessReceive(e);
}
}
else
{
CloseClientSocket(e);
}
}我修改了如何操作侦听者收到消息的代码——不是简单地回发给客户端&＃xff08;参见ProcessReceive方法&＃xff09;。在样例程序中&＃xff0c;我使用属性Buffer、Offset与BytesTransfered来接收消息&＃xff0c;SetBuffer方法把修改后的消息回返给客户端。
为了控制侦听者生存期时间&＃xff0c;使用了一个Mutex类的实例。基于原Init方法的Start方法创建Mutex对象&＃xff0c;相应的Stop方法释放Mutex对象。这些方法适用于实现作为Windows服务的Socket服务器。
// 启动服务器并开始侦听传入连接请求。
internal void Start(Object data)
{
Int32 port &＃61; (Int32)data;
// 获取主机相关信息。
IPAddress[] addressList &＃61;
Dns.GetHostEntry(Environment.MachineName).AddressList;
// 获取侦听者所需的端点&＃xff08;endpoint&＃xff09;。
IPEndPoint localEndPoint &＃61;
new IPEndPoint(addressList[addressList.Length - 1], port);
// 创建侦听传入连接的Socket。
this.listenSocket &＃61; new Socket(localEndPoint.AddressFamily,
SocketType.Stream, ProtocolType.Tcp);
if (localEndPoint.AddressFamily &＃61;&＃61; AddressFamily.InterNetworkV6)
{
// 设置Socket侦听者的双模式&＃xff08;IPv4与IPv6&＃xff09;。
// 27等价于IPV6_V6ONLY Socket
// Winsock片段中的如下选项&＃xff0c;
// 根据 Creating IP Agnostic Applications - Part 2 (Dual Mode Sockets)
// 创建IP的不可知应用——第2部分&＃xff08;双模式 Sockets&＃xff09;
this.listenSocket.SetSocketOption(SocketOptionLevel.IPv6,
(SocketOptionName)27, false);
this.listenSocket.Bind(new IPEndPoint(IPAddress.IPv6Any,
localEndPoint.Port));
}
else
{
// Socket与本地端点关联。
this.listenSocket.Bind(localEndPoint);
}
// 启动侦听队列最大等待数为100个连接的服务器。
this.listenSocket.Listen(100);
// 提交一个侦听Socket的接收任务。
this.StartAccept(null);
mutex.WaitOne();
}
// 停止服务器。
internal void Stop()
{
mutex.ReleaseMutex();
}现在&＃xff0c;我们有了一个Socket服务器&＃xff0c;下一步使用SocketAsyncEventArgs类建立一个Socket客户端。虽然MSDN说这个类特别设计给网络服务器应用&＃xff0c;但也没有限制在客户端代码中使用APM。下面给出了SocketClient类的样例代码&＃xff1a;
using System; void OnSendCompletion(IAsyncResult ar) { } void OnSendCompletion(object src, SocketAsyncEventArgs sae) { }
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Threading;
namespace SocketAsyncClient
{
// 实现Socket客户端的连接逻辑。
internal sealed class SocketClient: IDisposable
{
// Socket操作常数。
private const Int32 ReceiveOperation &＃61; 1, SendOperation &＃61; 0;
// 用于发送/接收消息的Socket。
private Socket clientSocket;
// Socket连接标志。
private Boolean connected &＃61; false;
// 侦听者端点。
private IPEndPoint hostEndPoint;
// 触发连接。
private static AutoResetEvent autoConnectEvent &＃61;
new AutoResetEvent(false);
// 触发发送/接收操作。
private static AutoResetEvent[]
autoSendReceiveEvents &＃61; new AutoResetEvent[]
{
new AutoResetEvent(false),
new AutoResetEvent(false)
};
// 创建一个未初始化的客户端实例。
// 启动传送/接收处理将调用Connect方法&＃xff0c;然后是SendReceive方法。
internal SocketClient(String hostName, Int32 port)
{
// 获取主机有关的信息。
IPHostEntry host &＃61; Dns.GetHostEntry(hostName);
// 主机地址。
IPAddress[] addressList &＃61; host.AddressList;
// 实例化端点和Socket。
hostEndPoint &＃61; new IPEndPoint(addressList[addressList.Length - 1], port);
clientSocket &＃61; new Socket(hostEndPoint.AddressFamily,
SocketType.Stream, ProtocolType.Tcp);
// 连接主机。
internal void Connect()
{
SocketAsyncEventArgs connectArgs &＃61; new SocketAsyncEventArgs();
connectArgs.UserToken &＃61; clientSocket;
connectArgs.RemoteEndPoint &＃61; hostEndPoint;
connectArgs.Completed &＃43;&＃61;
new EventHandler
clientSocket.ConnectAsync(connectArgs);
autoConnectEvent.WaitOne();
SocketError errorCode &＃61; connectArgs.SocketError;
if (errorCode !&＃61; SocketError.Success)
{
throw new SocketException((Int32)errorCode);
}
}
/// 与主机断开连接。
internal void isconnect()
{
clientSocket.Disconnect(false);
}
// 连接操作的回调方法
private void OnConnect(object sender, SocketAsyncEventArgs e)
{
// 发出连接完成信号。
autoConnectEvent.Set();
// 设置Socket已连接标志。
connected &＃61; (e.SocketError &＃61;&＃61; SocketError.Success);
}
// 接收操作的回调方法
private void OnReceive(object sender, SocketAsyncEventArgs e)
{
// 发出接收完成信号。
autoSendReceiveEvents[SendOperation].Set();
}
// 发送操作的回调方法
private void OnSend(object sender, SocketAsyncEventArgs e)
{
// 发出发送完成信号。
autoSendReceiveEvents[ReceiveOperation].Set();
if (e.SocketError &＃61;&＃61; SocketError.Success)
{
if (e.LastOperation &＃61;&＃61; SocketAsyncOperation.Send)
{
// 准备接收。
Socket s &＃61; e.UserToken as Socket;
byte [] receiveBuffer &＃61; new byte [255];
e.SetBuffer(receiveBuffer, 0, receiveBuffer.Length);
e.Completed &＃43;&＃61; new EventHandler
s.ReceiveAsync(e);
}
}
else
{
ProcessError(e);
}
}
// 失败时关闭Socket&＃xff0c;根据SocketError抛出异常。
private void ProcessError(SocketAsyncEventArgs e)
{
Socket s &＃61; e.UserToken as Socket;
if (s.Connected)
{
// 关闭与客户端关联的Socket
try
{
s.Shutdown(SocketShutdown.Both);
}
catch (Exception)
{
// 如果客户端处理已经关闭&＃xff0c;抛出异常
}
finally
{
if (s.Connected)
{
s.Close();
}
}
}
// 抛出SocketException
throw new SocketException((Int32)e.SocketError);
}
// 与主机交换消息。
internal String SendReceive(String message)
{
if (connected)
{
// 创建一个发送缓冲区。
Byte [] sendBuffer &＃61; Encoding.ASCII.GetBytes(message);
// 准备发送/接收操作的参数。
SocketAsyncEventArgs completeArgs &＃61; new SocketAsyncEventArgs();
completeArgs.SetBuffer(sendBuffer, 0, sendBuffer.Length);
completeArgs.UserToken &＃61; clientSocket;
completeArgs.RemoteEndPoint &＃61; hostEndPoint;
completeArgs.Completed &＃43;&＃61;
new EventHandler
// 开始异步发送。
clientSocket.SendAsync(completeArgs);
// 等待发送/接收完成。
AutoResetEvent.WaitAll(autoSendReceiveEvents);
// 从SocketAsyncEventArgs缓冲区返回数据。
return Encoding.ASCII.GetString(completeArgs.Buffer,
completeArgs.Offset, completeArgs.BytesTransferred);
}
else
{
throw new SocketException((Int32)SocketError.NotConnected);
}
}
#region IDisposable Members
// 释放SocketClient实例。
public void Dispose()
{
autoConnectEvent.Close();
autoSendReceiveEvents[SendOperation].Close();
autoSendReceiveEvents[ReceiveOperation].Close();
if (clientSocket.Connected)
{
clientSocket.Close();
}
}
#endregion
}
} 兴趣点
历史
翻译附注
IAsyncResult ar &＃61; socket.BeginSend(buffer, 0, buffer.Length,
SocketFlags.None, OnSendCompletion, state);在新版本中&＃xff0c;您还可以实现&＃xff1a;
SocketAsyncEventArgs sae &＃61; new SocketAsyncEventArgs();
sae.Completed &＃43;&＃61; OnSendCompletion;
sae.SetBuffer(buffer, 0, buffer.Length);
socket.SendAsync(sae);这里有一些明显的差别。封装操作上下文的是一个SocketAsyncEventArgs对象&＃xff0c;而不是IAsyncResult对象。该应用程序创建并管理&＃xff08;甚至可以重复使用&＃xff09;SocketAsyncEventArgs对象。套接字操作的所有参数都由SocketAsyncEventArgs对象的属性和方法指定。完成状态也由SocketAsyncEventArgs对象的属性提供。最后&＃xff0c;需要使用事件处理程序回调完成方法。
京公网安备 11010802041100号