C++線程池的實現代碼

//說明， 這段代碼我用了好久， 我刪除了自動調整規模的代碼(由於他還不成熟)
/******************************************************************
*  Thread Pool For Win32 
*  VC++ 6, BC++ 5.5(Free), GCC(Free)
*  Update : 2004.6.9 llBird  wushaojian@21cn.com
Use:
1):
void threadfunc(void *p)
{
 //...
}
 ThreadPool tp;
 for(i=0; i<100; i++)
  tp.Call(threadfunc);
 ThreadPool tp(20);//20爲初始線程池規模
 tp.Call(threadfunc, lpPara);
 tp.AdjustSize(50);//增長50
 tp.AdjustSize(-30);//減小30

2):
class MyThreadJob : public ThreadJob //線程對象從ThreadJob擴展
{
public:
 virtual void DoJob(void *p)//自定義的虛函數
 {
  //....
 }
};
 MyThreadJob mt[10];
 ThreadPool tp;
 for(i=0; i<100 i++)
  tp.Call(mt + i);//tp.Call(mt + i, para);
*******************************************************************/
#ifndef _ThreadPool_H_
#define _ThreadPool_H_
#pragma warning(disable: 4530)
#pragma warning(disable: 4786)
#include <cassert>
#include <vector>
#include <queue>
#include <windows.h>

class ThreadJob  //工做基類
{
public:
 //供線程池調用的虛函數
 virtual void DoJob(void *pPara) = 0;
};
class ThreadPool
{
public:
 //dwNum 線程池規模
 ThreadPool(DWORD dwNum = 4) : _lThreadNum(0), _lRunningNum(0) 
 {
  InitializeCriticalSection(&_csThreadVector);
  InitializeCriticalSection(&_csWorkQueue);
  _EventComplete = CreateEvent(0, false, false, NULL);
  _EventEnd = CreateEvent(0, true, false, NULL);
  _SemaphoreCall = CreateSemaphore(0, 0,  0x7FFFFFFF, NULL);
  _SemaphoreDel =  CreateSemaphore(0, 0,  0x7FFFFFFF, NULL);
  assert(_SemaphoreCall != INVALID_HANDLE_VALUE);
  assert(_EventComplete != INVALID_HANDLE_VALUE);
  assert(_EventEnd != INVALID_HANDLE_VALUE);
  assert(_SemaphoreDel != INVALID_HANDLE_VALUE);
  AdjustSize(dwNum <= 0 ? 4 : dwNum);
 }
 ~ThreadPool()
 {
  DeleteCriticalSection(&_csWorkQueue);
  CloseHandle(_EventEnd);
  CloseHandle(_EventComplete);
  CloseHandle(_SemaphoreCall);
  CloseHandle(_SemaphoreDel);
  
  vector<ThreadItem*>::iterator iter;
  for(iter = _ThreadVector.begin(); iter != _ThreadVector.end(); iter++)
  {
   if(*iter)
    delete *iter;
  }
  DeleteCriticalSection(&_csThreadVector);
 }
 //調整線程池規模
 int AdjustSize(int iNum)
 {
  if(iNum > 0)
  {
   ThreadItem *pNew;
   EnterCriticalSection(&_csThreadVector);
   for(int _i=0; _i<iNum; _i++)
   {
    _ThreadVector.push_back(pNew = new ThreadItem(this)); 
    assert(pNew);
    pNew->_Handle = CreateThread(NULL, 0, DefaultJobProc, pNew, 0, NULL);
    assert(pNew->_Handle);
   }
   LeaveCriticalSection(&_csThreadVector);
  }
  else
  {
   iNum *= -1;
   ReleaseSemaphore(_SemaphoreDel,  iNum > _lThreadNum ? _lThreadNum : iNum, NULL);
  }
  return (int)_lThreadNum;
 }
 //調用線程池
 void Call(void (*pFunc)(void  *), void *pPara = NULL)
 {
  assert(pFunc);
  EnterCriticalSection(&_csWorkQueue);
  _JobQueue.push(new JobItem(pFunc, pPara));
  LeaveCriticalSection(&_csWorkQueue);
  ReleaseSemaphore(_SemaphoreCall, 1, NULL);
 }
 //調用線程池
 inline void Call(ThreadJob * p, void *pPara = NULL)
 {
  Call(CallProc, new CallProcPara(p, pPara));
 }
 //結束線程池, 並同步等待
 bool EndAndWait(DWORD dwWaitTime = INFINITE)
 {
  SetEvent(_EventEnd);
  return WaitForSingleObject(_EventComplete, dwWaitTime) == WAIT_OBJECT_0;
 }
 //結束線程池
 inline void End()
 {
  SetEvent(_EventEnd);
 }
 inline DWORD Size()
 {
  return (DWORD)_lThreadNum;
 }
 inline DWORD GetRunningSize()
 {
  return (DWORD)_lRunningNum;
 }
 bool IsRunning()
 {
  return _lRunningNum > 0;
 }
protected:
 //工做線程
 static DWORD WINAPI DefaultJobProc(LPVOID lpParameter = NULL)
 {
  ThreadItem *pThread = static_cast<ThreadItem*>(lpParameter);
  assert(pThread);
  ThreadPool *pThreadPoolObj = pThread->_pThis;
  assert(pThreadPoolObj);
  InterlockedIncrement(&pThreadPoolObj->_lThreadNum);
  HANDLE hWaitHandle[3];
  hWaitHandle[0] = pThreadPoolObj->_SemaphoreCall;
  hWaitHandle[1] = pThreadPoolObj->_SemaphoreDel;
  hWaitHandle[2] = pThreadPoolObj->_EventEnd;
  JobItem *pJob;
  bool fHasJob;
  
  for(;;)
  {
   DWORD wr = WaitForMultipleObjects(3, hWaitHandle, false, INFINITE);
   //響應刪除線程信號
   if(wr == WAIT_OBJECT_0 + 1)  
    break;
   
   //從隊列裏取得用戶做業
   EnterCriticalSection(&pThreadPoolObj->_csWorkQueue);
   if(fHasJob = !pThreadPoolObj->_JobQueue.empty())
   {
    pJob = pThreadPoolObj->_JobQueue.front();
    pThreadPoolObj->_JobQueue.pop();
    assert(pJob);
   }
   LeaveCriticalSection(&pThreadPoolObj->_csWorkQueue);
   //受到結束線程信號 肯定是否結束線程(結束線程信號 && 是否還有工做)
   if(wr == WAIT_OBJECT_0 + 2 && !fHasJob)  
    break;
   if(fHasJob && pJob)
   {
    InterlockedIncrement(&pThreadPoolObj->_lRunningNum);
    pThread->_dwLastBeginTime = GetTickCount();
    pThread->_dwCount++;
    pThread->_fIsRunning = true;
    pJob->_pFunc(pJob->_pPara); //運行用戶做業
    delete pJob; 
    pThread->_fIsRunning = false;
    InterlockedDecrement(&pThreadPoolObj->_lRunningNum);
   }
  }
  //刪除自身結構
  EnterCriticalSection(&pThreadPoolObj->_csThreadVector);
  pThreadPoolObj->_ThreadVector.erase(find(pThreadPoolObj->_ThreadVector.begin(), pThreadPoolObj->_ThreadVector.end(), pThread));
  LeaveCriticalSection(&pThreadPoolObj->_csThreadVector);
  delete pThread;
  InterlockedDecrement(&pThreadPoolObj->_lThreadNum);
  if(!pThreadPoolObj->_lThreadNum)  //全部線程結束
   SetEvent(pThreadPoolObj->_EventComplete);
  return 0;
 }
 //調用用戶對象虛函數
 static void CallProc(void *pPara) 
 {
  CallProcPara *cp = static_cast<CallProcPara *>(pPara);
  assert(cp);
  if(cp)
  {
   cp->_pObj->DoJob(cp->_pPara);
   delete cp;
  }
 }
 //用戶對象結構
 struct CallProcPara  
 {
  ThreadJob* _pObj;//用戶對象 
  void *_pPara;//用戶參數
  CallProcPara(ThreadJob* p, void *pPara) : _pObj(p), _pPara(pPara) { };
 };
 //用戶函數結構
 struct JobItem 
 {
  void (*_pFunc)(void  *);//函數
  void *_pPara; //參數
  JobItem(void (*pFunc)(void  *) = NULL, void *pPara = NULL) : _pFunc(pFunc), _pPara(pPara) { };
 };
 //線程池中的線程結構
 struct ThreadItem
 {
  HANDLE _Handle; //線程句柄
  ThreadPool *_pThis;  //線程池的指針
  DWORD _dwLastBeginTime; //最後一次運行開始時間
  DWORD _dwCount; //運行次數
  bool _fIsRunning;
  ThreadItem(ThreadPool *pthis) : _pThis(pthis), _Handle(NULL), _dwLastBeginTime(0), _dwCount(0), _fIsRunning(false) { };
  ~ThreadItem()
  {
   if(_Handle)
   {
    CloseHandle(_Handle);
    _Handle = NULL;
   }
  }
 };
 
