一次.net Socket UDP编程的10万客户端测试记录

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();
           
       }

用StartReceive()方法启动接收。根据Socket类的 ReceiveFromAsync（）方法返回值判断操作是异步还是同步，如果返回值为False 时说明是同步的操作（说明在运行ReceiveFromAsync方法前数据已经准备好了，即已接收到客户端发来的数据包，并且放在了通知队列中），需要直接调用processReceived方法处理。在processReceived方法里又调用了StartReceive：处理完接收后再次进行数据接收，形成一个循环。注：一个Socket中的 receive 和send 最好分别对应一个SocketAsyncEventArgs，当同一个SocketAsyncEventArgs挂起后再调用receiveAsync 或者 SendAsync方法时会抛出异常。

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);
         
       }

    }

SocketAsyncEventArgsPool和BufferManager 是MSDN上的示例。在Init（）方法里初始化一定数量的SocketAsyncEventArgs对象，发送数据时从对象池中取出一个SocketAsyncEventArgs完成发送后再将其放回池中。使用对象池的好处就是不用反复创建SocketAsyncEventArgs对象减少消耗。

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
    }

在communicationRec_OnDataReceived事件里，接收到客户端的数据后立刻向数据的发送数据。其实可以将数据的发送独立开来，在listen_OnDataReceived里记录远程的客户端信息（例如：EndPoint队列）,然后通过 communicationSend.Send()进行数据的发送出去。

控制台代码

　　直接上代码

    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;
        }

    }

Interlocked.Increment以原子的方式进行递增操作，这样我们就可知道当前每秒的新客户连接数和收发数据数了。定期执行DrawDisplay方法，显示数据信息。

测试用客户端

　　客户端的设计和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。

UdpSocketAsyncEventArgs.rar