详细的线程池讲解，手写C与C++版本

在此感谢苏丙榅的教程讲的很详细，我看了他的C版本教程，对线程池有了深刻理解，手写了C版本，并自主改了C++版本。

线程池是消费者生产者模型的其中之一。这里面的线程城同步很重要，稍不注意就会造成死锁。主要用的是互斥锁mutex。学习线程池需要有线程的基础知识，比如线程创建，互斥锁，销毁线程，条件变量等等。接下来详细介绍线程池。

首先来看下图，下图所示了单进程单线程系统处理任务的方式，很多任务任务进来时，需要排队处理：

下面让我们看进程池的处理方式，看下图：

在这里简单概括下线程池工作原理：

简单来说，任务进来后被添加到任务队列，当线程池里有空闲的线程时（阻塞态），该线程会被唤醒，尝试拿线程池的锁，拿到锁后的线程会从任务队列里取任务，任务队列任务少一，处理完任务后，若任务队列为空则再次进入阻塞态，当任务队列任务过多时，该条件自己设定，会增加存活的线程数量，但小于总的线程池大小，可以看出这是一个典型的消费者生产者模型。

线程池的重要组成部分：

线程池的重要组成部分
1、任务队列：	存放任务，一个
2、工作者线程：	尝试从任务队列取任务并处理，N个
3、管理者线程：	用来管理线程池的线程，方便扩容与缩小，一个。

接下来手写线程池：

任务结构体：

用来存放任务函数与任务函数参数：

C语言版本

//任务结构体
typedef struct Task
{void (*fun)(void * arg);void * arg;
}TASK;

C++版本

//任务结构体
struct Task
{void (*fun)(void * arg);void * arg;
};

线程池结构体/类：

用来定义线程池里所需要的东西：线程池大小，忙的线程，存活的线程，任务队列，队列大小，队列容量，管理者线程ID，工作线程ID，线程池销毁标志，两把锁（锁忙线程数，锁线程池），两个条件变量（任务队列空，任务队列满），等等。

C语言版本：

