基于完成端口的文件传输设计

完成端口

说到完成端口,我想很多人都不太陌生,下面是一段摘录:
“完成端口”模型是迄今为止最为复杂的一种I/O模型。然而，假若一个应用程序同时需要管理为数众多的套接字，那么采用这种模型，往往可以达到最佳的系统性能！但不幸的是，该模型只适用于Windows NT和Windows 2000操作系统。因其设计的复杂性，只有在你的应用程序需要同时管理数百乃至上千个套接字的时候，而且希望随着系统内安装的CPU数量的增多，应用程序的性能也可以线性提升，才应考虑采用“完成端口”模型。要记住的一个基本准则是，假如要为Windows NT或Windows 2000开发高性能的服务器应用，同时希望为大量套接字I/O请求提供服务（Web服务器便是这方面的典型例子），那么I/O完成端口模型便是最佳选择！（节选自《Windows网络编程》第八章）

在.net中,一旦说到完成端口,我们就不得不提到SocketAsyncEventArgs这个高性能的类,其内部是基于完成端口实现的,然后被微软提供了诸多封装,所以使用起来也比较简单一些.由于这个类利用完成端口并且结合异步事件的方式进行设计的,所以我们可以大致的知道他的一些特点.

处理流程

关于此类,微软的解释很多,但是始终离不开高性能三个字.根据我的理解,如果利用此类设计一个服务端的消息传送中心,那么其运作流程如下:

１.创建SocketAsyncEventArgs池,并且创建缓冲管理中心,负责对池中的对象分配缓冲大小.

２.创建服务器套接字,并处于监听状态.同时创建基于SocketAsyncEventArgs的客户端接收对象,接收客户端的连接.

３.如果有客户端连接, 客户端接收对象将会把控制权移交给数据接收对象,数据接收对象开始接收数据.

这只是一个简单的流程,从这里我们可以看出,服务端套接字只是负责监听,一旦有客户端连接,就会把连接事件抛给接收对象;所以客户端一个一个的来,服务端一个一个的抛,性能自然会好了不少.

说了这么多,我们来浅析一下SocketAsyncEventArgs对象中的一些基本的知识点.

缓冲

首先,说到缓冲,我们很容易理解为一个存储池,里面可以放入东西,也可以拿走.在SocketAsyncEventArgs类中,我们可以利用其SetBuffer方法初始化缓冲区.

比如说:　receiveArgs.SetBuffer(new byte[10],0,5);他的意思就是为当前接收数据对象设置的缓冲区大小为10字节,位置从0开始,并且允许接收的最大的数据长度为5字节.

当数据被接收,然后写入到缓冲区的时候,数据会按照预先设定好的缓冲规则进行放置.比如这里我输入aaaaa,那么在缓冲区会放入如下数据:

 

但是当有新的数据再进来的时候,比如这里我输入了bbbb,那么缓冲区就变成了

看到了没有,缓冲区中第五位依然是我们前一次插入的值,所以,在这里我提醒大家,缓冲区的东西取完之后,一定要重置一下,否则脏数据会导致数据错误.

讲到这里,也许有人会说,假如我插入aaaaab会怎么样,其实,这个长度已经超过了缓冲区的长度,缓冲区将会做截断处理,然后当作两段放入缓存中,首先会是

,然后会是

 

所以如果这个时候你的数据没有被及时取走的话,将会得到最终结果　

 这不,已经发生粘包现象了.

 抛出方式

说完缓冲,这里我们需要说到的是服务端套接字是如何将接收的客户端通过事件的方式抛给SocketAsyncEventArgs对象的.

在进行服务端,我们一般都会有一个用于监听的套接字,套接字会利用serverSocket.AcceptAsync(SocketAsyncEventArgs)异步方法注册SocketAsyncEventArgs对象,然后一旦有客户端连接,就会激发SocketAsyncEventArgs对象的Completed事件,然后,在这个事件中,serverSocket会通过ReceiveAsync(SocketAsyncEventArgs)异步方法将数据处理的过程交给另外一个专门负责数据处理的对象去完成,这样,客户端连接,服务端只负责将连接事件抛给SocketAsyncEventArgs对象即可,比起传统的编程方式:　服务端既负责监听,又负责接收,　效率极大的提升.

同步方式

最后说到的是同步问题,因为在异步交互的系统中,同步问题确实很重要,尤其是当客户端和服务器同步的时候.

