首页 > 网站 > WEB开发 > 正文

C#编写高性能网络服务器(源码)

2024-04-27 15:14:50
字体:
来源:转载
供稿:网友

C#编写高性能网络服务器(源码)http://www.cnblogs.com/zhoutk/archive/2012/12/26/2833495.html

最近有项目要做一个高性能网络服务器,决定下功夫搞定完成端口(IOCP),最终花了一个星期终于把它弄清楚了,并用C++写了一个版本,效率很不错。

但,从项目的总体需求来考虑,最终决定上.net平台,因此又花了一天一夜弄出了一个C#版,在这与大家分享。

一些心得体会:

1、在C#中,不用去面对完成端口的操作系统内核对象,Microsoft已经为我们提供了SocketAsyncEventArgs类,它封装了IOCP的使用。请参考:http://msdn.microsoft.com/zh-cn/library/system.net.sockets.socketasynceventargs.aspx?cs-save-lang=1&cs-lang=cpp#code-snippet-1。

2、我的SocketAsyncEventArgsPool类使用List对象来存储对客户端来通信的SocketAsyncEventArgs对象,它相当于直接使用内核对象时的IoContext。我这样设计比用堆栈来实现的好处理是,我可以在SocketAsyncEventArgsPool池中找到任何一个与服务器连接的客户,主动向它发信息。而用堆栈来实现的话,要主动给客户发信息,则还要设计一个结构来存储已连接上服务器的客户。

3、对每一个客户端不管还发送还是接收,我使用同一个SocketAsyncEventArgs对象,对每一个客户端来说,通信是同步进行的,也就是说服务器高度保证同一个客户连接上要么在投递发送请求,并等待;或者是在投递接收请求,等待中。本例只做echo服务器,还未考虑由服务器主动向客户发送信息。

4、SocketAsyncEventArgs的UserToken被直接设定为被接受的客户端Socket。

5、没有使用BufferManager 类,因为我在初始化时给每一个SocketAsyncEventArgsPool中的对象分配一个缓冲区,发送时使用Arrary.Copy来进行字符拷贝,不去改变缓冲区的位置,只改变使用的长度,因此在下次投递接收请求时恢复缓冲区长度就可以了!如果要主动给客户发信息的话,可以new一个SocketAsyncEventArgs对象,或者在初始化中建立几个来专门用于主动发送信息,因为这种需求一般是进行信息群发,建立一个对象可以用于很多次信息发送,总体来看,这种花销不大,还减去了字符拷贝和消耗。

6、测试结果:(在我的笔记本上时行的,我的本本是T420 I7 8G内存)

100客户 100,000(十万次)不间断的发送接收数据(发送和接收之间没有Sleep,就一个一循环,不断的发送与接收)耗时3004.6325 秒完成总共 10,000,000 一千万次访问平均每分完成 199,691.6 次发送与接收平均每秒完成 3,328.2 次发送与接收

整个运行过程中,内存消耗在开始两三分种后就保持稳定不再增涨。

看了一下对每个客户端的延迟最多不超过2毫秒,CPU占用在8%左右。

7、下载地址:http://download.csdn.net/detail/ztk12/4928644

8、源码:

