c++线程池小例子

ThreadPool.h

#ifndef __THREADPOOL_H 
#define __THREADPOOL_H

#define HAVE_STRUCT_TIMESPEC

//#include "servant/Application.h"
#include <vector>  
#include <string>  
#include <pthread.h>  

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);     //线程开始执行，状态置为运行1

private:
	static  vector<CTask*> m_vecTaskList;        //任务列表

	static  bool shutdown;                       //线程退出标志
	int     m_iThreadNum;                        //线程池中启动的线程数
	static	vector<CThread> m_vecThread;         //线程列表

	static pthread_mutex_t m_pthreadMutex;       //线程同步锁
	static pthread_cond_t m_pthreadCond;         //线程同步的条件变量
};

#endif  

　　

ThreadPool.cpp

#include "ThreadPool.h"  
#include <iostream>  
#include <algorithm>
using namespace std;

void CTask::SetData(void * data)
{
	m_ptrData = data;
}

vector<CTask*> CThreadPool::m_vecTaskList;         //任务列表  
bool CThreadPool::shutdown = false;
vector<CThread> 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<CTask*>::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<CThread>::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<CThread>::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 <iostream>
#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:=threadpool
INC:= -I./
LIB_PATH:=
LIB:= -lpthread

CFLAGS:=-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 参见：https://blog.csdn.net/qianchenglenger/article/details/16907821

线程同步、条件变量说明参见：http://www.javashuo.com/article/p-tiflogox-be.html