本作品采用知识共享署名-非商业性使用-相同方式共享 4.0 国际许可协议进行许可。
C语言实现线程池
- intro
- 线程池的组成
- 任务队列
- 工作线程与管理者线程
- 线程池结构体
- 线程池头文件(.h)
- 具体函数实现
- 完整代码
- 参考
- 本文代码附带详细注释
- 本文主要讲述线程池实现的细节,其余不详细描述
- 为保证精简,代码无错误检查,附带错误检查的详细代码见文章末尾
- 任务队列:存储需要处理的任务
- 工作线程:处理任务 / n个
- 管理者线程:管理工作线程 / 1个
- 以结构体方式定义
typedef struct Task
{
void (*func)(void *arg);//指针函数:接受一个void*,无返回
void *arg; //具体任务
}Task;
工作线程与管理者线程
- 构成线程池的主要部分
struct ThreadPool
{
//任务队列
Task *taskQ;
int queueCapacity; //最大容量
int queueSize; //当前任务个数
int queueBegin; //队头 ： 取数据
int queueEnd; //队尾 ： 放数据
pthread_t managerID; //管理者线程ID
pthread_t *threadIDs; //工作线程ID
int minNum_Threads; //最小线程 个数
int maxNum_Threads; //最大线程 个数
int busyNum_Threads; //忙线程 个数
int liveNum_Threads; //存活线程 个数
int exitNum_Threads; //要销毁的线程 个数
pthread_mutex_t mutexPool; //大锁：锁线程池
pthread_mutex_t mutexBusy; //小锁：锁忙线程个数(此变量最常用)
pthread_cond_t Full; //任务队列是否满
pthread_cond_t Empty; //任务队列是否空
int shutdown; //是否销毁线程池：销毁为1，不销毁为0
};
线程池头文件(.h)
#ifndef _THREADPOOL_H #define _THREADPOOL_H #define NUM 2 #include具体函数实现#include #include #include #include typedef struct ThreadPool ThreadPool; // 创建线程池并初始化 ThreadPool *ThreadPoolCreate(int min, int max, int queueSize); // 工作线程任务函数 void *worker(void *arg); // 管理者线程任务函数 void *manager(void *arg); // 销毁线程池 int threadPoolDestroy(ThreadPool *pool); // 给线程池添加任务 void threadPoolAdd(ThreadPool *pool, void(*func)(void*), void*arg); // 单个线程退出 void threadExit(ThreadPool* pool); #endif // !_THREADPOOL_H
- 创建线程池并初始化
ThreadPool *ThreadPoolCreate(int min, int max, int queueSize)
//接受参数为 线程池最小线程数 线程池最大线程数 任务队列
{
ThreadPool *pool = (ThreadPool *)malloc(sizeof(ThreadPool));//malloc一块连续地址
do{
//线程
pool->threadIDs = (pthread_t *)malloc(sizeof(pthread_t)*max);
memset(pool->threadIDs, 0, sizeof(pthread_t)*max);
pool->maxNum_Threads = max;//最大线程个数
pool->minNum_Threads = min;//最小线程个数
pool->liveNum_Threads = min;//存活的线程个数:相等于最小数
pool->busyNum_Threads = 0;//忙的线程个数:现在还没有任务被读取
pool->exitNum_Threads = 0;//需要推出的线程的个数
//锁与条件变量
pthread_mutex_init(&pool->mutexPool, NULL);
pthread_mutex_init(&pool->mutexBusy, NULL);
pthread_cond_init(&pool->Empty, NULL);
pthread_cond_init(&pool->Full, NULL);
//任务队列
pool->taskQ = (Task*)malloc(sizeof(Task)*queueSize);
pool->queueCapacity = queueSize;//队列最大容量
pool->queueSize = 0;//队列当前任务个数
pool->queueBegin = 0;//队头
pool->queueEnd = 0;//队尾
//线程池是是否需要销毁flag
pool->shutdown = 0;
//创建管理者线程
pthread_create(&pool->managerID, NULL, manager, pool);
//循环创建工作线程
for(int i = 0; i < min; i++){
pthread_create(&pool->threadIDs[i], NULL, worker, pool);
}
return pool;
}while(0);
//do...while中并无break是因为当前代码不附带错误检查,只提供框架,详细代码参见文章末尾
//销毁资源:只有当从从do...while中异常跳出才需要
if(pool && pool->threadIDs) free(pool->threadIDs);
if(pool && pool->taskQ) free(pool->taskQ);
if(pool) free(pool);
return NULL;//说明创建失败了
}
- // 工作线程任务函数
void *worker(void* arg)
{
ThreadPool *pool = (ThreadPool*)arg;
while(1){
pthread_mutex_lock(&pool->mutexPool);
// 判断当前任务队列是否为空
while(pool->queueSize == 0 && !pool->shutdown){
pthread_cond_wait(&pool->Empty, &pool->mutexPool);//如果空或线程池需要销毁,则将线程阻塞于此处
if(pool->exitNum_Threads > 0){//根据需要退出的线程的个数来决定需要自杀的线程个数
//线程从此处往下执行就是去销毁自己
pool->exitNum_Threads--;
//再加一层判断,确保即使此线程自杀,存活的线程数仍大于要求的最小线程数
if(pool->liveNum_Threads > pool->minNum_Threads){
pool->liveNum_Threads--;
pthread_mutex_unlock(&pool->mutexPool);
threadExit(pool);
}
}
}
// 判断线程池是否已关闭
if(pool->shutdown == 1){
pthread_mutex_unlock(&pool->mutexPool);
threadExit(pool);
}
//如果任务队列不为空,线程池不需要销毁,那么工作线程就可以开始上班了
//从任务队列取出一个任务
Task task;
task.func = pool->taskQ[pool->queueBegin].func;
task.arg = pool->taskQ[pool->queueBegin].arg;
// 移动头结点
pool->queueBegin = (pool->queueBegin + 1) % pool->queueCapacity;//循环
pool->queueSize--;
//解锁
pthread_cond_signal(&pool->Full);//通知阻塞在Full的线程开始生产
pthread_mutex_unlock(&pool->mutexPool);
printf("thread %ld start workingn",pthread_self());
pthread_mutex_lock(&pool->mutexBusy);
pool->busyNum_Threads++;//改变busy值
pthread_mutex_unlock(&pool->mutexBusy);
task.func(task.arg);//指针函数处理arg
free(task.arg);
task.arg = NULL;
printf("thread %ld end workingn",pthread_self());
pthread_mutex_lock(&pool->mutexBusy);
pool->busyNum_Threads--;
pthread_mutex_unlock(&pool->mutexBusy);
}
return NULL;
}
- 管理者线程任务函数
void *manager(void * arg)
{
ThreadPool *pool = (ThreadPool*)arg;
while(pool->shutdown == 0){
// 每3s检测一次
sleep(3);
// 取出线程池中任务的数量和当前线程的数量
pthread_mutex_lock(&pool->mutexPool);
int queueSize = pool->queueSize;
int liveNum = pool->liveNum_Threads;
pthread_mutex_unlock(&pool->mutexPool);
// 取出忙线程个数
pthread_mutex_lock(&pool->mutexBusy);
int busyNum = pool->busyNum_Threads;
pthread_mutex_unlock(&pool->mutexBusy);
//添加线程(情况1)
//任务个数 > 存活线程个数 && 存活线程个数 < 最大线程数
if(queueSize > liveNum && liveNum < pool->maxNum_Threads){
pthread_mutex_lock(&pool->mutexPool);
int counter = 0;
for(int i = 0;
i < pool->maxNum_Threads &&
counter < NUM &&
pool->liveNum_Threads < pool->maxNum_Threads;i++){
if(pool->threadIDs[i] == 0){
pthread_create(&pool->threadIDs[i], NULL, worker, pool);
counter++;
pool->liveNum_Threads++;
}
}
pthread_mutex_unlock(&pool->mutexPool);
}
// 销毁线程(情况2)
// 忙的线程个数*2 < 存活线程数(此条逻辑取决于具体的业务需要) && 存活的线程 > 最小线程数
if(busyNum*2 < liveNum && liveNum > pool->minNum_Threads){
pthread_mutex_lock(&pool->mutexPool);
pool->exitNum_Threads = NUM;
pthread_mutex_unlock(&pool->mutexPool);
//唤醒堵塞线程使其自杀
for(int i = 0; i < NUM; i++){
pthread_cond_signal(&pool->Empty);
}
}
}
}
- 销毁线程池
int threadPoolDestroy(ThreadPool *pool)
{
// 关闭线程池
pool->shutdown = 1;
// 阻塞回收管理者线程
pthread_join(pool->managerID, NULL);
// 唤醒阻塞的消费者线程
for(int i = 0; i < pool->liveNum_Threads; i++){
pthread_cond_signal(&pool->Empty);//每次只能自杀一个,唤醒多了也没用
}
//释放malloc出来的堆内存
if(pool->taskQ){
free(pool->taskQ);
}
if(pool->threadIDs){
free(pool->threadIDs);
}
//解锁
pthread_mutex_destroy(&pool->mutexPool);
pthread_mutex_destroy(&pool->mutexBusy);
pthread_cond_destroy(&pool->Empty);
pthread_cond_destroy(&pool->Full);
free(pool);
pool = NULL;
return 0;
}
- 给线程池添加任务
void threadPoolAdd(ThreadPool * pool, void(*func)(void*), void*arg)
{
pthread_mutex_lock(&pool->mutexPool);
while(pool->queueSize == pool->queueCapacity &&
pool->shutdown == 0){
//队列满,阻塞生产者线程
pthread_cond_wait(&pool->Full, &pool->mutexPool);
}
if(pool->shutdown == 1){
//若需要销毁线程池
pthread_mutex_unlock(&pool->mutexPool);
return;
}
//添加任务
pool->taskQ[pool->queueEnd].func = func;
pool->taskQ[pool->queueEnd].arg = arg;
pool->queueEnd = (pool->queueEnd+1) % pool->queueCapacity;//循环
pool->queueSize++;
pthread_cond_signal(&pool->Empty);//有任务了,唤醒因空队列而阻塞的线程
pthread_mutex_unlock(&pool->mutexPool);
}
- 单个线程退出
void threadExit(ThreadPool *pool)
{
pthread_t tid = pthread_self();
for(int i = 0; i < pool->maxNum_Threads; i++){
if(pool->threadIDs[i] == tid){
pool->threadIDs[i] = 0;
printf("threadExit() called, %ld exitingn", tid);
break;
}
}
pthread_exit(NULL);//退出当前线程
}
完整代码
- 代码保存于我的GitHub仓库
- 代码含有错误检查
- 请在GUN/Linux环境下使用以下命令编译运行
gcc ThreadPool.c main.c -o ThreadPool -lpthread ./ThreadPool
- -lpthread是为了链接线程库
- main.c文件为测试demo
参考
- 《TLPI》
- 手写线程池 - C 语言版(爱编程的大丙)
- 大丙老师实在是太强了,关于线程池的更多细节可以去观看大丙老师的教学视频