//线程池结构体
struct pthread_pool
{/* 任务对列*/TASK * task;int queueCapacity; //容量int queueSize;     //当前任务个数int queueFront;    //队头——>取数据int queueRear;     //队尾——>存数据//线程池相关pthread_t manmagerID; //管理者线程IDpthread_t *workerIDs;   //工作的线程IDint minNum;  //最小线程数int maxNum;  //最大线程数int busyNum; //忙碌线程数//busy单独加锁，因为经常修改int liveNum; //存活线程数int exitNum; //要销毁的线程数pthread_mutex_t mutexpool; //锁整个线程池pthread_mutex_t mutexBusy; //锁busyNum变量pthread_cond_t  notFull; //任务对列是否满pthread_cond_t  notEmpty;//任务对列是否为空int shutdown; //是不是要销毁线程池，销毁为1，不销毁为0；};

C++版本：

//创建线程池类
class pthread_pool{private:int max_num{}; //线程池最大容量int min_num{}; //线程池最小容量int queueCapacity{};//任务队列容量std::queue<task> Task;//创建任务队列pthread_t manager{};//管理者线程IDstd::vector<pthread_t> workerIDs; //存储工作者线程IDsint shutdown{};//关闭线程池标志int busy_num{};//忙的线程数量int alive_num{};//活的线程数量int destory_or_add_num{}; //要销毁的线程数//两把锁，两个条件变量pthread_mutex_t mutexbusy{};pthread_mutex_t mutexpool{};pthread_cond_t  not_empty{};pthread_cond_t  notfull{};public:pthread_pool();pthread_pool(int min_num,int max_num,int queue_capacity);static void * pthread_manager(void * arg);static void * pthread_worker(void * arg);void pthread_add_work(pthread_pool * pool,void(*func)(void*),void * arg);int pthread_destory_pool(pthread_pool * pool);void pthreadExit(pthread_pool *pool);~pthread_pool();};

创建线程：

C语言版本：有点c++内容，把new换成malloc即可，nullptr换成NULL

pthread_pool * pthread_pool_create(int min,int max,int queueSize){pthread_pool * pool=new pthread_pool;if(pool==nullptr){printf("pool new is error");return nullptr;}do{pool->workerIDs = new pthread_t[max];if(pool->workerIDs==nullptr){printf("pool->workerIDs new is error");break;}memset(pool->workerIDs,0,sizeof(pool->workerIDs)); //这里有点不一样pool->minNum = min;pool->maxNum = max;pool->busyNum= 0;pool->liveNum= min;pool->exitNum= 0;//初始化两个锁，两个条件变量if(pthread_mutex_init(&pool->mutexpool,nullptr)!=0||pthread_mutex_init(&pool->mutexBusy,nullptr)!=0||pthread_cond_init(&pool->notEmpty,nullptr)!=0||pthread_cond_init(&pool->notFull,nullptr)!=0){printf("mutex or cond is error");break;}//任务队列pool->task = new TASK[queueSize];pool->queueCapacity = queueSize;pool->queueFront = 0;pool->queueRear = 0;pool->shutdown =0;//创建线程pthread_create(&pool->manmagerID,nullptr,manager,pool);for(int i = 0;i<min;++i){pthread_create(&pool->workerIDs[i],nullptr,worker,pool);}return pool;}while(0);if(pool&&pool->workerIDs) delete [] pool->workerIDs;if(pool&&pool->task) delete [] pool->task;if(pool) delete pool;return nullptr;
}

c++版本：

pthread_pool::pthread_pool(int min_num, int max_num,int queue_capacity) {this->workerIDs.insert(this->workerIDs.begin(),max_num,0);this->min_num=min_num;this->max_num=max_num;this->busy_num=0;this->alive_num=min_num;this->queueCapacity=queue_capacity;this->shutdown=0;this->destory_or_add_num=0;//创建管理者线程pthread_create(&this->manager, nullptr,pthread_manager, this);//传this指针//创建消费者线程for(auto &w_IDs:workerIDs){pthread_create(&w_IDs, nullptr,pthread_worker,this);//传this指针}//两把锁两个条件变量if(pthread_mutex_init(&this->mutexbusy, nullptr)!=0|| pthread_mutex_init(&this->mutexpool, nullptr)|| pthread_cond_init(&this->not_empty, nullptr)!=0|| pthread_cond_init(&notfull,nullptr)!=0){printf("construct is failed");exit(-1);}}

管理者线程：

当任务过多的时候，可以适当的创建一些新的工作线程
当任务过少的时候，可以适当的销毁一些工作的线程

C语言版本：

//管理者线程
void * manager(void * arg){pthread_pool * pool=(pthread_pool *) arg;while(!pool->shutdown){//每隔N秒检测一次sleep(N);//取出线程池中任务数量和当前线程数量pthread_mutex_lock(&pool->mutexpool);int queueSize = pool->queueSize;int liveNum   = pool->liveNum;pthread_mutex_unlock(&pool->mutexpool);//取出忙的线程的数量pthread_mutex_lock(&pool->mutexBusy);int busyNum = pool->busyNum;pthread_mutex_unlock(&pool->mutexBusy);//添加线程if(queueSize > liveNum&&liveNum<pool->maxNum){int count=0;pthread_mutex_lock(&pool->mutexpool);for(int i=0;i<pool->maxNum&&pool->liveNum<pool->maxNum&&count<create_or_destory_P;++i){if(pool->workerIDs[i]==0){count++;pthread_create(&pool->workerIDs[i],nullptr,worker,pool);pool->liveNum++;}}  pthread_mutex_unlock(&pool->mutexpool);         }//删除线程if(busyNum*2<liveNum&&liveNum>pool->minNum){int count=0;pthread_mutex_lock(&pool->mutexpool);pool->exitNum = create_or_destory_P;pthread_mutex_unlock(&pool->mutexpool);         //让线程自杀for(int i=0;i<create_or_destory_P;++i){pthread_cond_signal(&pool->notEmpty);}}}return nullptr;
}

c++版本

void * pthread_pool::pthread_manager(void * arg) {auto * pool= (pthread_pool *)arg;while(!pool->shutdown){sleep(N);//取出忙的线程数pthread_mutex_lock(&pool->mutexbusy);int busy=pool->busy_num;pthread_mutex_unlock(&pool->mutexbusy);pthread_mutex_lock(&pool->mutexpool);int queue_size=pool->Task.size();int live_num=pool->alive_num;pthread_mutex_unlock(&pool->mutexpool);if(queue_size>live_num&&pool->alive_num<pool->max_num){int count=0;pthread_mutex_lock(&pool->mutexpool);for(int i=0;i<pool->max_num&&count<d_or_a&&pool->alive_num<pool->max_num;++i){if(pool->workerIDs[i]==0){pthread_create(&pool->workerIDs[i], nullptr,pthread_worker,pool);count++;pool->alive_num++;}}pthread_mutex_unlock(&pool->mutexpool);}if(busy*2>live_num&&pool->alive_num>pool->min_num){pthread_mutex_lock(&pool->mutexpool);pool->destory_or_add_num=d_or_a;pthread_mutex_unlock(&pool->mutexpool);for(int i=0;i<d_or_a;++i){pthread_cond_signal(&pool->not_empty);}}}return nullptr;
}

工作线程：

线程池中维护了一定数量的工作线程，他们的作用是是不停的读任务队列，从里边取出任务并处理，作的线程相当于是任务队列的消费者角色，如果任务队列为空，工作的线程将会被阻塞 (使用条件变量 / 信号量阻塞)如果阻塞之后有了新的任务，由生产者将阻塞解除，工作线程开始工作。

C语言版本

//工作线程
void * worker(void *arg){printf("********************\n");pthread_pool * pool=(pthread_pool *) arg;while(1){pthread_mutex_lock(&pool->mutexpool);//当前任务队列是否为空while(pool->queueSize == 0 && !pool->shutdown){pthread_cond_wait(&pool->notEmpty,&pool->mutexpool);if(pool->exitNum>0){pool->exitNum--;if(pool->liveNum>pool->minNum){pool->liveNum--;pthread_mutex_unlock(&pool->mutexpool);pthreadExit(pool);}    }}//判断线程池是否关闭if(pool->shutdown){pthread_mutex_unlock(&pool->mutexpool); //为什么需要再次拿锁pthreadExit(pool);}//从任务队列取一个任务TASK _task;_task.fun = pool->task[pool->queueFront].fun;_task.arg = pool->task[pool->queueFront].arg;//移动头结点 环形队列pool->queueFront = (pool->queueFront+1) % pool->queueCapacity;pool->queueSize--;pthread_cond_signal(&pool->notFull);pthread_mutex_unlock(&pool->mutexpool);printf("start\n");//修改busynumpthread_mutex_lock(&pool->mutexBusy);pool->busyNum++;pthread_mutex_unlock(&pool->mutexBusy);_task.fun(_task.arg);delete _task.arg;_task.arg=nullptr;//(*_task.fun)(_task.arg);//修改busysumpthread_mutex_lock(&pool->mutexBusy);pool->busyNum--;pthread_mutex_unlock(&pool->mutexBusy);printf("end\n");}return nullptr;
}

C++版本

void * pthread_pool::pthread_worker(void *arg) {printf("***********\n");auto * pool= (pthread_pool *)arg;while(1){pthread_mutex_lock(&pool->mutexpool);while(!pool->shutdown && pool->Task.size()<=0){pthread_cond_wait(&pool->not_empty,&pool->mutexpool);if(pool->destory_or_add_num>0){pool->destory_or_add_num--;if(pool->alive_num>pool->min_num){pool->alive_num--;pthread_mutex_unlock(&pool->mutexpool);pool->pthreadExit(pool);}}}if(pool->shutdown){pthread_mutex_unlock(&pool->mutexpool);pool->pthreadExit(pool);}task _task;_task.fun=pool->Task.front().fun;_task.arg=pool->Task.front().arg;pool->Task.pop();pthread_cond_signal(&pool->notfull);pthread_mutex_unlock(&pool->mutexpool);pthread_mutex_lock(&pool->mutexbusy);pool->busy_num++;pthread_mutex_unlock(&pool->mutexbusy);pthread_mutex_lock(&pool->mutexpool);printf("start!!!!!!\n");_task.fun(_task.arg);//delete arg;printf("end!!!!!!\n");printf("***********\n");pthread_mutex_unlock(&pool->mutexpool);pthread_mutex_lock(&pool->mutexbusy);pool->busy_num--;pthread_mutex_unlock(&pool->mutexbusy);}return nullptr;
}

添加任务到任务队列函数

C语言版本

void pthreadpooladd(pthread_pool *pool,void(*func)(void*),void *arg){pthread_mutex_lock(&pool->mutexpool);while(pool->queueSize==pool->queueCapacity&&!pool->shutdown){ //判断线程池任务队列是否有容量pthread_cond_wait(&pool->notFull,&pool->mutexpool);}if(pool->shutdown){pthread_mutex_unlock(&pool->mutexpool);return;}pool->task[pool->queueRear].fun=func;pool->task[pool->queueRear].arg=arg;pool->queueRear = (pool->queueRear+1)%pool->queueCapacity;pool->queueSize++;pthread_cond_signal(&pool->notEmpty);pthread_mutex_unlock(&pool->mutexpool);
}

C++版本

void pthread_pool::pthread_add_work(pthread_pool *pool,void(*func)(void*),void *arg) {pthread_mutex_lock(&pool->mutexpool);while(pool->Task.size()==pool->queueCapacity&&!pool->shutdown){pthread_cond_wait(&notfull,&pool->mutexpool);}if(pool->shutdown){pthread_mutex_unlock(&pool->mutexpool);return;}task _task;_task.fun=func;_task.arg=arg;pool->Task.push(_task);pthread_cond_signal(&pool->not_empty);pthread_mutex_unlock(&pool->mutexpool);
}

未来计划：把线程池加入《干饭聊天室》

在此感谢苏丙榅的教程;