当前位置:编程学习 > C#/ASP.NET >>

C# 完成端口组件实现精髓

前面用C++实现了windows平台上的网络完成端口组件,那么到C#中如何做了?起初我打算通过PInvoke来调用win底层API来仿照C++实现,但问题很快就出来了--C#中的Unsafe指针无法稳定的指向一块缓冲区的首地址,也就是说当垃圾回收进行的时候,我们的unsafe指针的值可能已经无效了。用pin?我也想过,可是锁住所有的TCP接收缓冲区,会极大的降低运行时的效率。难道没有办法了吗?想想完成端口模型的本质思想是将"启动异步操作的线程"和"提供服务的线程"(即工作者线程)拆伙。做到这一点不就够了吗,这是本质的东西。

        分析一下我们需要几种类型的线程,首先我们需要一个线程来接收TCP连接请求,这就是所谓监听线程,当成功的接收到一个连接后,就向连接发送一个异步接收数据的请求,由于是异步操作,所以会立即返回,然后再去接收新的连接请求,如此监听线程就循环运作起来了。值得提出的是,在异步接收的回调函数中,应该对接收到的数据进行处理,完成端口模型所做的就是将接收到的数据放在了完成端口队列中,注意,是一个队列。
        第二种线程类型,就是工作者线程。工作者线程的个数有个经验值是( Cpu个数×2 + 2),当然具体取多少,还要取决于你的应用的要求。工作者线程的任务就是不断地从完成端口队列中取出数据,并处理它,然后如果有回复,再将回复写入对应的连接。

刚才已经提到,我打算在C#中不使用Pinvoke来实现完成端口,而在.NET平台中是没有特定的完成端口队列这个组件或类的,所以我们首先要实现一个这样的队列。

         好,让我们来定义接口IRequestQueueManager,用于模拟完成端口的队列。

下面给出其定义和默认实现。

using System;
using System.Collections ;

namespace EnterpriseServerBase.Network
{
 /// <summary>
 /// IRequestQueueManager 用于模拟完成端口的队列。
 /// </summary>
 public interface IRequestQueueManager :IRequestPusher
 {
  //void Push(object package) ;
  object Pop() ;
  void Clear() ;
  int Length {get ;}
 }

 //向队列中的Push一个请求包
 public interface IRequestPusher
 {
  void Push(object package) ;
 }
 
 /// <summary>
 /// IRequestQueueManager 的默认实现
 /// </summary>
 public class RequestQueueManager :IRequestQueueManager
 {
  private Queue queue = null ;

  public RequestQueueManager()
  {
   this.queue = new Queue() ;
  }

  public void Push(object package)
  {
   lock(this.queue)
   {
    this.queue.Enqueue(package) ;
   }
  }

  public object Pop()
  {
   object package = null ;

   lock(this.queue)
   {
    if(this.queue.Count > 0)
    {   
     package = this.queue.Dequeue() ;
    }
   }

   return package ;
  }

  public void Clear()
  {
   lock(this.queue)
   {
    this.queue.Clear() ;
   }
  }

  public int Length
  {
   get
   {
    return this.queue.Count ;
   }
  }

 }
}

在IRequestQueueManager的基础上,可以将工作者线程和启动异步操作的线程拆开了。由于工作者线程只与端口队列相关,所以我决定将它们一起封装起来--IIOCPManager

来看看完成端口类如何实现:

using System;
using System.Threading ;

namespace EnterpriseServerBase.Network
{
 /// <summary>
 /// IIOCPManager 完成端口管理者,主要管理工作者线程和完成端口队列。
 /// 2005.05.23
 /// </summary>
 public interface IIOCPManager : IRequestPusher
 {
  void Initialize(IOCPPackageHandler i_packageHandler ,int threadCount) ;
  void Start() ; //启动工作者线程
  void Stop() ;  //退出工作者线程

  int WorkThreadCount{get ;}

  event CallBackPackageHandled PackageHandled ;
 }

 //IOCPPackageHandler 用于处理从完成端口队列中取出的package
 public interface IOCPPackageHandler
 {
   void HandlerPackage(object package) ; //一般以同步实现
 }

 public delegate void CallBackPackageHandled(object package) ;

 /// <summary>
 /// IOCPManager 是IIOCPManager的默认实现
 /// </summary>
 public class IOCPManager :IIOCPManager
 {
  private IRequestQueueManager requestQueueMgr = null ;
  private IOCPPackageHandler packageHandler ;
  private int workThreadCount = 0 ; //实际的工作者线程数
  private int MaxThreadCount  = 0 ;

  private bool stateIsStop = true ;
 
  public IOCPManager()
  {  
  }


  #region ICPWorkThreadManager 成员
  public event CallBackPackageHandled PackageHandled ;

  public void Initialize(IOCPPackageHandler i_packageHandler ,int threadCount )
  {
   this.requestQueueMgr = new RequestQueueManager() ;
   this.MaxThreadCount = threadCount ;
   this.packageHandler = i_packageHandler ;
  }

  public void Push(object package)
  {
   this.requestQueueMgr.Push(package) ;
  }

  public void Start()
  {
   if(! this.stateIsStop)
   {
    return ;
   }

   this.stateIsStop = false ;
   this.CreateWorkThreads() ;

  }

  public void Stop()
  {
   if(this.stateIsStop)
   {
    return ;
   }

   this.stateIsStop = true ;

   //等待所有工作者线程结束
   int count = 0 ;
   while(this.workThreadCount != 0)
   {
    if(count < 10)
    {
     Thread.Sleep(200) ;
    }
    else
    {   
     throw new Exception("WorkThread Not Terminated !") ;
    }

    ++ count ;
   }

   this.requestQueueMgr.Clear() ;
  }  

  public int WorkThreadCount
  {
   get
   {
    return this.workThreadCount ;
   }
  }
  #endregion

  #region CreateWorkThreads
  private void CreateWorkThreads()
  {
   for(int i= 0 ;i< this.MaxThreadCount ;i++)
   {
    Thread t = new Thread(new ThreadStart(this.ServeOverLap)) ;
    Interlocked.Increment(ref this.workThreadCount) ;
    t.Start() ;
   }
  }
  #endregion

  #region ServeOverLap 工作者线程
  private void ServeOverLap()
  {
   while(! this.stateIsStop)
   {
    object package = this.requestQueueMgr.Pop() ;
&

补充:软件开发 , C# ,
CopyRight © 2012 站长网 编程知识问答 www.zzzyk.com All Rights Reserved
部份技术文章来自网络,