快速理解Linux C/C++线程池

9.5.1  线程池的定义

这里的池是形象的说法。线程池就是有一堆已经创建好了的线程，初始都处于空闲等待状态，当有新的任务需要处理的时候，就从这堆线程（这堆线程比喻为线程池）中取一个空闲等待的线程来处理该任务，当任务处理完毕后，就把该线程放回池中（一般就是将线程状态置为空闲），以供后面的任务继续使用。当池子里的线程全都处于忙碌状态时，线程池中没有可用的空闲等待线程，此时可以根据需要选择创建一个新的线程并置入池中，或者通知任务当前线程池里所有线程都在忙，等待片刻再尝试。这个过程可以用图9-2来表示。

图9-2

线程的创建和销毁相对于进程的创建和销毁来说是轻量级的（开销没有进程那么大），但是当我们的任务需要进行大量线程的创建和销毁操作时，这些开销合在一起就比较大了。比如，设计一个压力性能测试框架的时候，需要连续产生大量的并发操作。线程池在这种场合是非常适用的。线程池的好处就在于线程复用，某个线程在处理完一个任务后，可以继续处理下一个任务，而不用销毁后再创建，这样可以避免无谓的开销，因此尤其适用于连续产生大量并发任务的场合。

在知道了线程池的基本概念后，下面我们用Linux C++来实现一个基本的线程池，该线程池虽然简单，但可以体现线程池的基本工作思想。另外，线程池的实现是千变万化的，有时候要根据实际应用场合来定制，但万变不离其宗，原理都是一样的。现在我们从简单的、基本的线程池开始实践，为以后工作中设计复杂高效的线程池做准备。

【例9.9】用C++实现一个简单的线程池

（1）打开Visual Studio Code，新建文本文件并输入如下代码：

