from:http://www.cnblogs.com/keyindex/archive/2010/12/15/1901875.html
最近想写一个网络服务器端的程序，想看看在大量客户端数下程序的运行情况。于是对.net 的Socket编程进行了一些研究，发现.net 3.5 里SocketAsyncEventArgs 是基于IOCP实现。MSDN上有相关的示例，但它是基于TCP协议的，而我想要的是基于UDP协议的。网上很难找到基于UDP协议的SocketAsyncEventArgs示例（UDP需要用IOCP吗？），于是决定自己写一个基于UDP协议的SocketAsyncEventArgs示例，看看它在和大量客户端通讯时的运行情况。
示例简介
程序分为服务器端和客户端，它们使用UDP协议进行通讯。众所周知UDP是无连接的，可我又想计算出有多少客户端和服务器通信，其中又有多少是新的客户端。所以设计让服务器端程序绑定两个端口。一个端口专门用于接收客户端第一次发送过来的数据包；另一个端口负责和已经接入的客户端进行通讯（是不是有点像TCP的接入，例子本身也在模仿Tcp编程）。客户端比较简单让它生成足够多的Socket，然后不断的向服务器端发送数据包即可。
UDPReceiceSocket
UdpReceiceSocket类负责接收从客户端发送来的数据包，客户端第一次发送过来的数据包和后期的通讯数据都是由它负责接收的。新建实例时会创建一个 Socket 套接字和一个SocketAsyncEventArgs,并且对其进行相应的初始化。OnDataReceived 事件用于计算收到了多少数据。
代码
public class UdpReceiveSocket
{
/// <summary>
/// 接收用Socket;
/// </summary>
private Socket receiveSocket;
private SocketAsyncEventArgs receiveSocketArgs;
private IPEndPoint localEndPoint;
private byte[] receivebuffer;
/// <summary>
/// 接收到数据包时触发
/// </summary>
public event EventHandler<SocketAsyncEventArgs> OnDataReceived;
/// <summary>
/// 初始化Socket 和 地址。
/// </summary>
/// <param name="port">端口</param>
/// <param name="socket"></param>
public UdpReceiveSocket(int port)
{
receiveSocket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp);
localEndPoint = new IPEndPoint(IPAddress.Any, port);
receiveSocket.Bind(localEndPoint);
receivebuffer = new byte[1024];
receiveSocketArgs = new SocketAsyncEventArgs();
receiveSocketArgs.RemoteEndPoint = localEndPoint;
receiveSocketArgs .Completed +=new EventHandler<SocketAsyncEventArgs>(receiveSocketArgs_Completed);
receiveSocketArgs.SetBuffer(receivebuffer, 0, receivebuffer.Length);
}
/// <summary>
/// 开始接收数据
/// </summary>
public void StartReceive()
{
if (!receiveSocket.ReceiveFromAsync(receiveSocketArgs ))
{
processReceived(receiveSocketArgs);
}
}
void receiveSocketArgs_Completed(object sender, SocketAsyncEventArgs e)
{
switch (e.LastOperation)
{
case SocketAsyncOperation.ReceiveFrom:
this.processReceived(e);
break;
default:
throw new ArgumentException("The last operation completed on the socket was not a receive");
}
}
/// <summary>
/// 接收完成时处理请求。
/// </summary>
/// <param name="sender"></param>
/// <param name="e"></param>
void processReceived( SocketAsyncEventArgs e)
{
if (e.BytesTransferred > 0 && e.SocketError == SocketError.Success)
{
if (OnDataReceived != null)
{
//不要进行耗时操作
OnDataReceived(receiveSocket , e);
}
}
StartReceive();
}
UDPSendSocket
负责发送数据包给客户端。
代码
public class UdpSendSocket
{
private SocketAsyncEventArgsPool socketArgsPool;
private BufferManager bfManager;
private Socket socket;
private SocketAsyncEventArgs socketArgs;
private int numClient;
public event EventHandler<SocketAsyncEventArgs> DataSent;
private static readonly object asyncLock = new object();
/// <summary>
/// 最大客户端数
/// </summary>
/// <param name="numClient"></param>
public UdpSendSocket(int numClient)
{
socket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp);
this.numClient = numClient;
int bufferSize = 1024;
bfManager = new BufferManager(numClient * bufferSize * 2, bufferSize);
socketArgsPool = new SocketAsyncEventArgsPool(numClient);
}
/// <summary>
/// 初始化
/// </summary>
public void Init()
{
//初始化数据池
bfManager.InitBuffer();
//生成一定数量的对象池
for (int i = 0; i < numClient; i++)
{
socketArgs = new SocketAsyncEventArgs();
socketArgs .Completed +=new EventHandler<SocketAsyncEventArgs>(socketArgs_Completed);
//设置SocketAsyncEventArgs的Buffer
bfManager.SetBuffer(socketArgs);
socketArgsPool.Push(socketArgs);
}
}
/// <summary>
/// 发送数据包
/// </summary>
/// <param name="data">要发送的数据包</param>
public void Send(EndPoint remoteEndPoint)
{
//每次发送前都取一个新的SocketAsyncEventArgs对象。
socketArgs = socketArgsPool.Pop();
socketArgs.RemoteEndPoint = remoteEndPoint;
if (socketArgs.RemoteEndPoint != null)
{
if (!socket.SendToAsync(socketArgs))
{
ProcessSent(socketArgs);
}
}
}
public void Send(byte[] content, EndPoint remoteEndPoint)
{
socketArgs = socketArgsPool.Pop();
socketArgs.RemoteEndPoint = remoteEndPoint;
//设置发送的内容
bfManager.SetBufferValue(socketArgs, content);
if (socketArgs.RemoteEndPoint != null)
{
if (!socket.SendToAsync(socketArgs))
{
ProcessSent(socketArgs);
}
}
}
private void socketArgs_Completed(object sender, SocketAsyncEventArgs e)
{
switch (e.LastOperation)
{
case SocketAsyncOperation.SendTo:
this.ProcessSent(e);
break;
default:
throw new ArgumentException("The last operation completed on the socket was not a send");
}
}
private void ProcessSent(SocketAsyncEventArgs e)
{
if (e.BytesTransferred > 0 && e.SocketError == SocketError.Success)
{
if (DataSent != null)
{
//用于统计发送了多少数据
DataSent(socket, e);
}
}
//发送完成后将SocketAsyncEventArgs对象放回对象池中
socketArgsPool.Push(e);
}
}
UdpServer
主要负责对UdpReveivedSocket 和UdpSendSocket进行封装，初始化一些对象，并以事件方式提供数据接收和发送的处理。
代码
public class UdpServer
{
/// <summary>
/// 接收新客户端的端口
/// </summary>
private int listenPort;
/// <summary>
/// 数据通讯的端口
/// </summary>
private int CommunicationPort;
/// <summary>
/// 最大的客户端数
/// </summary>
private int numClient;
/// <summary>
/// 负责接收新的客户端的数据
/// </summary>
private UdpReceiveSocket listen;
/// <summary>
/// 负责接收旧客户端的数据
/// </summary>
private UdpReceiveSocket communicationRec;
/// <summary>
/// 负责向客户端发送数据
/// </summary>
private UdpSendSocket communicationSend;
private bool isStartSend;
public event EventHandler<SocketAsyncEventArgs> OnNewClient;
public event EventHandler<SocketAsyncEventArgs> OnReceivedData;
public event EventHandler<SocketAsyncEventArgs> OnSentData;
/// <summary>
///
/// </summary>
/// <param name="acceptPort">新客户端接收端口</param>
/// <param name="dataComPort">数据通讯端口</param>
/// <param name="maxNumClient">最大客户端数</param>
public UdpServer(int acceptPort, int dataComPort, int maxNumClient)
{
listenPort = acceptPort;
CommunicationPort = dataComPort;
numClient = maxNumClient;
isStartSend = false;
listen = new UdpReceiveSocket(listenPort);
listen.OnDataReceived += new EventHandler<SocketAsyncEventArgs>(listen_OnDataReceived);
communicationRec = new UdpReceiveSocket(CommunicationPort);
communicationRec.OnDataReceived += new EventHandler<SocketAsyncEventArgs>(communicationRec_OnDataReceived);
communicationSend = new UdpSendSocket(numClient);
communicationSend.DataSent += new EventHandler<SocketAsyncEventArgs>(communicationSend_DataSent);
communicationSend.Init();
}
public void Start()
{
//接收新的客户端
listen.StartReceive();
//接收数据
communicationRec.StartReceive();
}
#region 事件
void communicationRec_OnDataReceived(object sender, SocketAsyncEventArgs e)
{
if (OnReceivedData != null)
{
OnReceivedData(sender, e);
}
//向客户端发送数据
communicationSend.Send(e.RemoteEndPoint);
}
void listen_OnDataReceived(object sender, SocketAsyncEventArgs e)
{
if (OnNewClient != null)
{
OnNewClient(sender, e);
}
}
void communicationSend_DataSent(object sender, SocketAsyncEventArgs e)
{
if (OnSentData != null)
{
OnSentData(sender, e);
}
}
#endregion
}
控制台代码
直接上代码
代码
class Program
{
/// <summary>
/// 当前通信中的客户端
/// </summary>
static Int32 numClient = 0;
/// <summary>
/// 当前收到的新客户端
/// </summary>
static Int32 client = 0;
/// <summary>
/// 即时收到消息数
/// </summary>
static Int32 receivedMessages = 0;
/// <summary>
/// 即时发送消息数
/// </summary>
static Int32 sentMessages = 0;
/// <summary>
/// 即时收到字节
/// </summary>
static Int32 receivedBytes = 0;
/// <summary>
/// 即时发送消息数
/// </summary>
static Int32 sentBytes = 0;
static void Main(string[] args)
{
Console.WriteLine("输入监听端口号");
int listenPort = int.Parse(Console.ReadLine());
Console.WriteLine("输入通讯端口号");
int port = int.Parse(Console.ReadLine());
Console.WriteLine("最大允许的客户端数");
int numClient = int.Parse(Console.ReadLine());
UdpServer server = new UdpServer(listenPort, port, numClient);
server.OnNewClient += new EventHandler<SocketAsyncEventArgs>(server_OnNewClientAccept);
server.OnReceivedData += new EventHandler<SocketAsyncEventArgs>(server_OnDataReceived);
server.OnSentData += new EventHandler<SocketAsyncEventArgs>(server_OnDataSending);
server.Start();
Timer timer = new Timer(new TimerCallback(DrawDisplay), null, 200, 1000);
Console.ReadKey();
}
#region 事件处理
static void server_OnDataReceived(object sender, EventArgs e)
{
receivedMessages = Interlocked.Increment(ref receivedMessages);
receivedBytes = Interlocked.Add(ref receivedBytes, (e as SocketAsyncEventArgs).BytesTransferred);
}
static void server_OnNewClientAccept(object sender, EventArgs e)
{
numClient = Interlocked.Increment(ref numClient);
client = Interlocked.Increment(ref client);
}
static void server_OnDataSending(object sender, EventArgs e)
{
sentMessages = Interlocked.Increment(ref sentMessages);
sentBytes = Interlocked.Add(ref sentBytes, (e as SocketAsyncEventArgs).BytesTransferred);
}
#endregion
static void DrawDisplay(Object obj)
{
Console.Clear();
Console.WriteLine(String.Format("服务器 运行中...\n\n" +
"新客户端数：{0}\n"+
"当前客户端数： {1}\n\n" +
"当前每秒收到消息： {2}\n" +
"当前每秒发送消息：{3}\n\n"+
"当前每秒收到字节：{4}\n"+
"当前每秒发送字节：{5}\n\n"+
"按任意键结束。。。。。。。。。。。",client, numClient , receivedMessages ,sentMessages,receivedBytes ,sentBytes));
receivedBytes = 0;
receivedMessages = 0;
sentBytes = 0;
sentMessages = 0;
client = 0;
}
}
测试用客户端
客户端的设计和UdpSendSocket很像，为了模拟大量的套接字，每个套接字使用不同的端口循环向服务器端发送数据包（为了简便操作这里没有设计客户端的数据接收）。比较麻烦的事Socket是很耗资源的，而且系统的端口数也有限（65535），这给我后面的测试多少带来些麻烦。
代码
class Program
{
static string serverIp;
static string name = string.Empty;
static IPEndPoint ipe;
static Socket server;
static int port;
static int port2;
static Mutex mutex = new Mutex();
static void Main(string[] args)
{
Console.WriteLine("输入服务器IP地址");
serverIp = Console.ReadLine();
Console.WriteLine("输入端口");
port = int.Parse(Console.ReadLine());
Console.WriteLine("输入传输端口");
port2 = int.Parse(Console.ReadLine());
ipe = new IPEndPoint(IPAddress.Parse(serverIp), port);
while (run()) ;
Console.ReadLine();
}
//循环测试，如果输入连接数小于0将退出
static bool run()
{
int txtNum;
Console.WriteLine("输入最大连接数");
txtNum = int.Parse(Console.ReadLine());
for (int i = 0; i < txtNum; i++)
{
StartSend(txtNum);
}
return txtNum > 0;
}
/// <summary>
/// 创建一个新的Socket 和SocketAsyncEventArgs 对象，并开始向服务器端发送数据
/// </summary>
/// <param name="txtNum"></param>
static void StartSend(int txtNum)
{
server = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp);
SocketAsyncEventArgs socketAsyncEventArgs = new SocketAsyncEventArgs();
string welcome = "测试" + ",申请链接,";
byte[] data = new byte[1024];
data = Encoding.Unicode.GetBytes(welcome);
socketAsyncEventArgs.SetBuffer(data, 0, data.Length);
socketAsyncEventArgs.UserToken = server;
socketAsyncEventArgs.Completed += new EventHandler<SocketAsyncEventArgs>(socketAsyncEventArgs_Completed);
//向服务器端发送第一个数据包，相当于申请链接
server.SendTo(data,ipe );
//向服务器进行异步的数据发送
Send(socketAsyncEventArgs);
Thread.Sleep(20);
}
static void Send(SocketAsyncEventArgs e)
{
Socket s = e.UserToken as Socket;
e.RemoteEndPoint = new IPEndPoint(IPAddress.Parse(serverIp), port2);
s.SendToAsync(e);
}
static void socketAsyncEventArgs_Completed(object sender, SocketAsyncEventArgs e)
{
Thread.Sleep(3000);
//循环的发送数据
Send(e);
}
}
测试
本次的测试目标是：10万个客户端和服务器程序通讯时的运行情况。因为每台电脑的端口有限最大也就65535 ，而且我们也不可能开到这么多个套接字，所以我就在不同的机器上运行了上面的测试用客户端程序。
测试环境：局域网。
服务器端程序运行环境：Win7 操作系统、2GB内存、CPU:P core E5200 2.50GHz。
客户端环境：3台电脑、一台win7 两台Win2003。
运行服务器端程序设置两个端口，并设定最大的客户端数为110000。下面是客户端没有发送数据时的计数器截图：
接下来就是在各台电脑上奔走，运行客户端的测试程序向服务器发送数据包。为了加快并发数，实际上我实在每台电脑上运行多个客户端程序。下图是在客户端连接数到达6万时的计数器情况。
对比两张图片可以看出Private Bytes 和 Virtual Bytes 即 Handle Count 有所上升。Context Switches线程间的切换开销变大很多。接下来看看连接数上升到9万时的情况。
Private Bytes还在慢慢上升当中，难道有Memory Leak！最后再来一张客户端数达到10万时的计数器截图
整个过程中Handle Count 和Private Bytes总在上升当中(也有可能是因为不断的有新来的客户端加入的原因)，程序可能有Handle Leak和Memory Leak。