客户端里面的线程同步,我们可以利用AutoResetEvent来实现,客户端和服务端同步的时候,我们需要AutoResetEvent并且结合一些标记信息来进行,也就是客户端往服务器发送的一些小标记,比如１代表可以进行,０代表取消等等.

在设计的时候,我们还需要记住的是,代码的核心只是负责数据的传输,不要加过多的逻辑判断在里面,否则会影响性能,逻辑判断等最好移到界面处理处来进行.

代码精解

好了,下面上代码来解释解释.

首先我们先从服务端开始:

  private int listenClientCount;
        private int poolSize;
        private int bufferSize;

        private EndPoint endPoint;
        private Socket serverSocket;
        private SocketAsyncEventArgsPool readWritePool;
        private BufferManager bufferManager;
        private SocketAsyncEventArgs acceptArgs;

        private SocketAsyncEventArgs receiveArgs;
        private SocketAsyncEventArgs sendArgs;

        public Action<Socket> OnStarted;
        public Action<Socket> OnConnected;
        public Action<SocketAsyncEventArgs> OnReceive;
        public Action<SocketAsyncEventArgs> OnSent;
        public Action<SocketAsyncEventArgs> OnDisconnect;
        public Action OnStopped;

在服务端,我们会定义服务端套接字,用于连接对象的acceptArgs,用于接收数据的receiveArgs以及用于抛出事件的一些Action等等,这些Action让外部引用进行注册并进行操纵.

然后这里是Start函数,

public void Start()
        {
            if (this.serverSocket != null)
                throw new Exception("Server is running, can't init another one.");

            this.serverSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
            try
            {
                this.serverSocket.Bind(endPoint);
                this.serverSocket.Listen(listenClientCount);
            }
            catch (ArgumentNullException ex)
            {
                throw new Exception(ex.Message);
            }
            catch (SocketException ex)
            {
                throw new Exception(ex.Message);
            }

            if (OnStarted != null)
                OnStarted(serverSocket);

            StartAccept(null);
        }

这个函数的作用就是开启监听,并且将接收事件传递给acceptArgs对象.
然后,当客户端连接过来的时候,服务端套接字开始将控制权传递给receiveArgs对象:

 private void ProcessAccept(SocketAsyncEventArgs e)
        {
            if (OnConnected != null)
            {
                OnConnected(e.AcceptSocket);
            }
            receiveArgs = this.readWritePool.Pop();
            this.bufferManager.SetBuffer(receiveArgs);
            receiveArgs.AcceptSocket = e.AcceptSocket;
            receiveArgs.Completed += IO_Completed;

            if (!e.AcceptSocket.ReceiveAsync(receiveArgs))
            {
                ProcessReceive(receiveArgs);
            }

            StartAccept(e);
        }

注意,这里有一个IO_Completed事件,表示一旦当前数据的数据接收完成,receiveArgs对象将会触发该事件,用于进行下一步操作,一般会取lastOperation来判断是否继续接收.
最后需要说的就是真正的数据接收函数:

        private void ProcessReceive(SocketAsyncEventArgs e)
        {
            if (e.BytesTransferred > 0 && e.SocketError == SocketError.Success)
            {
                if (this.OnReceive != null)
                    this.OnReceive(e);

                if (!e.AcceptSocket.ReceiveAsync(e))
                {
                    this.ProcessReceive(e);
                }
            }
            else
            {
                CloseClient(e);
            }
        }

这个函数利用了循环的方式来接收数据,以便保证所有的数据都能够进来,直至数据接收完成.在这个接收函数中,不应该有任何的逻辑判断,否则会比较影响吞吐性能.

测试结果

好了,下面就看下界面截图吧(请注意,这里我得客户端是在上海,服务器端在香港,真实模拟数据传输情况).

(图１,这是服务端开启时候的情景)

(图２,这是客户端开启时候的情景)

(图３,客户端准备发送文件给服务器端的情景,请注意这里的同步方式,客户端会等待服务端准备完成,才会开始传送数据)

(图４,数据传送中,请看,左边是客户端的缓冲区的缓冲数据,右边是服务器端的接收缓冲区的缓冲数据)

(图５,数据传输完毕的情景)

(图６,请注意,即使我们服务端设置的接收缓冲区为327670,长度为32767,但是有时候数据还是不会完整的到达的,遇到这种情况,我们只需等待下一批数据到达即可.)

源码下载

点击这里下载