IoContextPool.csusing System;using System.Collections.Generic;using System.Text;using System.Net.Sockets;namespace IocpServer{    /// <summary>    /// 与每个客户Socket相关联,进行Send和Receive投递时所需要的参数    /// </summary>    internal sealed class IoContextPool    {        List<SocketAsyncEventArgs> pool;        //为每一个Socket客户端分配一个SocketAsyncEventArgs,用一个List管理,在程序启动时建立。        Int32 capacity;                         //pool对象池的容量        Int32 boundary;                         //已分配和未分配对象的边界,大的是已经分配的,小的是未分配的        internal IoContextPool(Int32 capacity)        {            this.pool = new List<SocketAsyncEventArgs>(capacity);            this.boundary = 0;            this.capacity = capacity;        }        /// <summary>        /// 往pool对象池中增加新建立的对象,因为这个程序在启动时会建立好所有对象,        /// 故这个方法只在初始化时会被调用,因此,没有加锁。        /// </summary>        /// <param name="arg"></param>        /// <returns></returns>        internal bool Add(SocketAsyncEventArgs arg)        {            if (arg != null && pool.Count < capacity)            {                pool.Add(arg);                boundary++;                return true;            }            else                return false;        }        /// <summary>        /// 取出集合中指定对象,内部使用        /// </summary>        /// <param name="index"></param>        /// <returns></returns>        //internal SocketAsyncEventArgs Get(int index)        //{        //    if (index >= 0 && index < capacity)        //        return pool[index];        //    else        //        return null;        //}        /// <summary>        /// 从对象池中取出一个对象,交给一个socket来进行投递请求操作        /// </summary>        /// <returns></returns>        internal SocketAsyncEventArgs Pop()        {            lock (this.pool)            {                if (boundary > 0)                {                    --boundary;                    return pool[boundary];                }                else                    return null;            }        }        /// <summary>        /// 一个socket客户断开,与其相关的IoContext被释放,重新投入Pool中,备用。        /// </summary>        /// <param name="arg"></param>        /// <returns></returns>        internal bool Push(SocketAsyncEventArgs arg)        {            if (arg != null)            {                lock (this.pool)                {                    int index = this.pool.IndexOf(arg, boundary);  //找出被断开的客户,此处一定能查到,因此index不可能为-1,必定要大于0。                    if (index == boundary)         //正好是边界元素                        boundary++;                    else                    {                        this.pool[index] = this.pool[boundary];     //将断开客户移到边界上,边界右移                        this.pool[boundary++] = arg;                    }                }                return true;            }            else                return false;        }    }}

IoServer.csusing System;using System.Collections.Generic;using System.Text;using System.Net.Sockets;using System.Threading;using System.Net;namespace IocpServer{    /// <summary>    /// 基于SocketAsyncEventArgs 实现 IOCP 服务器    /// </summary>    internal sealed class IoServer    {        /// <summary>        /// 监听Socket,用于接受客户端的连接请求        /// </summary>        PRivate Socket listenSocket;        /// <summary>        /// 用于服务器执行的互斥同步对象        /// </summary>        private static Mutex mutex = new Mutex();        /// <summary>        /// 用于每个I/O Socket操作的缓冲区大小        /// </summary>        private Int32 bufferSize;        /// <summary>        /// 服务器上连接的客户端总数        /// </summary>        private Int32 numConnectedSockets;        /// <summary>        /// 服务器能接受的最大连接数量        /// </summary>        private Int32 numConnections;        /// <summary>        /// 完成端口上进行投递所用的IoContext对象池        /// </summary>        private IoContextPool ioContextPool;        public MainForm mainForm;        /// <summary>        /// 构造函数,建立一个未初始化的服务器实例        /// </summary>        /// <param name="numConnections">服务器的最大连接数据</param>        /// <param name="bufferSize"></param>        internal IoServer(Int32 numConnections, Int32 bufferSize)        {            this.numConnectedSockets = 0;            this.numConnections = numConnections;            this.bufferSize = bufferSize;            this.ioContextPool = new IoContextPool(numConnections);            // 为IoContextPool预分配SocketAsyncEventArgs对象            for (Int32 i = 0; i < this.numConnections; i++)            {                SocketAsyncEventArgs ioContext = new SocketAsyncEventArgs();                ioContext.Completed += new EventHandler<SocketAsyncEventArgs>(OnIOCompleted);                ioContext.SetBuffer(new Byte[this.bufferSize], 0, this.bufferSize);                // 将预分配的对象加入SocketAsyncEventArgs对象池中                this.ioContextPool.Add(ioContext);            }        }        /// <summary>        /// 当Socket上的发送或接收请求被完成时,调用此函数        /// </summary>        /// <param name="sender">激发事件的对象</param>        /// <param name="e">与发送或接收完成操作相关联的SocketAsyncEventArg对象</param>        private void OnIOCompleted(object sender, SocketAsyncEventArgs e)        {            // Determine which type of Operation just completed and call the associated handler.            switch (e.LastOperation)            {                case SocketAsyncOperation.Receive:                    this.ProcessReceive(e);                    break;                case SocketAsyncOperation.Send:                    this.ProcessSend(e);                    break;                default:                    throw new ArgumentException("The last operation completed on the socket was not a receive or send");            }        }        /// <summary>        ///接收完成时处理函数        /// </summary>        /// <param name="e">与接收完成操作相关联的SocketAsyncEventArg对象</param>        private void ProcessReceive(SocketAsyncEventArgs e)        {            // 检查远程主机是否关闭连接            if (e.BytesTransferred > 0)            {                if (e.SocketError == SocketError.Success)                {                    Socket s = (Socket)e.UserToken;                    //判断所有需接收的数据是否已经完成                    if (s.Available == 0)                    {                        // 设置发送数据                        Array.Copy(e.Buffer, 0, e.Buffer, e.BytesTransferred, e.BytesTransferred);                        e.SetBuffer(e.Offset, e.BytesTransferred * 2);                        if (!s.SendAsync(e))        //投递发送请求,这个函数有可能同步发送出去,这时返回false,并且不会引发SocketAsyncEventArgs.Completed事件                        {                            // 同步发送时处理发送完成事件                            this.ProcessSend(e);                        }                    }                    else if (!s.ReceiveAsync(e))    //为接收下一段数据,投递接收请求,这个函数有可能同步完成,这时返回false,并且不会引发SocketAsyncEventArgs.Completed事件                    {                        // 同步接收时处理接收完成事件                        this.ProcessReceive(e);                    }                }                else                {                    this.ProcessError(e);                }            }            else            {                this.CloseClientSocket(e);            }        }        /// <summary>        /// 发送完成时处理函数        /// </summary>        /// <param name="e">与发送完成操作相关联的SocketAsyncEventArg对象</param>        private void ProcessSend(SocketAsyncEventArgs e)        {            if (e.SocketError == SocketError.Success)            {                Socket s = (Socket)e.UserToken;                //接收时根据接收的字节数收缩了缓冲区的大小,因此投递接收请求时,恢复缓冲区大小                e.SetBuffer(0, bufferSize);                if (!s.ReceiveAsync(e))     //投递接收请求                {                    // 同步接收时处理接收完成事件                    this.ProcessReceive(e);                }            }            else            {                this.ProcessError(e);            }        }        /// <summary>        /// 处理socket错误        /// </summary>        /// <param name="e"></param>        private void ProcessError(SocketAsyncEventArgs e)        {            Socket s = e.UserToken as Socket;            IPEndPoint localEp = s.LocalEndPoint as IPEndPoint;            this.CloseClientSocket(s, e);            string outStr = String.Format("套接字错误 {0}, IP {1}, 操作 {2}。", (Int32)e.SocketError, localEp, e.LastOperation);            mainForm.Invoke(mainForm.setlistboxcallback, outStr);            //Console.WriteLine("Socket error {0} on endpoint {1} during {2}.", (Int32)e.SocketError, localEp, e.LastOperation);        }        /// <summary>        /// 关闭socket连接        /// </summary>        /// <param name="e">SocketAsyncEventArg associated with the completed send/receive operation.</param>        private void CloseClientSocket(SocketAsyncEventArgs e)        {            Socket s = e.UserToken as Socket;            this.CloseClientSocket(s, e);        }        private void CloseClientSocket(Socket s, SocketAsyncEventArgs e)        {            Interlocked.Decrement(ref this.numConnectedSockets);            // SocketAsyncEventArg 对象被释放,压入可重用队列。            this.ioContextPool.Push(e);                        string outStr = String.Format("客户 {0} 断开, 共有 {1} 个连接。", s.RemoteEndPoint.ToString(), this.numConnectedSockets);            mainForm.Invoke(mainForm.setlistboxcallback, outStr);                        //Console.WriteLine("A client has been disconnected from the server. There are {0} clients connected to the server", this.numConnectedSockets);            try            {                s.Shutdown(SocketShutdown.Send);            }            catch (Exception)            {                // Throw if client has closed, so it is not necessary to catch.            }            finally            {                s.Close();            }        }        /// <summary>        /// accept 操作完成时回调函数        /// </summary>        /// <param name="sender">Object who raised the event.</param>        /// <param name="e">SocketAsyncEventArg associated with the completed accept operation.</param>        private void OnAcceptCompleted(object sender, SocketAsyncEventArgs e)        {            this.ProcessAccept(e);        }        /// <summary>        /// 监听Socket接受处理        /// </summary>        /// <param name="e">SocketAsyncEventArg associated with the completed accept operation.</param>        private void ProcessAccept(SocketAsyncEventArgs e)        {            Socket s = e.AcceptSocket;            if (s.Connected)            {                try                {                    SocketAsyncEventArgs ioContext = this.ioContextPool.Pop();                    if (ioContext != null)                    {                        // 从接受的客户端连接中取数据配置ioContext                        ioContext.UserToken = s;                        Interlocked.Increment(ref this.numConnectedSockets);                        string outStr = String.Format("客户 {0} 连入, 共有 {1} 个连接。",  s.RemoteEndPoint.ToString(),this.numConnectedSockets);                        mainForm.Invoke(mainForm.setlistboxcallback,outStr);                        //Console.WriteLine("Client connection accepted. There are {0} clients connected to the server",                            //this.numConnectedSockets);                        if (!s.ReceiveAsync(ioContext))                        {                            this.ProcessReceive(ioContext);                        }                    }                    else        //已经达到最大客户连接数量,在这接受连接,发送“连接已经达到最大数”,然后断开连接                    {                        s.Send(Encoding.Default.GetBytes("连接已经达到最大数!"));                        string outStr = String.Format("连接已满,拒绝 {0} 的连接。", s.RemoteEndPoint);                        mainForm.Invoke(mainForm.setlistboxcallback, outStr);                        s.Close();                   }                }                catch (SocketException ex)                {                    Socket token = e.UserToken as Socket;                    string outStr = String.Format("接收客户 {0} 数据出错, 异常信息: {1} 。", token.RemoteEndPoint, ex.ToString());                    mainForm.Invoke(mainForm.setlistboxcallback, outStr);                    //Console.WriteLine("Error when processing data received from {0}:/r/n{1}", token.RemoteEndPoint, ex.ToString());                }                catch (Exception ex)                {                    mainForm.Invoke(mainForm.setlistboxcallback, "异常:" + ex.ToString());                }                // 投递下一个接受请求                this.StartAccept(e);            }        }        /// <summary>        /// 从客户端开始接受一个连接操作        /// </summary>        /// <param name="acceptEventArg">The context object to use when issuing         /// the accept operation on the server's listening socket.</param>        private void StartAccept(SocketAsyncEventArgs acceptEventArg)        {            if (acceptEventArg == null)            {                acceptEventArg = new SocketAsyncEventArgs();                acceptEventArg.Completed += new EventHandler<SocketAsyncEventArgs>(OnAcceptCompleted);            }            else            {                // 重用前进行对象清理                acceptEventArg.AcceptSocket = null;            }            if (!this.listenSocket.AcceptAsync(acceptEventArg))            {                this.ProcessAccept(acceptEventArg);            }        }        /// <summary>        /// 启动服务,开始监听        /// </summary>        /// <param name="port">Port where the server will listen for connection requests.</param>        internal void Start(Int32 port)        {            // 获得主机相关信息            IPAddress[] addressList = Dns.GetHostEntry(Environment.MachineName).AddressList;            IPEndPoint localEndPoint = new IPEndPoint(addressList[addressList.Length - 1], port);            // 创建监听socket            this.listenSocket = new Socket(localEndPoint.AddressFamily, SocketType.Stream, ProtocolType.Tcp);            this.listenSocket.ReceiveBufferSize = this.bufferSize;            this.listenSocket.SendBufferSize = this.bufferSize;            if (localEndPoint.AddressFamily == AddressFamily.InterNetworkV6)            {                // 配置监听socket为 dual-mode (IPv4 & IPv6)                 // 27 is equivalent to IPV6_V6ONLY socket option in the winsock snippet below,                this.listenSocket.SetSocketOption(SocketOptionLevel.IPv6, (SocketOptionName)27, false);                this.listenSocket.Bind(new IPEndPoint(IPAddress.IPv6Any, localEndPoint.Port));            }            else            {                this.listenSocket.Bind(localEndPoint);            }            // 开始监听            this.listenSocket.Listen(this.numConnections);            // 在监听Socket上投递一个接受请求。            this.StartAccept(null);            // Blocks the current thread to receive incoming messages.            mutex.WaitOne();        }        /// <summary>        /// 停止服务        /// </summary>        internal void Stop()        {            this.listenSocket.Close();            mutex.ReleaseMutex();        }    }}


发表评论 共有条评论
用户名: 密码:
验证码: 匿名发表