多线程之线程池-程序员充电站

文章目录

线程池的优点
线程池的四大核心（组件）
- 任务队列
- 工作线程
- 管理者
- 控制开关
实现步骤
- 创建线程池
- 工作线程的行为
- 添加任务
- 销毁线程池
关键数据结构
示例
注意事项

如果每个任务创建新线程，任务结束即销毁线程，资源的消耗和开销也不小
固定数量的线程处理多个任务，实现线程复用

线程池的优点

避免频繁创建/销毁线程的开销
控制并发线程数量，避免资源耗尽
提高响应速度（任务无需等待线程创建）

线程池的四大核心（组件）

任务队列（待办清单）
工作线程（员工团队）
管理者（协调工具）
控制开关

任务队列

使用链表实现的 FIFO 队列
每个任务包含：
- 任务函数指针（要执行的操作）
- 参数（执行所需数据）
- 指向下一个任务的指针

structTask{void(*function)(void*);// 任务函数（做什么）void*arg;// 参数（用什么做）structTask*next;// 下一个任务};

工作线程

固定数量的线程（如 CPU 核心数 × 2）
每个线程循环：等待任务 → 取出任务 → 执行任务

whileTrue:等待任务->取出任务->执行任务

管理者

互斥锁（mutex）：保护共享资源（任务队列）
条件变量（cond）：线程间通信，通知新任务到达

控制开关

shutdown 标志：安全关闭信号，优雅终止线程池

实现步骤

第一步：创建线程池
第二步：工作线程的行为
第三步：添加任务
第四步：销毁线程池

创建线程池

ThreadPool*thread_pool_create(){// 分配内存// 初始化互斥锁和条件变量// 初始化任务队列// 设置 shutdown 标志为 0// 创建固定数量的工作线程}

工作线程的行为

void*worker(void*arg){while(1){上锁;while(无任务且不关闭){等待条件变量;}if(需要关闭)退出;取出任务;解锁;执行任务;释放任务内存;}}

添加任务

intthread_pool_add_task(ThreadPool*pool,函数指针,参数){// 创建新任务节点// 上锁// 加入队列尾部// 发送条件信号// 解锁}

销毁线程池

voidthread_pool_destroy(ThreadPool*pool){// 设置 shutdown 标志为 1// 广播通知所有线程// 等待所有线程退出// 清理剩余任务// 销毁锁和条件变量// 释放内存}

关键数据结构

typedefstructTask{void(*function)(void*);// 任务函数void*arg;// 参数structTask*next;// 下一个任务}task_t;typedefstructThreadPool{pthread_mutex_tmutex;// 互斥锁pthread_cond_tcond;// 条件变量pthread_tthreads[THREAD_NUM];// 线程数组task_t*task_head;// 任务队列头task_t*task_tail;// 任务队列尾intshutdown;// 关闭标志}tpool_t;

示例

#include<pthread.h>#include<stdlib.h>#include<stdio.h>#include<unistd.h>#defineTHREAD_NUM4//任务队列typedefstructTask{void(*function)(void*);void*arg;structTask*next;}task_t;//线程池结构体typedefstructthread_pool{pthread_mutex_tmutex;pthread_cond_tcond;pthread_ttid[THREAD_NUM];task_t*task_head;task_t*task_tail;intshutdown;//线程池的开关，0代表使用，1代表销毁}tpool_t;//工作线程void*worker(void*arg){tpool_t*pool=(tpool_t*)arg;while(1){//1. 锁定临界资源pthread_mutex_lock(&pool->mutex);//2. 等待任务while(pool->task_head==NULL&&!pool->shutdown){pthread_cond_wait(&pool->cond,&pool->mutex);}//3. 检查线程开关if(pool->shutdown){pthread_mutex_unlock(&pool->mutex);pthread_exit(NULL);}//4. 取出任务task_t*task=pool->task_head;if(task==NULL){pthread_mutex_unlock(&pool->mutex);continue;}//5. 列新任务队列pool->task_head=task->next;if(pool->task_head==NULL){pool->task_tail=NULL;}//6. 解锁pthread_mutex_unlock(&pool->mutex);//7. 执行任务task->function(task->arg);free(task);}returnNULL;}//创建线程池tpool_t*thread_pool_init(){//1. 为线程池结构体申请内存tpool_t*pool=malloc(sizeof(tpool_t));if(pool==NULL){returnNULL;}//2. 初始化互斥量，条件变量pthread_mutex_init(&pool->mutex,NULL);pthread_cond_init(&pool->cond,NULL);//3. 初始化队列pool->task_head=NULL;pool->task_tail=NULL;//4. 初始化线程池的开关pool->shutdown=0;//5. 创建工作线程for(inti=0;i<THREAD_NUM;i++){pthread_create(&pool->tid[i],NULL,worker,pool);}returnpool;}//添加任务队列 添加成功返回0，失败返回 -1intthread_pool_add_task(tpool_t*pool,void(*function)(void*),void*arg){//1. 创建新任务task_t*new_task=malloc(sizeof(task_t));if(new_task==NULL){return-1;}new_task->function=function;new_task->arg=arg;new_task->next=NULL;//2. 锁定临界资源pthread_mutex_lock(&pool->mutex);//3. 加入队列if(pool->task_tail==NULL){pool->task_head=pool->task_tail=new_task;}else{pool->task_tail->next=new_task;pool->task_tail=new_task;}//4. 通知工作线程，并解锁pthread_cond_signal(&pool->cond);pthread_mutex_unlock(&pool->mutex);return0;}//销毁线程池intthread_pool_destroy(tpool_t*pool){//1. 关闭线程池pthread_mutex_lock(&pool->mutex);pool->shutdown=1;pthread_mutex_unlock(&pool->mutex);//2. 唤醒所有等待的线程pthread_cond_broadcast(&pool->cond);//3. 等待所有线程退出for(inti=0;i<THREAD_NUM;i++){pthread_join(pool->tid[i],NULL);}//4. 释放各种资源task_t*p=pool->task_head;while(p!=NULL){task_t*temp=p;p=p->next;free(temp);}pthread_mutex_destroy(&pool->mutex);pthread_cond_destroy(&pool->cond);free(pool);return0;}/*线程池功能测试*/voidcook(void*arg){intorder_id=*(int*)arg;printf("厨师%lu 开始制作订单%d\n",pthread_self(),order_id);sleep(1);// 模拟烹饪时间printf("订单%d 完成！\n",order_id);}intmain(intargc,constchar*argv[]){// 创建线程池tpool_t*pool=thread_pool_init();if(pool==NULL){}// 添加任务intorders[10];for(inti=0;i<10;i++){orders[i]=1000+i;thread_pool_add_task(pool,cook,&orders[i]);}sleep(4);// 销毁线程池thread_pool_destroy(pool);return0;}