马上注册,结交更多好友,享用更多功能,让你轻松玩转社区。

您需要 登录 才可以下载或查看,没有帐号?立即注册

x


  关键代码解析:

  1  //开始服务
  2  bool __cdecl CQueueService::StartService()
  3  {
  4    //效验参数
  5    ASSERT(m_bService==false);
  6    ASSERT(m_hCompletionPort==NULL);
  7    ASSERT(m_pIQueueServiceSink!=NULL);
  8
  9    //建立完成端口
  10    //! 需要追踪一下这里的完成端口句柄
  11    //! 首先分析这里的创建,INVALID_HANDLE_VALUE表示没有关联任何文件句柄,也就是
  12    //! 说不存在某个实现某个操作完成以后系统自动给这个完成端口post一个完成消息的概念
  13    //! 然后这里用限制工作线程数是1,也就是同意时刻只会有一条线程受到完成消息
  14    m_hCompletionPort=CreateIoCompletionPort(INVALID_HANDLE_VALUE,NULL,NULL,1);
  15    if (m_hCompletionPort==NULL) throw TEXT("队列对象完成端口创建失败");
  16
  17    //启动线程
  18    if (m_QueueServiceThread.InitThread(m_hCompletionPort)==false) throw TEXT("队列对象线程初始化失败");
  19    if (m_QueueServiceThread.StartThead()==false) throw TEXT("队列对象线程启动失败");
  20
  21    //设置参数
  22    m_bService=true;
  23
  24    return true;
  25  }

  m_QueueServiceThread.InitThread(m_hCompletionPort);

  他悄悄的给完成端口句柄扔给了服务线程,偶们去看看服务线程拿这玩意干了啥,,,

  /运行函数

  bool CQueueServiceThread::RepetitionRun()

  {

  //效验参数

  ASSERT(m_hCompletionPort!=NULL);

  //变量定义

  DWORD dwThancferred=0;

  OVERLAPPED * pOverLapped=NULL;

  CQueueService * pQueueService=NULL;

  //等待完成端口

  if (GetQueuedCompletionStatus(m_hCompletionPort,&dwThancferred,(PULONG_PTR)&pQueueService,&pOverLapped,INFINITE))

  {

  //判断退出

  if (pQueueService==NULL) return false;

  //获取数据

  tagDataHead DataHead;

  bool bSuccess=pQueueService->GetData(DataHead,m_cbBuffer,sizeof(m_cbBuffer));

  ASSERT(bSuccess==true);

  //处理数据

  if (bSuccess==true) pQueueService->OnQueueServiceThread(DataHead,m_cbBuffer,DataHead.wDataSize);

  return true;

  }

  return false;

  }

  秘密在此:

  GetQueuedCompletionStatus(m_hCompletionPort,&dwThancferred,(PULONG_PTR)&pQueueService,&pOverLapped,INFINITE)

  m_hCompletionPort是通过线程参数传递进来的,dwThancferred和pQueueService是在之前CQueueService::AddToQueue()中Post过来的参数,,,

  /加入数据

  bool __cdecl CQueueService::AddToQueue(WORD wIdentifier, void * const pBuffer, WORD wDataSize)

  {

  CThreadLockHandle LockHandle(&m_ThreadLock);

  m_DataStorage.AddData(wIdentifier,pBuffer,wDataSize);

  PostQueuedCompletionStatus(m_hCompletionPort,wDataSize,(ULONG_PTR)this,NULL);

  return true;

  }

  数据经过完成端口在CQueueServiceThread绕一圈以后会回到CQueueService::OnQueueServiceThread()

  /数据消息

  void CQueueService::OnQueueServiceThread(const tagDataHead & DataHead, void * pBuffer, WORD wDataSize)

  {

  ASSERT(m_pIQueueServiceSink!=NULL);

  try

  {

  m_pIQueueServiceSink->OnQueueServiceSink(DataHead.wIdentifier,pBuffer,DataHead.wDataSize,DataHead.dwInsertTime);

  }

  catch (...) {}

  return;

  }

  这样队列服务就完成了他的使命,将执行的消息,异步的交给指定接口去处理,,,

  稍后将对列队列服务做个宏观上的的分析,,,下班时间到,,,回去继续,,,

  用一句话来描述QueueService模块就是:

  将队列的插入和处理通过关联到完成端口上以实现在IOCP线程池管理下异步处理;

  Add以后投递完成消息,在完成端口上监听的线程受到消息以后读取数据并处理;

分享到 :
0 人收藏
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

Powered by 狐霸源码论坛 X3.3 © 2001-2018