在某些时候,系统中的一些任务可以并行操作来提高性能,此时如果频繁的创建&销毁线程又会造成大量地开销,因此需要借助线程池来操作。
(一)功能
- 添加任务到线程池
- 获取任务执行结果
- 维护线程池中线程状态
(二)设计思路
- 生产者&消费者模式
任务队列: 生产者与消费者公用的资源
生产者 : 将请求分解为多个小任务,并将其放入任务队列
消费者 : 线程池中的工作线程从任务队列中取任务进行处理
- 等待任务执行完毕
1)如何获得任务执行的结果?
将任务执行结果结构体的指针作为参数传给任务队列,在任务执行后直接将结果写到对应位置。
2)将任务A拆分为A1、A2、A3、A4、A5放入任务队列,如何知道任务已执行完毕?
对每个请求维护任务计数,当新增任务时计数++,当任务执行完毕时计数--;在每个请求后条件等待直到计数为0,表明该请求的任务执行完毕。
- 确定线程池 线程数&任务队列size
1)根据目前服务请求量计算线程个数
2)任务队列尽量设置够大
3)根据实际情况调整线程数
(三)结构
- 线程池管理
初始化若干工作线程&销毁工作线程
- 工作线程
如果队列不空,从人物队列取任务&执行任务
- 上层请求线程
拆分请求为多个任务&依次放入任务队列&等待任务完成&任务结果
- 任务队列
存放&提取任务,控制任务的先进先出
(四)数据结构
- 任务结构
序号 | 成员 | 类型 | 含义 |
1 | *function(void*, void*, int) | 函数指针 | 指向任务函数的指针 |
2 | *argument | void* | 指向任务函数参数的指针 |
3 | *control_task | control_task_t | 控制任务计数的结构变量 |
4 | task_type | int | 任务类型(根据不同类型,实现不同的任务) |
- 任务计数控制(control_task_t)
序号 | 成员 | 类型 | 含义 |
1 | task_count_mutex | pthread_mutex_t | 控制未完成任务计数的互斥锁 |
2 | task_count_cond | pthread_cond_t | 控制等待任务完成的条件锁 |
3 | unfinished_task_count | int | 未完成任务计数 |
- 线程池
序号 | 成员 | 类型 | 含义 |
1 | lock | pthread_mutex_t | 控制任务队列的互斥锁 |
2 | notify | pthread_cont_t | 控制任务队列的条件锁 |
3 | threads | pthread_t* | 工作线程 |
4 | queue | threadpool_task_t* | 任务队列 |
5 | thread_count | int | 工作线程数量 |
6 | queue_size | int | 任务队列大小 |
7 | head | int | 队头 |
8 | tail | int | 队尾 |
9 | count | int | pending任务数 |
10 | shutdown | int | 关闭线程池标记 |
11 | started | int | 运行工作线程 |
- 异常状态处理
值 | 含义 | |
threadpool_invalid | -1 | 线程池无效 |
threadpool_lock_failure | -2 | 线程池加锁失败 |
threadpool_queue_full | -3 | 任务队列满 |
thread_shutdown | -4 | 线程池已关闭 |
threadpool_thread_failture | -5 | 销毁线程时join线程失败 |
(五)实现
1)添加任务(threadpool_add)
互斥锁->任务结构入队尾->条件信号->解锁
2)工作线程(*threadpool_thread(void *threadpool))
取任务:锁->条件等队列不空->从队头取出任务->解锁
执行任务:(*function)(argument,control_task,task_type)
3)创建线程池(threadpool* threadpool_create(thread_count,queue_size,flag))
4)销毁线程池(threadpool_destroy(*pool,flags))
标记置位,等待所有工作线程执行完毕
(六)使用
1)任务分解&入任务队列
锁->未完成任务数++->解锁->threadpool_add放入任务
- 为什么要在放入任务前进行++操作,而不是放入后?
在放入后执行++操作可能造成计数不一致
2)任务实现
在任务执行完后,未完成任务数-- -> 如果未完成任务数==0,发送信号
3)搜集任务结果
锁->如果未完成任务数>0,则条件等待 ->解锁
(七)总结&优化
- 减小锁粒度
所有请求使用一个锁 => 每个请求拥有自己的锁
- 减小锁范围
锁addTask和count++ => 调整操作顺序,只锁count++