 std::queue<JobItem *> _JobQueue;  //工做隊列
 std::vector<ThreadItem *>  _ThreadVector; //線程數據
 CRITICAL_SECTION _csThreadVector, _csWorkQueue; //工做隊列臨界, 線程數據臨界
 HANDLE _EventEnd, _EventComplete, _SemaphoreCall, _SemaphoreDel;//結束通知, 完成事件, 工做信號， 刪除線程信號
 long _lThreadNum, _lRunningNum; //線程數, 運行的線程數
};
#endif //_ThreadPool_H_

　　

轉載自 http://blog.csdn.net/pjchen/archive/2004/11/06/170606.aspx

基本上是拿來就用了，對WIN32 API不熟，但對線程池的邏輯仍是比較熟的，認爲這個線程池寫得很清晰，我拿來用在一個多線程下載的模塊中。很實用的東東。

調用方法

void threadfunc(void *p)

{

     YourClass* yourObject = (YourClass*)    p;

 //...
}
 ThreadPool tp;
 for(i=0; i<100; i++)
  tp.Call(threadfunc);

ThreadPool tp(20);//20爲初始線程池規模

 tp.Call(threadfunc, lpPara);
     

使用時注意幾點：

1. ThreadJob  沒什麼用，直接寫線程函數吧。 

2. 線程函數（threadfunc）的入口參數void* 能夠轉成自定義的類型對象，這個對象能夠記錄下線程運行中的數據，並設置線程當前狀態，以此與線程進行交互。

3. 線程池有一個EndAndWait函數，用於讓線程池中全部計算正常結束。有時線程池中的一個線程可能要運行很長時間，怎麼辦？能夠經過線程函數threadfunc的入口參數對象來處理，好比：

class YourClass {

  int cmd; // cmd = 1是上線程中止計算，正常退出。

};

threadfunc(void* p) {

  YourClass* yourObject = (YourClass*)p;

  while (true) {

    // do some calculation

    if (yourClass->cmd == 1)

      break;

  }

}

在主線程中設置yourClass->cmd = 1，該線程就會天然結束。

很簡潔通用的線程池實現。