共计 4718 个字符,预计需要花费 12 分钟才能阅读完成。
导读 | 线程池也是多线程的处理方式。是将“生产者”线程提出任务添加到“任务队列”,然后一些线程自动完成“任务队列”上的任务。 |
多线程编程,创建一个线程,指定去完成某一个任务,等待线程的退出。虽然能够满足编程需求,但是当我们需要创建大量的线程的时候,在创建过程以及销毁线程的过程中可能会消耗大量的 CPU. 增加很大开销。如:文件夹的 copy、WEB 服务器的响应。
线程池就是用来解决类似于这样的一个问题的,可以降低频繁地创建和销毁线程所带来地开销。
线程池技术思路:一般采用预创建线程技术,也就是提前把需要用线程先创建一定数目。这些线程提前创建好了之后,“任务队列”里面假设没有任务,那么就让这些线程休眠,一旦有任务,就唤醒线程去执行任务,任务执行完了,也不需要去销毁线程,直到当你想退出或者是关机时,这个时候,那么你调用销毁线程池地函数去销毁线程。
线程完成任务之后不会销毁,而是自动地执行下一个任务。而且,当任务有很多,你可以有函数接口去增加线程数量,当任务较少时,你可以有函数接口去销毁部分线程。
如果,创建和销毁线程的时间对比执行任务的时间可以忽略不计,那么我们在这种情况下面也就没有必要用线程池。
“任务队列”是一个共享资源“互斥访问”
线程池本质上也是一个数据结构,需要一个结构体去描述它:
struct pthread_pool // 线程池的实现 | |
{ | |
// 一般会有如下成员 | |
// 互斥锁,用来保护这个“任务队列”pthread_mutex_t lock; // 互斥锁 | |
// 线程条件变量 表示“任务队列”是否有任务 | |
pthread_cond_t cond; // 条件变量 | |
bool shutdown; // 表示是否退出程序 bool:类型 false / true | |
// 任务队列(链表),指向第一个需要指向的任务 | |
// 所有的线程都从任务链表中获取任务 "共享资源" | |
struct task * task_list; | |
// 线程池中有多个线程,每一个线程都有 tid, 需要一个数组去保存 tid | |
pthread_t * tids; //malloc() | |
// 线程池中正在服役的线程数,当前线程个数 | |
unsigned int active_threads; | |
// 线程池任务队列最大的任务数量 | |
unsigned int max_waiting_tasks; | |
// 线程池任务队列上当前有多少个任务 | |
unsigned int cur_waiting_tasks; | |
//...... | |
}; | |
// 任务队列 (链表) 上面的任务结点,只要能够描述好一个任务就可以了,// 线程会不断地任务队列取任务 | |
struct task // 任务结点 | |
{// 1. 任务结点表示的任务,“函数指针”指向任务要执行的函数(cp_file) | |
void*(* do_task)(void * arg); | |
//2. 指针,指向任务指向函数的参数(文件描述符) | |
void * arg; | |
//3. 任务结点类型的指针,指向下一个任务 | |
struct task * next; | |
}; |
线程池框架代码如下,功能自填:
操作线程池所需要的函数接口:pthread_pool.c、pthread_pool.h
把“线程池”想象成一个外包公司,你需要去完成的就是操作线程池所提供的函数接口。
pthread_pool.c
#include "pthread_pool.h" | |
/* | |
init_pool: 线程池初始化函数,初始化指定的线程池中有 thread_num 个初始线程 | |
@pool:指针,指向您要初始化的那个线程池 | |
@threa_num: 您要初始化的线程池中开始的线程数量 | |
返回值:成功 0 | |
失败 -1 | |
*/ | |
int init_pool(pthread_pool * pool , unsigned int threa_num) | |
{ | |
// 初始化线程池的结构体 | |
// 初始化线程互斥锁 | |
pthread_mutex_init(&pool->lock, NULL); | |
// 初始化线程条件变量 | |
pthread_cond_init(&pool->cond, NULL); | |
pool->shutdown = false ;// 不退出 | |
pool->task_list = (struct task*)malloc(sizeof(struct task)); | |
pool->tids = (pthread_t *)malloc(sizeof(pthread_t) * MAX_ACTIVE_THREADS); | |
if(pool->task_list == NULL || pool->tids == NULL) | |
{perror("malloc memery error"); | |
return -1; | |
} | |
pool->task_list->next = NULL; | |
// 线程池中一开始初始化多少个线程来服役 | |
pool->active_threads = threa_num; | |
// 表示线程池中最多有多少个任务 | |
pool->max_waiting_tasks = MAX_WAITING_TASKS; | |
// 线程池中任务队列当前的任务数量 | |
pool->cur_waiting_tasks = 0; | |
// 创建 thread_num 个线程,并且让线程去执行任务调配函数,// 记录所有线程的 tid | |
int i = 0; | |
for(i = 0; i tids)[i], NULL, routine, (void*)pool); | |
if(ret != 0) | |
{perror("create thread error"); | |
return -1; | |
} | |
printf("[%lu]:[%s] ===> tids[%d]:[%lu]",pthread_self(), | |
__FUNCTION__, i , pool->tids[i]); | |
} | |
return 0; | |
} | |
/* | |
routine: 任务调配函数。所有线程开始都执行此函数,此函数会不断的从线程池的任务队列 | |
中取下任务结点,去执行。任务结点中包含“函数指针”h "函数参数" | |
*/ | |
void * routine(void * arg) | |
{ | |
//arg 表示你的线程池的指针 | |
while() | |
{ | |
// 获取线程互斥锁,lock | |
// 当线程池没有结束的时候,不断地从线程池的任务队列取下结点 | |
// 去执行。// 释放线程互斥锁,unlock | |
// 释放任务结点 | |
} | |
} | |
/* | |
destroy_pool: 销毁线程池,销毁前要保证所有的任务已经完成 | |
*/ | |
int destroy_pool(pthread_pool * pool) | |
{// 释放所有空间 等待任务执行完毕(join)。// 唤醒所有线程 | |
// 利用 join 函数回收每一个线程资源。} | |
/* | |
add_task:给任务队列增加任务,把 do_task 指向的任务 (函数指针) 和 | |
arg 指向的参数保存到一个任务结点,添加到 pool 任务队列中。@pool : 您要添加任务的线程池 | |
@do_task:您需要添加的任务(cp_file) | |
@arg:您要执行的任务的参数(文件描述符) | |
*/ | |
int add_task(pthread_pool *pool,void*(* do_task)(void * arg), void*arg) | |
{ | |
// 把第二个参数和第三个参数封装成 struct task | |
// 再把它添加到 pool->task 任务队列中去 | |
// 注意任务队列是一个共享资源 | |
// 假如任务后要唤醒等待的线程。} | |
// 如果任务多的时候,往线程池中添加线程 pthread_create | |
int add_threads(pthread_pool * pool, unsigned int num); | |
{ | |
// 新创建 num 个线程,让每一个线程去执行线程调配函数 | |
// 将每一个新创建的线程 tid,添加到 pool-> tids | |
} | |
// 如果任务少的时候,减少线程池中线程的数量 pthread_cancel join | |
int remove_threads(pthread_pool * pool, unsigned int num) | |
{ | |
// 用 pthread_cancel 取消 num 个线程 | |
// 利用 pthread_join 函数去回收资源。} |
pthread_pool.h
// 表示线程池中最多有多少个线程 | |
// 表示线程池中最多有多少个任务 | |
// 任务队列 (链表) 上面的任务结点,只要能够描述好一个任务就可以了,// 线程会不断地任务队列取任务 | |
struct task // 任务结点 | |
{// 1. 任务结点表示的任务,“函数指针”指向任务要执行的函数(cp_file) | |
void*(* do_task)(void * arg); | |
//2. 指针,指向任务指向函数的参数(文件描述符) | |
void * arg; | |
//3. 任务结点类型的指针,指向下一个任务 | |
struct task * next; | |
}; | |
struct pthread_pool // 线程池的实现 | |
{ | |
// 一般会有如下成员 | |
// 互斥锁,用来保护这个“任务队列”pthread_mutex_t lock; // 互斥锁 | |
// 线程条件变量 表示“任务队列”是否有任务 | |
pthread_cond_t cond; // 条件变量 | |
bool shutdown; // 表示是否退出程序 bool:类型 false / true | |
// 任务队列(链表),指向第一个需要指向的任务 | |
// 所有的线程都从任务链表中获取任务 "共享资源" | |
struct task * task_list; | |
// 线程池中有多个线程,每一个线程都有 tid, 需要一个数组去保存 tid | |
pthread_t * tids; //malloc() | |
// 线程池中正在服役的线程数,当前线程个数 | |
unsigned int active_threads; | |
// 线程池任务队列最大的任务数量 | |
unsigned int max_waiting_tasks; | |
// 线程池任务队列上当前有多少个任务 | |
unsigned int cur_waiting_tasks; | |
//...... | |
}; | |
/* | |
init_pool: 线程池初始化函数,初始化指定的线程池中有 thread_num | |
个初始线程 | |
@pool:指针,指向您要初始化的那个线程池 | |
@threa_num: 您要初始化的线程池中开始的线程数量 | |
返回值:成功 0 | |
失败 -1 | |
*/ | |
int init_pool(pthread_pool * pool , unsigned int threa_num); | |
/* | |
routine: 任务调配函数。所有线程开始都执行此函数,此函数会不断的从线程池的任务队列 | |
中取下任务结点,去执行。任务结点中包含“函数指针”h "函数参数" | |
*/ | |
void * routine(void * arg); | |
/* | |
destroy_pool: 销毁线程池,销毁前要保证所有的任务已经完成 | |
*/ | |
int destroy_pool(pthread_pool * pool); | |
/* | |
add_task:给任务队列增加任务,把 do_task 指向的任务 (函数指针) 和 | |
arg 指向的参数保存到一个任务结点,添加到 pool 任务队列中。@pool : 您要添加任务的线程池 | |
@do_task:您需要添加的任务(cp_file) | |
@arg:您要执行的任务的参数(文件描述符) | |
*/ | |
int add_task(pthread_pool *pool,void*(* do_task)(void * arg), void*arg); | |
// 如果任务多的时候,往线程池中添加线程 pthread_create | |
int add_threads(pthread_pool * pool, unsigned int num); | |
// 如果任务少的时候,减少线程池中线程的数量 pthread_cancel join | |
int remove_threads(pthread_pool * pool, unsigned int num); | |
正文完
星哥玩云-微信公众号
