// 線程要執(zhí)行的函數(shù)int Count(int nWork){  int nResult = nWork * nWork;  printf("count result is %d/n",nResult);  return 0;}int main() {  // 創(chuàng)建線程管理類的實(shí)例，把要執(zhí)行的線程函數(shù)和最大線程數(shù)傳進(jìn)去  CThreadManager* pManager = new CThreadManager(Count, 3);  // 把要進(jìn)行計(jì)算的數(shù)放到工作隊(duì)列中  pManager->PushWorkQue(5);  pManager->PushWorkQue(20);  // 設(shè)置信號(hào)量，喚醒線程  pManager->PostSem();  pManager->PostSem();  // 等待子線程執(zhí)行  sleep(1);  return 0;}

class CThreadManager {  friend void* ManageFuction(void*);private:  sem_t m_sem;  // 信號(hào)量  pthread_mutex_t m_mutex; // 互斥鎖  queue<int> m_queWork; // 工作隊(duì)列  list<CThread*> m_lstThread; // 線程list  int (*m_threadFuction)(int); //函數(shù)指針，指向main函數(shù)傳過來的線程執(zhí)行函數(shù)public:  CThreadManager(int (*threadFuction)(int), int nMaxThreadCnt);  virtual ~CThreadManager();  int WaitSem();  int PostSem();  int LockMutex();  int UnlockMutex();  void PushWorkQue(int nWork);  int PopWorkQue();  int RunThreadFunction(int nWork);};// 線程執(zhí)行函數(shù)，它只是個(gè)殼子，處理信號(hào)量和互斥鎖等，// 最后調(diào)用main函數(shù)傳過來的線程執(zhí)行函數(shù)來實(shí)現(xiàn)業(yè)務(wù)處理void* ManageFuction(void* argv){  CThreadManager* pManager = (CThreadManager*)argv;  // 進(jìn)行無限循環(huán)（意味著線程是不銷毀的，重復(fù)利用）  while(true)  {    // 線程開啟后，就在這里阻塞著，直到main函數(shù)設(shè)置了信號(hào)量    pManager->WaitSem();    printf("thread wakeup./n");    // 從工作隊(duì)列中取出要處理的數(shù)    pManager->LockMutex();    int nWork = pManager->PopWorkQue();    pManager->UnlockMutex();    printf("call Count function./n");    pManager->RunThreadFunction(nWork);  }  return 0;}// 構(gòu)造方法CThreadManager::CThreadManager(int (*threadFuction)(int), int nMaxThreadCnt) {  sem_init(&m_sem, 0, 0);  pthread_mutex_init(&m_mutex, NULL);  m_threadFuction = threadFuction;  for(int i=0; i<nMaxThreadCnt; i++)  {    CThread* pThread = new CThread(ManageFuction, this);    printf("thread started./n");    m_lstThread.push_back(pThread);  }}

CThread::CThread(void* (*threadFuction)(void*),void* threadArgv) {  // 初始化線程屬性  pthread_attr_t threadAttr;  pthread_attr_init(&threadAttr);  pthread_create(&m_thread, &threadAttr, threadFuction, threadArgv);}

/* * 多線程管理類 *  */ #ifndef CTHREADPOOLMANAGE_H#define CTHREADPOOLMANAGE_H#include <iostream>#include <pthread.h>#include <unistd.h> #include <list>#include <vector>#include <time.h>#include <asm/errno.h> #define USLEEP_TIME 100#define CHECK_TIME 1  using namespace std;class CDoit{public: virtual int start(void *){}; virtual int end(){};};  class CthreadPoolManage{private: int _minThreads;  //最少保留幾個(gè)線程 int _maxThreads;  //最多可以有幾個(gè)線程 int _waitSec;      //空閑多少秒后將線程關(guān)閉 class threadInfo{  public:  threadInfo(){   isbusy = false;    doFlag = true;  }  //  pthread_mutex_t mtx=PTHREAD_MUTEX_INITIALIZER;  pthread_cond_t cond=PTHREAD_COND_INITIALIZER;  bool isbusy;   //是否空閑  bool doFlag;  //  time_t beginTime;     //線程不工作開始時(shí)間  pthread_t cThreadPid; //線程id  pthread_attr_t cThreadAttr; //線程屬性  CDoit * doit;         //任務(wù)類  void  * value;        //需要傳遞的值 }; //線程函數(shù) static void* startThread(void*); //任務(wù)隊(duì)列鎖 pthread_mutex_t _duty_mutex; //任務(wù)隊(duì)列 list<threadInfo*> _dutyList; //線程隊(duì)列鎖 pthread_mutex_t _thread_mutex; //線程隊(duì)列 list<threadInfo*> _threadList;  ///初始化，創(chuàng)建最小個(gè)數(shù)線程///void initThread(); ///任務(wù)分配線程///static void* taskAllocation(void*arg);pthread_t tasktPid;///線程銷毀、狀態(tài)檢查線程///static void* checkThread(void* arg);pthread_t checktPid;bool  checkrun; //線程異常退出清理static void threadCleanUp(void* arg); //int addThread(list<threadInfo*> *plist,threadInfo* ptinfo); public:CthreadPoolManage();/*保留的最少線程，最多線程數(shù)，空閑多久銷毀，保留幾個(gè)線程的冗余 */CthreadPoolManage(int min,int max,int waitSec);~CthreadPoolManage(); int start();//任務(wù)注入器int putDuty(CDoit *,void *); int getNowThreadNum(); }; #endif // CTHREADPOOLMANAGE_H

/* * 線程池，線程管理類 *  */ #include "cthreadpoolmanage.h" CthreadPoolManage::CthreadPoolManage(){ _minThreads = 5;  //最少保留幾個(gè)線程 _maxThreads = 5;  //最多可以有幾個(gè)線程 _waitSec = 10;      //空閑多少秒后將線程關(guān)閉 pthread_mutex_init(&_duty_mutex, NULL); pthread_mutex_init(&_thread_mutex, NULL); checkrun = true;}  CthreadPoolManage::CthreadPoolManage(int min, int max, int waitSec){  CthreadPoolManage();  _minThreads = min;  //最少保留幾個(gè)線程  _maxThreads = max;  //最多可以有幾個(gè)線程  _waitSec = waitSec;      //空閑多少秒后將線程關(guān)閉} CthreadPoolManage::~CthreadPoolManage(){ }void CthreadPoolManage::threadCleanUp(void* arg){ threadInfo* tinfo = (threadInfo*)arg; tinfo->isbusy = false; pthread_mutex_unlock(&tinfo->mtx); pthread_attr_destroy (&tinfo->cThreadAttr); delete tinfo;} void* CthreadPoolManage::startThread(void* arg){ cout<<"線程開始工作"<<endl; threadInfo* tinfo = (threadInfo*)arg; pthread_cleanup_push(threadCleanUp,arg); while(tinfo->doFlag){  pthread_mutex_lock(&tinfo->mtx);  if(tinfo->doit == NULL)  {   cout<<"開始等待任務(wù)"<<endl;   pthread_cond_wait(&tinfo->cond,&tinfo->mtx);   cout<<"有任務(wù)了"<<endl;  }  tinfo->isbusy = true;  tinfo->doit->start(tinfo->value);  tinfo->doit->end();  tinfo->doit=NULL;  tinfo->isbusy = false;  time( &tinfo->beginTime);  pthread_mutex_unlock(&tinfo->mtx); } //0正常執(zhí)行到這兒不執(zhí)行清理函數(shù)，異常會(huì)執(zhí)行 pthread_cleanup_pop(0); pthread_attr_destroy (&tinfo->cThreadAttr); delete tinfo; cout<<"線程結(jié)束"<<endl;} void CthreadPoolManage::initThread(){ int i = 0; for(i = 0;i<this->_minThreads;i++) {   threadInfo *tinfo = new threadInfo;   tinfo->doit = NULL;   tinfo->value = NULL;   tinfo->isbusy = false;   tinfo->doFlag = true;  // PTHREAD_CREATE_DETACHED (分離線程) 和 PTHREAD _CREATE_JOINABLE (非分離線程)   pthread_attr_init(&tinfo->cThreadAttr);   pthread_attr_setdetachstate(&tinfo->cThreadAttr,PTHREAD_CREATE_DETACHED );   cout<<"初始化了一個(gè)線程"<<endl;   if(pthread_create(&tinfo->cThreadPid,&tinfo->cThreadAttr,startThread,(void *)tinfo) != 0)  {  cout<<"創(chuàng)建線程失敗"<<endl;  break;  }  this->_threadList.push_back(tinfo); }} int CthreadPoolManage::addThread(std::list< CthreadPoolManage::threadInfo* >* plist, CthreadPoolManage::threadInfo* ptinfo){   threadInfo *tinfo = new threadInfo;   tinfo->doit = ptinfo->doit;   tinfo->value = ptinfo->value;   tinfo->isbusy = true;   if(pthread_create(&tinfo->cThreadPid,NULL,startThread,(void *)tinfo) != 0)  {  cout<<"創(chuàng)建線程失敗"<<endl;  return -1;  }  plist->push_back(tinfo);  return 0;}  int CthreadPoolManage::putDuty(CDoit* doit, void* value){ threadInfo *tinfo = new threadInfo; time( &tinfo->beginTime); tinfo->doit= doit; tinfo->value = value; pthread_mutex_lock(&_duty_mutex);  this->_dutyList.push_back(tinfo); pthread_mutex_unlock(&_duty_mutex); return 0;} void* CthreadPoolManage::taskAllocation(void*arg){ CthreadPoolManage * ptmanage = (CthreadPoolManage*)arg; int size_1 = 0; int size_2 = 0; int i_1 = 0; int i_2 = 0; bool a_1 = true; bool a_2 = true; threadInfo* ptinfo; threadInfo* ptinfoTmp; while(true){   size_1 = 0;   size_2 = 0;   pthread_mutex_lock(&ptmanage->_duty_mutex);   pthread_mutex_lock(&ptmanage->_thread_mutex);   size_1 = ptmanage->_dutyList.size();   size_2 =ptmanage->_threadList.size();   for(list<threadInfo*>::iterator itorti1 = ptmanage->_dutyList.begin();itorti1 !=ptmanage->_dutyList.end();)   {  ptinfo = *itorti1;  a_1 = true;  for(list<threadInfo*>::iterator itorti2 = ptmanage->_threadList.begin();itorti2!=ptmanage->_threadList.end();itorti2++){   ptinfoTmp = *itorti2;   if(EBUSY == pthread_mutex_trylock(&ptinfoTmp->mtx))   {    continue;   }   if(!ptinfoTmp->isbusy)   {    ptinfoTmp->doit = ptinfo->doit;    ptinfoTmp->value = ptinfo->value;    ptinfoTmp->isbusy = true;    pthread_cond_signal(&ptinfoTmp->cond);    pthread_mutex_unlock(&ptinfoTmp->mtx);    a_1 = false;    delete ptinfo;    break;   }   pthread_mutex_unlock(&ptinfoTmp->mtx);    }    if(a_1){   if(ptmanage->_threadList.size()>ptmanage->_maxThreads||ptmanage->addThread(&ptmanage->_threadList,ptinfo)!=0)   {    itorti1++;    continue;   }else{    itorti1 = ptmanage->_dutyList.erase(itorti1);   }   delete ptinfo;    }else{   itorti1 = ptmanage->_dutyList.erase(itorti1);    }   }   pthread_mutex_unlock(&ptmanage->_duty_mutex);   pthread_mutex_unlock(&ptmanage->_thread_mutex);   usleep(USLEEP_TIME); } return 0;} void* CthreadPoolManage::checkThread(void* arg){ CthreadPoolManage * ptmanage = (CthreadPoolManage*)arg; threadInfo* ptinfo; time_t nowtime; while(ptmanage->checkrun){  sleep(CHECK_TIME);  pthread_mutex_lock(&ptmanage->_thread_mutex);  if(ptmanage->_threadList.size()<=ptmanage->_minThreads)  {   pthread_mutex_unlock(&ptmanage->_thread_mutex);   continue;  }  for(list<threadInfo*>::iterator itorti2 = ptmanage->_threadList.begin();itorti2!=ptmanage->_threadList.end();){   ptinfo = *itorti2;   if(EBUSY == pthread_mutex_trylock(&ptinfo->mtx))  {   itorti2++;   continue;  }  time(&nowtime);  if(ptinfo->isbusy == false && nowtime-ptinfo->beginTime>ptmanage->_waitSec)  {   ptinfo->doFlag = false;   itorti2 = ptmanage->_threadList.erase(itorti2);  }else{   itorti2++;  }  pthread_mutex_unlock(&ptinfo->mtx);  }  pthread_mutex_unlock(&ptmanage->_thread_mutex); }} int CthreadPoolManage::start(){ //初始化 this->initThread(); //啟動(dòng)任務(wù)分配線程  if(pthread_create(&tasktPid,NULL,taskAllocation,(void *)this) != 0)  {  cout<<"創(chuàng)建任務(wù)分配線程失敗"<<endl;  return -1;  }  //創(chuàng)建現(xiàn)程狀態(tài)分配管理線程  if(pthread_create(&checktPid,NULL,checkThread,(void *)this) != 0)  {  cout<<"創(chuàng)建線程狀態(tài)分配管理線程失敗"<<endl;  return -1;  } return 0;} ///////////////////////////////int CthreadPoolManage::getNowThreadNum(){ int num = 0; pthread_mutex_lock(&this->_thread_mutex);  num = this->_threadList.size(); pthread_mutex_unlock(&this->_thread_mutex); return num ;}