- 线程池
- 1.API接口设计
- 2.线程池初始化：init_pool()
- 3.增加活跃线程：add_thread( )
- 4.删除活跃线程：remove_thread()
- 5.销毁线程池：destroy_pool()
线程池是一种多线程的处理形式，处理过程中将任务添加到任务队列中，然后创建线程后启动这些任务
注意：
1，任务队列中刚开始没有任何任务，是一个具有头结点的空链队列。
2，使用互斥锁来保护这个队列。
3，使用条件变量来代表任务队列中的任务个数的变化——将来如果主线程往队列中投放任务，那么可以通过条件变量来唤醒那些睡着了的线程。
4，通过一个公共开关——shutdown，来控制线程退出，进而销毁整个线程池。
1.API接口设计1、线程池相关的结构体
一个任务的结构体
任务节点，包含需要执行的函数及其参数，通过链表连成一个任务队列
struct task //任务节点
{
void *(*task)(void *arg); //任务函数
void *arg; //参数
struct task* next; //后继指针
};
//任务实例最终形成一条单向链表
线程池实例，包含一个线程池的所有信息
struct thread_pool //线程池
{
pthread_mutex_t lock; // 互斥锁，保护任务队列
pthread_cond_t cond; // 条件变量，同步所有线程
struct task *task_list; // 任务链队列指针
pthread_t *tids; // 线程 ID 存放位置
bool shutdown; // 线程池销毁标记
unsigned int waiting_tasks; // 任务链队列中等待的任务个数
unsigned int active_threads; // 当前活跃线程个数
}
//活跃线程个数可修改，但至少有 1 条活跃线程
2.线程池初始化：init_pool()
功能描述 创建一个新的线程池，包含threads_number个活跃线程pool:线程池指针 原型 bool init_pool(thread_pool *pool,unsigned int threads_number); 参数 threads_number:初始活跃线程个数(大于等于1) 返回值 成功返回true，失败返回false 备注 线程池最少线程个数为13.增加活跃线程：add_thread( )
功能描述 增加线程池中活跃线程的个数 原型 int add_thread(thread_pool *pool,unsigned int additional_threads); 参数 pool: 需要增加线程的线程池指针 additional_threads: 新增线程个数 返回值 >0: 实际新增线程个数 -1:失败4.删除活跃线程：remove_thread()
功能描述 删除线程池中活跃线程的个数pool:需要删除线程的线程池指针 原型 int remove_thread(thread_pool*pool, unsigned int removing_threads); 参数 removing_threads:要删除的线程个数。该参数设置为0时直接返回当前线程池线程总数，对线程池不造成任何其它影响 返回值 >0:当前线程池剩余线程个数 -1:失败 备注 1，线程池至少会存在1条活跃线程 2，如果被删除的线程正在执行任务，则将等待其完成任务之后删除5.销毁线程池：destroy_pool()
功能描述 阻塞等待所有任务完成，然后立即销毁整个线程池，释放所有资源和内存 原型 bool destroy_pool(thread_pool *pool); 参数 pool:将要销毁的线程池 返回值 成功返回true，失败返回false
thread_pool.h
#ifndef _THREAD_POOL_H_ #define _THREAD_POOL_H_ #include#include #include #include #include #include #include #include #define MAX_WAITING_TASKS 1000 #define MAX_ACTIVE_THREADS 20 //任务结构体 struct task { //任务接口 void *(*task)(void *arg); void *arg; //任务链表 struct task *next; }; //线程池的组成部分 typedef struct thread_pool { pthread_mutex_t lock; //互斥锁 pthread_cond_t cond; //条件变量 struct task *task_list; //任务结构体 pthread_t *tids; //存放tid unsigned waiting_tasks; //等待任务数 unsigned active_threads; //活跃线程数 bool shutdown; //关闭标志 }thread_pool; //初始化线程池 bool init_pool(thread_pool *pool, unsigned int threads_number); //增加任务 bool add_task(thread_pool *pool, void *(*task)(void *arg), void *arg); //增加线程 int add_thread(thread_pool *pool, unsigned int additional_threads_number); //删除线程 int remove_thread(thread_pool *pool, unsigned int removing_threads_number); //销毁线程池 bool destroy_pool(thread_pool *pool); //线程函数 void *routine(void *arg); #endif
thread_pool.c
void handler(void *arg)
{
pthread_mutex_unlock((pthread_mutex_t *)arg);
}
//线程函数
void *routine(void *arg)
{
thread_pool *pool = (thread_pool *)arg;
struct task *p;
while(1)
{
//防止死锁
pthread_cleanup_push(handler, (void *)&pool->lock);
//加锁
pthread_mutex_lock(&pool->lock);
//若没有任务，则睡眠
while(pool->waiting_tasks == 0 && !pool->shutdown)
{
pthread_cond_wait(&pool->cond, &pool->lock);
}
//若没有任务，且线程池关闭，就退出线程（任务执行完毕）
if(pool->waiting_tasks == 0 && pool->shutdown == true)
{
pthread_mutex_unlock(&pool->lock);
pthread_exit(NULL);
}
//取节点，删除节点
p = pool->task_list->next;
pool->task_list->next = p->next;
//任务数-1
pool->waiting_tasks--;
//解锁
pthread_mutex_unlock(&pool->lock);
pthread_cleanup_pop(0);
//设置线程不可取消
pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);
//执行任务
p->task(p->arg);
//设置线程可取消
pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
free(p);
}
pthread_exit(NULL);
}
//初始化线程池
bool init_pool(thread_pool *pool, unsigned int threads_number)
{
//初始化互斥锁和条件变量
pthread_mutex_init(&pool->lock, NULL);
pthread_cond_init(&pool->cond, NULL);
pool->shutdown = false; //标志为关
pool->task_list = malloc(sizeof(struct task));
pool->tids = malloc(sizeof(pthread_t) * MAX_ACTIVE_THREADS);
if(pool->task_list == NULL || pool->tids == NULL)
{
perror("allocate memory error");
return false;
}
//初始化链表
pool->task_list->next = NULL;
pool->waiting_tasks = 0; //开始没有任务
pool->active_threads = threads_number;
int i;
for(i=0; iactive_threads; i++)
{
if(pthread_create(&((pool->tids)[i]), NULL,
routine, (void *)pool) != 0)
{
perror("create threads error");
return false;
}
}
return true;
}
//增加任务
bool add_task(thread_pool *pool,
void *(*task)(void *arg), void *arg)
{
//新增节点
struct task *new_task = malloc(sizeof(struct task));
if(new_task == NULL)
{
perror("allocate memory error");
return false;
}
new_task->task = task;
new_task->arg = arg;
new_task->next = NULL;
pthread_mutex_lock(&pool->lock);
//任务过多，无法添加
if(pool->waiting_tasks >= MAX_WAITING_TASKS)
{
pthread_mutex_unlock(&pool->lock);
fprintf(stderr, "too many tasks.n");
free(new_task);
return false;
}
//将新增节点插入链表，尾插法
struct task *tmp = pool->task_list;
while(tmp->next != NULL)
tmp = tmp->next;
tmp->next = new_task;
//任务数+1
pool->waiting_tasks++;
pthread_mutex_unlock(&pool->lock);
pthread_cond_signal(&pool->cond);
return true;
}
//增加线程
int add_thread(thread_pool *pool, unsigned additional_threads)
{
if(additional_threads == 0)
return 0;
//总线程数 = 活跃线程数 + 新增线程数
unsigned total_threads =
pool->active_threads + additional_threads;
int i, actual_increment = 0; //实际新增线程数
for(i = pool->active_threads;
i < total_threads && i < MAX_ACTIVE_THREADS;
i++)
{
if(pthread_create(&((pool->tids)[i]),
NULL, routine, (void *)pool) != 0)
{
perror("add threads error");
if(actual_increment == 0)
return -1;
break;
}
actual_increment++;
}
//活跃线程数 += 实际新增线程数
pool->active_threads += actual_increment;
//返回实际新增线程数
return actual_increment;
}
//删除线程
int remove_thread(thread_pool *pool, unsigned int removing_threads)
{
//如果参数为0，则可以查看当前活跃线程数
if(removing_threads == 0)
return pool->active_threads;
//还剩的线程数 = 活跃线程数 - 要删除的线程数
int remain_threads = pool->active_threads - removing_threads;
remain_threads = remain_threads>0 ? remain_threads:1;
int i;
for(i=pool->active_threads-1; i>remain_threads-1; i--)
{
//取消线程
errno = pthread_cancel(pool->tids[i]);
if(errno != 0)
break;
}
if(i == pool->active_threads-1)
return -1;
else
{
pool->active_threads = i+1;
return i+1;
}
}
//销毁线程池
bool destroy_pool(thread_pool *pool)
{
//关闭标志变为真
pool->shutdown = true;
//广播通知睡眠的线程
pthread_cond_broadcast(&pool->cond);
int i;
for(i=0; iactive_threads; i++)
{
//回收线程
errno = pthread_join(pool->tids[i], NULL);
if(errno != 0)
{
printf("join tids[%d] error: %sn",
i, strerror(errno));
}
else
printf("[%u] is joinedn", (unsigned)pool->tids[i]);
}
free(pool->task_list);
free(pool->tids);
free(pool);
return true;
}
参考：
线程池