#ifndef __THREAD_POOL_H
#define __THREAD_POOL_H#include <vector>
#include <string>
#include <pthread.h>using namespace std;/*执行任务的类：设置任务数据并执行*/
class CTask {
protected:string m_strTaskName;   	// 任务的名称void* m_ptrData;    			// 要执行的任务的具体数据public:CTask() = default;CTask(string &taskName): m_strTaskName(taskName), m_ptrData(NULL) {}virtual int Run() = 0;void setData(void* data); 	// 设置任务数据virtual ~CTask() {}};/*线程池管理类*/
class CThreadPool {
private:static vector<CTask*> m_vecTaskList;    	// 任务列表static bool shutdown;   					// 线程退出标志int m_iThreadNum;   						// 线程池中启动的线程数pthread_t *pthread_id;static pthread_mutex_t m_pthreadMutex; 		// 线程同步锁static pthread_cond_t m_pthreadCond;  		// 线程同步条件变量protected:static void* ThreadFunc(void *threadData);	// 新线程的线程回调函数static int MoveToIdle(pthread_t tid);	// 线程执行结束后，把自己放入空闲线程中static int MoveToBusy(pthread_t tid);	// 移入到忙碌线程中去int Create();   							// 创建线程池中的线程public:CThreadPool(int threadNum);int AddTask(CTask *task);   			// 把任务添加到任务队列中int StopAll();  						// 使线程池中的所有线程退出int getTaskSize();  					// 获取当前任务队列中的任务数
};#endif

（2）保存代码为头文件thread_pool.h，再新建一个thread_pool.cpp文件，并输入如下代码：

#include "thread_pool.h"
#include <cstdio>void CTask::setData(void* data) {m_ptrData = data;
}// 静态成员初始化
vector<CTask*> CThreadPool::m_vecTaskList;
bool CThreadPool::shutdown = false;
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;printf("I will create %d threads.\n", threadNum);Create();
}// 线程回调函数
void* CThreadPool::ThreadFunc(void* threadData) {pthread_t tid = pthread_self();while (1){pthread_mutex_lock(&m_pthreadMutex);// 如果队列为空，等待新任务进入任务队列while (m_vecTaskList.size() == 0 && !shutdown)pthread_cond_wait(&m_pthreadCond, &m_pthreadMutex);// 关闭线程if (shutdown){pthread_mutex_unlock(&m_pthreadMutex);printf("[tid: %lu]\texit\n", pthread_self());pthread_exit(NULL);}printf("[tid: %lu]\trun: ", tid);vector<CTask*>::iterator iter = m_vecTaskList.begin();// 取出一个任务并处理CTask* task = *iter;if (iter != m_vecTaskList.end()){task = *iter;m_vecTaskList.erase(iter);}pthread_mutex_unlock(&m_pthreadMutex);task->Run();    // 执行任务printf("[tid: %lu]\tidle\n", tid);}return (void*)0;
}// 往任务队列里添加任务并发出线程同步信号
int CThreadPool::AddTask(CTask *task) { pthread_mutex_lock(&m_pthreadMutex);    m_vecTaskList.push_back(task);  pthread_mutex_unlock(&m_pthreadMutex);  pthread_cond_signal(&m_pthreadCond);    return 0;
}// 创建线程
int CThreadPool::Create() { pthread_id = new pthread_t[m_iThreadNum];for (int i = 0; i < m_iThreadNum; i++)pthread_create(&pthread_id[i], NULL, ThreadFunc, NULL);return 0;
}// 停止所有线程
int CThreadPool::StopAll() {    // 避免重复调用if (shutdown)return -1;printf("Now I will end all threads!\n\n");// 唤醒所有等待线程，线程池也要销毁了shutdown = true;pthread_cond_broadcast(&m_pthreadCond);// 清理僵尸线程for (int i = 0; i < m_iThreadNum; i++)pthread_join(pthread_id[i], NULL);delete[] pthread_id;pthread_id = NULL;// 销毁互斥锁和条件变量pthread_mutex_destroy(&m_pthreadMutex);pthread_cond_destroy(&m_pthreadCond);return 0;
}// 获取当前队列中的任务数
int CThreadPool::getTaskSize() {    return m_vecTaskList.size();
}

（3）新建一个main.cpp，输入如下代码：

#include "thread_pool.h"
#include <cstdio>
#include <stdlib.h>
#include <unistd.h>class CMyTask : public CTask {   
public:CMyTask() = default;    int Run() {printf("%s\n", (char*)m_ptrData);int x = rand() % 4 + 1;sleep(x);   return 0;}~CMyTask() {}
};int main() {CMyTask taskObj;char szTmp[] = "hello!";taskObj.setData((void*)szTmp);CThreadPool threadpool(5);  // 线程池大小为5for (int i = 0; i < 10; i++)threadpool.AddTask(&taskObj);while (1) {printf("There are still %d tasks need to handle\n", threadpool.
getTaskSize());// 任务队列里已经没有任务了if (threadpool.getTaskSize() == 0) {// 清除线程池if (threadpool.StopAll() == -1) {printf("Thread pool clear, exit.\n");exit(0);}}sleep(2);printf("2 seconds later...\n");}   return 0;
}

（4）把这3个文件上传到Linux，在命令行下编译并运行：

# g++ thread_pool.cpp test.cpp -o test -lpthread
# ./test
I will create 5 threads.
There are still 10 tasks need to handle
[tid: 139992529053440]  run: hello!
[tid: 139992520660736]  run: hello!
[tid: 139992512268032]  run: hello!
[tid: 139992503875328]  run: hello!
[tid: 139992495482624]  run: hello!
2 seconds later...
There are still 5 tasks need to handle
[tid: 139992512268032]  idle
[tid: 139992512268032]  run: hello!
[tid: 139992495482624]  idle
[tid: 139992495482624]  run: hello!
[tid: 139992520660736]  idle
[tid: 139992520660736]  run: hello!
[tid: 139992529053440]  idle
[tid: 139992529053440]  run: hello!
[tid: 139992503875328]  idle
[tid: 139992503875328]  run: hello!
2 seconds later...
There are still 0 tasks need to handle
Now I will end all threads![tid: 139992520660736]  idle
[tid: 139992520660736]  exit
[tid: 139992495482624]  idle
[tid: 139992495482624]  exit
[tid: 139992512268032]  idle
[tid: 139992512268032]  exit
[tid: 139992529053440]  idle
[tid: 139992529053440]  exit
[tid: 139992503875328]  idle
[tid: 139992503875328]  exit
2 seconds later...
There are still 0 tasks need to handle
Thread pool clear, exit.