线程池是一种池式组件，通过创建和维护一定数量的线程，实现这些线程的重复使用，避免了频繁创建和销毁线程的开销，从而提升了性能
线程池的作用：
1.复用线程资源；
2.减少线程创建和销毁的开销；
3.可异步处理生产者线程的任务；
4.减少了多个任务（不是一个任务）的执行时间；
代码实现
一、结构体定义
1.任务结构体：
typedef struct task_s { void *next; // 指向下一个task handler_pt func; // 该task执行的函数 void *arg; // 函数的参数 }task_t;
2.队列结构体：
typedef struct task_queue_s { // task队列 void *head; // 一级指针，指向队列的第一个task结构体 void **tail; // 二级指针，指向队列的最后一个task结构体的第一个成员void *next指针 int block; // 阻塞标志 spinlock_t lock; // 自旋锁变量 pthread_mutex_t mutex; // 互斥锁变量 pthread_cond_t cond; // 条件变量 } task_queue_t;
3.线程池结构体：
struct threadpool_s { task_queue_t *task_queue; // task队列指针 atomic_int quit; // 原子变量 int thread_count; // 线程池中的线程数量 pthread_t *threads; // 线程句柄数组 };
二、资源创建和销毁
1.创建任务队列：
static task_queue_t *__taskqueue_create() { // 创建一个任务队列 int ret; task_queue_t *queue = (task_queue_t *)malloc(sizeof(task_queue_t)); // 回滚式编程，创建资源 if (queue) { ret = pthread_mutex_init(&queue->mutex, NULL); if (ret == 0) { ret = pthread_cond_init(&queue->cond, NULL); if (ret == 0) { // 全部资源创建成功 spinlock_init(&queue->lock); queue->head = NULL; // 队列为空 queue->tail = &queue->head; // tail是二级指针，大小是一个指针八字节,队列为空时指向head指针首地址 queue->block = 1; // 设置为阻塞 return queue; } pthread_mutex_destroy(&queue->mutex); // queue和init成功，但此{}内其它资源至少有一个创建失败 } free(queue); // queue创建成功，但此{}内其它资源没有全部创建成功，至少有一个失败 } return NULL; // queue创建失败 }
2.销毁任务队列：
static void __taskqueue_destroy(task_queue_t *queue) { // 销毁一个任务队列 task_t *task; // 先将任务队列中的任务全部销毁 while ((task = __pop_task(queue))) { free(task); } // 销毁任务队列的成员 spinlock_destroy(&queue->lock); pthread_cond_destroy(&queue->cond); pthread_mutex_destroy(&queue->mutex); free(queue); // 释放队列结构体空间 }
3.创建线程池：
static int __threads_create(threadpool_t *pool, size_t thread_count) { // 为线程池创建若干线程 pthread_attr_t attr; // 用于设置线程属性的变量，作为pthread_create的第二个参数 int ret; ret = pthread_attr_init(&attr); // 初始化线程属性变量 // 资源创建-->回滚式编程 if (ret == 0) { // 为线程句柄数组分配空间 pool->threads = (pthread_t *)malloc(sizeof(pthread_t) * thread_count); if (pool->threads) { // 创建若干线程开始执行__threadpool_worker函数 int i = 0; for (; i < thread_count; i++) { if (pthread_create(&pool->threads[i], &attr, __threadpool_worker, pool) != 0) { break; // 出现创建失败的情况 } } pool->thread_count = i; // 更新线程数量，因为可能中途创建失败，数量小于预期值 pthread_attr_destroy(&attr); // 销毁线程属性变量 if (i == thread_count) return 0; // 资源全部创建成功 __threads_terminate(pool); free(pool->threads); } ret = -1; // 为线程句柄数组分配空间失败 } return ret; // 初始化线程属性变量失败 } threadpool_t *threadpool_create(int thread_count) { // 创建一个线程池，参数是线程数量 threadpool_t *pool; pool = (threadpool_t *)malloc(sizeof(*pool)); // 为线程池分配内存 if (pool) { task_queue_t *queue = __taskqueue_create(); // 创建线程池的工作队列 if (queue) { pool->task_queue = queue; atomic_init(&pool->quit, 0); if (__threads_create(pool, thread_count) == 0) // 为线程池创建若干线程 return pool; __taskqueue_destroy(queue); // 线程创建失败则逐层释放已经创建的资源 } free(pool); } return NULL; }
4.回收线程池：
static void __threads_terminate(threadpool_t *pool) { // 回收线程池的所有线程 atomic_store(&pool->quit, 1); // 设置quit状态，使得不再有新的线程开始工作 __nonblock(pool->task_queue); // 设置队列为非阻塞，使得get_task的线程不会进入条件等待，直接退出 int i; for (i = 0; i < pool->thread_count; i++) { pthread_join(pool->threads[i], NULL); // 等待所有线程结束 } }
三、任务的添加、删除和执行
1.向任务队列中添加一个任务：
static inline void __add_task(task_queue_t *queue, void *task) { // 向任务队列中添加一个任务 // void *，不限定任务类型，只要该任务的结构体起始内存是一个用于链接下一个节点的指针 void **link = (void **)task; //转换task指向的类型从结构体变为指针：所指范围是8字节，也就是task_t结构体对象的前8个字节，此时task相当于指向其结构体的第一个成员next，与tail用法相同 *link = NULL; // (*link等价于next指针)将该task指向的next指针成员指向NULL // 修改共享变量queue时加锁 spinlock_lock(&queue->lock); *queue->tail = link; // 等价于 queue->tail->next = link, 因为二级指针tail实际上指向一个task_t结构体的第一个成员：next指针，link指向这个task的第一个成员next指针 queue->tail = link; // 实际上第一行的next指针和第二行的tail指针都保存着link也就是task的首地址 spinlock_unlock(&queue->lock); pthread_cond_signal(&queue->cond); // 添加一个任务后，广播唤醒处于等待状态的任务 } int threadpool_post(threadpool_t *pool, handler_pt func, void *arg) { // 向线程池添加一个任务 if (atomic_load(&pool->quit) == 1) // 如果线程池的状态是quit：已终止，则退出 return -1; task_t *task = (task_t *)malloc(sizeof(task_t)); if (!task) return -1; task->func = func; task->arg = arg; __add_task(pool->task_queue, task); return 0; }
2.从任务队列中取出一个任务：
static inline void *__pop_task(task_queue_t *queue) { // 从队列中取出一个task // 上锁 spinlock_lock(&queue->lock); if (queue->head == NULL) { // 队列为空 spinlock_unlock(&queue->lock); return NULL; } task_t *task; task = queue->head; // 取出第一个 void **link = (void **)task; // link指向task即head的下一个task queue->head = *link; // *link即link指向的task指针 if (queue->head == NULL) { // 判断取出一个节点后队列是否为空，很必要！ queue->tail = &queue->head; } spinlock_unlock(&queue->lock); return task; } static inline void *__get_task(task_queue_t *queue) { // 从队列中获取一个task，调用了pop_task task_t *task; // 使用while， 一定不能用if，如果被虚假唤醒，队列仍为空，返回的task == NULL while ((task = __pop_task(queue)) == NULL) { // 队列为空 // 上锁 ,一定要在while内此处上锁，如果在while外，则其它没有抢到锁的线程会直接获得一个未初始化的野指针task pthread_mutex_lock(&queue->mutex); if (queue->block == 0) { // 队列为非阻塞状态 pthread_mutex_unlock(&queue->mutex); return NULL; // 队列为空就直接返回 } // 队列为阻塞状态 pthread_cond_wait(&queue->cond, &queue->mutex); // pthread_cond_wait函数执行的内容： // 1.先进行unlock（&queue->mutex） // 2.再使当前线程休眠 // --- 接收到broadcast广播时唤醒线程 // 3.唤醒当前线程 // 4.加上lock（&queue->mutex） // 5.函数返回 // 6.pthread_cond_wait的执行并不是原子性的，所以需要使用锁 pthread_mutex_unlock(&queue->mutex); } return task; }
3.执行任务（每一个消费者线程所执行的函数）
static void *__threadpool_worker(void *arg) { // 线程池中的全部线程所执行的函数 threadpool_t *pool = (threadpool_t *)arg; // 传入一个线程池的指针 task_t *task; void *ctx; while (atomic_load(&pool->quit) == 0) { // 线程池没有停止运转（被销毁前停止、被终止前停止） task = (task_t *)__get_task(pool->task_queue); // 从线程池的任务队列中获取一个任务 if (!task) break; // 线程池被标记终止"quit"，get_task返回NULL handler_pt func = task->func; // 获取该任务所执行的函数 ctx = task->arg; // 任务执行函数所需的参数 free(task); // 任务被执行后销毁 func(ctx); // 执行任务函数 } return NULL; }
