ThreadPool.h

#ifndef __THREADPOOL_H #define __THREADPOOL_H#define HAVE_STRUCT_TIMESPEC//#include "servant/Application.h"#include 
     
        #include 
      
         #include 
       
          using namespace std;/*** 执行任务的类，设置任务数据、定义执行方法(纯虚函数)*/class CTask{protected:	string m_strTaskName;   //任务的名称	void* m_ptrData;        //具体数据public:	CTask() {}	CTask(string taskName)	{		m_strTaskName = taskName;		m_ptrData = NULL;	}	virtual int Run() = 0;       //任务执行方法	void SetData(void* data);    //设置任务数据public:	virtual ~CTask() {}};/*** 线程结构体*/struct CThread{	pthread_t pthread_id; //线程id	int iStat;            //线程状态	CThread()		: iStat(0)	{	}	bool operator == (const CThread &obj) const	{		return (long)&pthread_id == (long)&obj.pthread_id;	}};/*** 线程池管理类的实现*/class CThreadPool{public:	CThreadPool(int threadNum = 10);	int AddTask(CTask *task);                    //把任务添加到任务队列中    int getTaskSize();                           //获取当前任务队列中的任务数	int StopAll();                               //使线程池中的线程退出protected:	int Create();								 //创建线程池中的线程	static void* ThreadFunc(void * threadData);  //新线程的线程回调函数	static int MoveToIdle(CThread *pThread);     //线程执行结束后，状态置为空闲0	static int MoveToBusy(CThread *pThread);     //线程开始执行，状态置为运行1private:	static  vector
        
          m_vecTaskList;        //任务列表	static  bool shutdown;                       //线程退出标志	int     m_iThreadNum;                        //线程池中启动的线程数	static	vector
         
           m_vecThread;         //线程列表	static pthread_mutex_t m_pthreadMutex;       //线程同步锁	static pthread_cond_t m_pthreadCond;         //线程同步的条件变量};#endif
         
        
       
      
     

　　

ThreadPool.cpp

#include "ThreadPool.h"  #include 
     
        #include 
      
       using namespace std;void CTask::SetData(void * data){	m_ptrData = data;}vector
       
         CThreadPool::m_vecTaskList;         //任务列表  bool CThreadPool::shutdown = false;vector
        
          CThreadPool::m_vecThread;          //线程列表pthread_mutex_t CThreadPool::m_pthreadMutex = PTHREAD_MUTEX_INITIALIZER;pthread_cond_t CThreadPool::m_pthreadCond = PTHREAD_COND_INITIALIZER;/*** 线程池管理类构造函数*/CThreadPool::CThreadPool(int threadNum){	this->m_iThreadNum = threadNum;	cout << "threadNum:" << threadNum << " threads will be created." << endl;	Create(); //创建线程}/*** 创建线程*/int CThreadPool::Create(){	m_vecThread.resize(m_iThreadNum);	for (size_t i = 0; i < m_vecThread.size(); i++)	{		pthread_create(&m_vecThread[i].pthread_id, NULL, ThreadFunc, &m_vecThread[i]);	}	return 0;}/*** 线程回调函数*/void* CThreadPool::ThreadFunc(void* threadData){	CThread *pThread = (CThread*)threadData;	while (1)	{		pthread_mutex_lock(&m_pthreadMutex); //lock		while (m_vecTaskList.size() == 0 && !shutdown)		{			pthread_cond_wait(&m_pthreadCond, &m_pthreadMutex);             /*            pthread_cond_wait前要先加锁            pthread_cond_wait把线程放进阻塞队列后,内部会解锁，然后等待条件变量被其它线程唤醒            pthread_cond_wait被唤醒后会再自动加锁            */		}		if (shutdown)		{			pthread_mutex_unlock(&m_pthreadMutex);			cout << "thread:" << (long)&pThread->pthread_id << " will exit." << endl;			pthread_exit(NULL);		}        //线程状态置1		MoveToBusy(pThread);        //取出一个任务        CTask* task = NULL;		vector
         
          ::iterator iter = m_vecTaskList.begin();		if (iter != m_vecTaskList.end())		{			task = *iter;			m_vecTaskList.erase(iter);		}		pthread_mutex_unlock(&m_pthreadMutex); //unlock        //执行任务        if (task)        {            task->Run();        }        //线程状态置0		MoveToIdle(pThread);	}	return (void*)0;}int CThreadPool::MoveToIdle(CThread *pThread){	vector
          
           ::iterator iter_thread = std::find(m_vecThread.begin(), m_vecThread.end(), *pThread); if (iter_thread != m_vecThread.end()) { iter_thread->iStat = 0; cout << "tid:" << (long)&pThread->pthread_id << " idle." << endl; } return 0;}int CThreadPool::MoveToBusy(CThread *pThread){ vector
           
            ::iterator iter_thread = std::find(m_vecThread.begin(), m_vecThread.end(), *pThread); if (iter_thread != m_vecThread.end()) { iter_thread->iStat = 1; cout << "tid:" << (long)&pThread->pthread_id << " run." << endl; } return 0;}/*** 往任务队列里边添加任务并发出线程同步信号*/int CThreadPool::AddTask(CTask *task){ pthread_mutex_lock(&m_pthreadMutex); this->m_vecTaskList.push_back(task); pthread_cond_signal(&m_pthreadCond); pthread_mutex_unlock(&m_pthreadMutex); return 0;}/*** 获取当前队列中任务数*/int CThreadPool::getTaskSize(){ return m_vecTaskList.size();}/*** 停止所有线程*/int CThreadPool::StopAll(){ /** 避免重复调用 */ if (shutdown) { return -1; } cout << "All threads will be stoped." << endl; /** 唤醒所有等待线程，线程池要销毁了 */ shutdown = true; pthread_cond_broadcast(&m_pthreadCond); /** 阻塞等待线程退出，否则就成僵尸了 */ for (size_t i = 0; i < m_vecThread.size(); i++) { pthread_join(m_vecThread[i].pthread_id, NULL); } m_vecThread.clear(); /** 销毁条件变量和互斥体 */ pthread_mutex_destroy(&m_pthreadMutex); pthread_cond_destroy(&m_pthreadCond); return 0;}
           
          
         
        
       
      
     

　　

main.cpp

#include 
     
      #include "ThreadPool.h"using namespace std;class CMyTask : public CTask{public:	CMyTask() {}	inline int Run()	{        cout << (char*)this->m_ptrData << endl;		return 0;	}};int main(){	CThreadPool threadPool(10);    CMyTask taskObj;    char szTmp[] = "this is the first thread running";    taskObj.SetData((void*)szTmp);    for (int i = 0; i < 10; i++)    {        threadPool.AddTask(&taskObj);    }	while (1)	{        cout << "there are still " << threadPool.getTaskSize() << " tasks need to handle" << endl;		if (threadPool.getTaskSize() == 0)		{			if (threadPool.StopAll() == -1)			{                cout << "Now I will exit from main" << endl;				return 0;			}		}	}	return 0;}
     

　　

makeflie

TARGET:=threadpoolINC:= -I./LIB_PATH:=LIB:= -lpthreadCFLAGS:=-Wall -g -O0 -D_REENTRANT -Wl,-rpath=./ $(INC) $(LIB_PATH)CPPFLAGS:=$(CFLAGS)SRC:=$(shell echo *.cpp)OBJ:=$(patsubst %.cpp,%.o,$(SRC))all: $(TARGET)$(TARGET): $(OBJ)	$(CXX) $^ $(CFLAGS) $(LIB) -o $@clean:	rm -f $(OBJ)	rm -f $(TARGET)

　　

windows下配置 pthread 参见：