线程池的实现-阿里云开发者社区

线程池的重点

为什么使用线程池
线程池的构成，其中最重要的是队列的设计（结合线程池网络编程的实例）。

1、线程池的概念

1.1、为什么要使用线程池

对于某些特别耗时的任务，严重影响了该线程处理其他任务，需要将这些任务交给其他线程异步处理，但是频繁创建和销毁线程也会给系统带来大量开销。线程池的出现可以很好解决上述问题，线程池中的线程数量需要我们在线程资源的开销和 cpu 核心间进行平衡选择，要做到既能减少线程创建和销毁的开销，又能最大限度的提高 cpu 利用率。

1.2、线程池的作用

复用线程资源，减少线程创建和销毁的开销
可异步处理生产者线程的任务。

1.3、线程池的构成

线程池其实是一个生产者和消费者模型，生产者线程发布任务，将任务放入队列中。线程池中的空闲消费者线程从队列中取出任务，并执行任务。

线程池的工作原理

如图，线程池有三个核心组件

生产者线程：发布任务
队列：存储任务结点，其中包括任务的上下文、任务的执行函数等
线程池（消费者）：取出任务，执行任务。此过程涉及到线程调度的问题，即从无到有唤醒，从有到无休眠。

1.4、线程池线程数量的平衡选择

线程池线程数量的经验公式：（io等待时间 + cpu运算时间）* 核心数 / cpu运算时间

io 密集型：2 * cpu 核心数 + 有效磁盘数（2）
cpu 密集型：与 cpu 核心数一致

2、线程池的实现

2.1、线程池的组件

生产者线程：主线程
任务队列

// 任务结点
 typedef struct task_s {
  handler_pt func; // 任务的执行函数
  void * arg;      // 任务的上下文
 } task_t;
 // 任务队列
 typedef struct task_queue_s {
  uint32_t head;  // 队列的头指针
  uint32_t tail;  // 队列的尾指针
  uint32_t count; // 队列中的任务结点数量
  task_t *queue;  // 队列数组
 } task_queue_t;

线程池

struct thread_pool_t {
  pthread_mutex_t mutex;
  pthread_cond_t condition;
  pthread_t *threads;
  task_queue_t task_queue;
  int closed;     // 退出标记
  int started;    // 当前运行的线程数
  int thrd_count; // 线程的数量
  int queue_size; // 队列的长度，设置数组，一次性分配内存，不使用内存池
 };

2.2、线程池的接口

2.2.1、创建线程池

线程池的线程数量：根据经验公式计算得出
队列的长度：线程的栈空间固定大小(默认8M)，由线程数量和线程栈空间大小共同决定

thread_pool_t* thread_pool_create(int thrd_count, int queue_size) {
     thread_pool_t *pool;
     if (thrd_count <= 0 || queue_size <= 0) {
         return NULL;
     }
     // 线程池分配内存
     pool = (thread_pool_t*) malloc(sizeof(*pool));
     if (pool == NULL) {
         return NULL;
     }
     // 初始化线程池
     // 为什么不直接用thrd_count赋值？而是选择从0开始计数
     // 每当成功创建1个线程，计数+1，避免线程创建失败造成计数混乱
     pool->thrd_count = 0;  
     pool->queue_size = queue_size;
     pool->task_queue.head = 0;
     pool->task_queue.tail = 0;
     pool->task_queue.count = 0;
     pool->started = pool->closed = 0;
     // 创建任务队列
     pool->task_queue.queue = (task_t*)malloc(sizeof(task_t)*queue_size);
     if (pool->task_queue.queue == NULL) {
         // TODO: free pool
         return NULL;
     }
     // 创建线程
     pool->threads = (pthread_t*) malloc(sizeof(pthread_t) * thrd_count);
     if (pool->threads == NULL) {
         // TODO: free pool
         return NULL;
     }
     // 依次创建好线程
     for (int i = 0; i < thrd_count; ++i) {
         if (pthread_create(&(pool->threads[i]), NULL, thread_worker, (void*)pool) != 0) {
             // TODO: free pool
             return NULL;
         }
         pool->thrd_count++;
         pool->started++;
     }
     return pool;
 }
 // 线程池中的线程（消费者）该干的事儿
 static void* thread_worker(void *thrd_pool) {
     thread_pool_t *pool = (thread_pool_t*)thrd_pool;
     task_queue_t *que;
     task_t task;
     for (;;) {
         pthread_mutex_lock(&(pool->mutex));
         que = &pool->task_queue;
         // 虚假唤醒问题
         // while 判断：没有任务而且线程池没有关闭
         while (que->count == 0 && pool->closed == 0) {
             // pthread_mutex_unlock(&(pool->mutex))
             // 阻塞在 condition
             // ===================================
             // 解除阻塞
             // pthread_mutex_lock(&(pool->mutex));
             pthread_cond_wait(&(pool->condition), &(pool->mutex));
         }
         // 线程池关闭
         if (pool->closed == 1) break;
         // 获取任务
         task = que->queue[que->head];
         que->head = (que->head + 1) % pool->queue_size;
         que->count--;
         pthread_mutex_unlock(&(pool->mutex));
         // 执行任务
         (*(task.func))(task.arg);
     }
     // 销毁该线程
     pool->started--;
     pthread_mutex_unlock(&(pool->mutex));
     pthread_exit(NULL);
     return NULL;
 }

虚假唤醒

上述代码注释中提到了虚假唤醒，那么，什么是虚假唤醒？

虚假唤醒指的是在多线程环境下，多个线程等待在同一个条件上。当条件满足时，可能唤醒多个线程。如果这个资源只能由一个线程获得，剩余线程无法获得该资源。对于无法获得资源的线程来说，这种唤醒是无意义的，这种现象称为虚假唤醒。

虚假唤醒产生的原因

信号中断导致的问题（Linux2.6以后已解决）
pthread_cond_signal，至少唤醒1个线程，即可能同时唤醒多个线程。

为避免虚假唤醒的发生，每个被唤醒的线程都需要再检查一次条件是否满足。如果不满足，应该继续睡眠；只有满足了才能往下执行。因此，需要把条件变量的判断从 if 判断换成 while 判断，此时，一次只能唤醒一个线程。

2.2.2、销毁线程池

标记线程池退出
通知所有线程

int thread_pool_destroy(thread_pool_t *pool) {
     if (pool == NULL) {
         return -1;
     }
     // 阻止产生新的任务
     if (pthread_mutex_lock(&(pool->mutex)) != 0) {
         return -2;
     }
     // 判断是否已经退出，防止重复释放空间
     if (pool->closed) {
         thread_pool_free(pool);
         return -3;
     }
     // 标记线程池退出
     pool->closed = 1;
     // 唤醒所有阻塞在cond上的线程，并释放互斥锁
     if (pthread_cond_broadcast(&(pool->condition)) != 0 || 
             pthread_mutex_unlock(&(pool->mutex)) != 0) {
         thread_pool_free(pool);
         return -4;
     }
     // 等待所有线程退出
     wait_all_done(pool);
     // 释放线程池空间
     thread_pool_free(pool);
     return 0;
 }

2.2.3、生产者抛出任务

构造任务
放入队列
唤醒线程

int thread_pool_post(thread_pool_t *pool, handler_pt func, void *arg) {
     if (pool == NULL || func == NULL) {
         return -1;
     }
     task_queue_t *task_queue = &(pool->task_queue);
     if (pthread_mutex_lock(&(pool->mutex)) != 0) {
         return -2;
     }
     // 判断线程池是否关闭
     if (pool->closed) {
         pthread_mutex_unlock(&(pool->mutex));
         return -3;
     }
     // 判断任务队列是否已满
     if (task_queue->count == pool->queue_size) {
         pthread_mutex_unlock(&(pool->mutex));
         return -4;
     }
     // 1、主线程（生产者线程）构造任务，放入任务队列
     // 队列的操作，使用自旋锁
     task_queue->queue[task_queue->tail].func = func;
     task_queue->queue[task_queue->tail].arg = arg;
     task_queue->tail = (task_queue->tail + 1) % pool->queue_size;
     task_queue->count++;
     // 2、唤醒线程池中的线程
     if (pthread_cond_signal(&(pool->condition)) != 0) {
         pthread_mutex_unlock(&(pool->mutex));
         return -5;
     }
     pthread_mutex_unlock(&(pool->mutex));
     return 0;
 }

2.3、实例

// threadpool.h
 #ifndef _THREAD_POOL_H 
 #define _THREAD_POOL_H
 typedef struct thread_pool_t thread_pool_t;
 typedef void (*handler_pt) (void *);
 // 创建线程池
 thread_pool_t *thread_pool_create(int thrd_count, int queue_size);
 // 抛出任务
 int thread_pool_post(thread_pool_t *pool, handler_pt func, void *arg);
 // 销毁线程池
 int thread_pool_destroy(thread_pool_t *pool);
 // 等待所有线程的退出
 int wait_all_done(thread_pool_t *pool);
 #endif
 // threadpool.c
 #include <pthread.h>
 #include <stdint.h>
 #include <stddef.h>
 #include <stdlib.h>
 #include "thrd_pool.h"
 // 任务结点
 typedef struct task_s {
     handler_pt func;    // 任务的执行函数
     void * arg;         // 任务的上下文
 } task_t;
 // 任务队列
 typedef struct task_queue_s {
     uint32_t head;  // 队列的头指针
     uint32_t tail;  // 队列的尾指针
     uint32_t count; // 队列中的任务结点数量
     task_t *queue;  // 队列数组
 } task_queue_t;
 // 线程池
 struct thread_pool_t {
     pthread_mutex_t mutex;
     pthread_cond_t condition;
     pthread_t *threads;
     task_queue_t task_queue;
     int closed;     // 退出标记
     int started;    // 当前运行的线程数
     int thrd_count; // 线程的数量
     int queue_size; // 队列的长度，设置数组，一次性分配内存，不使用内存池
 };
 static void * thread_worker(void *thrd_pool);
 static void thread_pool_free(thread_pool_t *pool);
 // 创建线程池
 thread_pool_t* thread_pool_create(int thrd_count, int queue_size) {
     thread_pool_t *pool;
     if (thrd_count <= 0 || queue_size <= 0) {
         return NULL;
     }
     // 线程池分配内存
     pool = (thread_pool_t*) malloc(sizeof(*pool));
     if (pool == NULL) {
         return NULL;
     }
     // 初始化线程池
     // 为什么不直接用thrd_count赋值？而是选择从0开始计数
     // 每当成功创建1个线程，计数+1，避免线程创建失败造成计数混乱
     pool->thrd_count = 0;  
     pool->queue_size = queue_size;
     pool->task_queue.head = 0;
     pool->task_queue.tail = 0;
     pool->task_queue.count = 0;
     pool->started = pool->closed = 0;
     // 创建任务队列
     pool->task_queue.queue = (task_t*)malloc(sizeof(task_t)*queue_size);
     if (pool->task_queue.queue == NULL) {
         // TODO: free pool
         return NULL;
     }
     // 创建线程
     pool->threads = (pthread_t*) malloc(sizeof(pthread_t) * thrd_count);
     if (pool->threads == NULL) {
         // TODO: free pool
         return NULL;
     }
     // 依次创建好线程
     int i = 0;
     for (; i < thrd_count; ++i) {
         if (pthread_create(&(pool->threads[i]), NULL, thread_worker, (void*)pool) != 0) {
             // TODO: free pool
             return NULL;
         }
         pool->thrd_count++;
         pool->started++;
     }
     return pool;
 }
 // 生产者抛出任务
 int thread_pool_post(thread_pool_t *pool, handler_pt func, void *arg) {
     if (pool == NULL || func == NULL) {
         return -1;
     }
     task_queue_t *task_queue = &(pool->task_queue);
     if (pthread_mutex_lock(&(pool->mutex)) != 0) {
         return -2;
     }
     // 判断线程池是否关闭
     if (pool->closed) {
         pthread_mutex_unlock(&(pool->mutex));
         return -3;
     }
     // 判断任务队列是否已满
     if (task_queue->count == pool->queue_size) {
         pthread_mutex_unlock(&(pool->mutex));
         return -4;
     }
     // 1、主线程（生产者线程）构造任务，放入任务队列
     // 队列的操作，使用自旋锁
     task_queue->queue[task_queue->tail].func = func;
     task_queue->queue[task_queue->tail].arg = arg;
     task_queue->tail = (task_queue->tail + 1) % pool->queue_size;
     task_queue->count++;
     // 2、唤醒线程池中的线程
     if (pthread_cond_signal(&(pool->condition)) != 0) {
         pthread_mutex_unlock(&(pool->mutex));
         return -5;
     }
     pthread_mutex_unlock(&(pool->mutex));
     return 0;
 }
 // 释放线程池空间
 static void thread_pool_free(thread_pool_t *pool) {
     if (pool == NULL || pool->started > 0) {
         return;
     }
     if (pool->threads) {
         free(pool->threads);
         pool->threads = NULL;
         pthread_mutex_lock(&(pool->mutex));
         pthread_mutex_destroy(&pool->mutex);
         pthread_cond_destroy(&pool->condition);
     }
     if (pool->task_queue.queue) {
         free(pool->task_queue.queue);
         pool->task_queue.queue = NULL;
     }
     free(pool);
 }
 // 等待所有线程的退出
 int wait_all_done(thread_pool_t *pool) {
     int i, ret = 0;
     for (i = 0; i < pool->thrd_count; i++) {
         if (pthread_join(pool->threads[i], NULL) != 0) {
             ret = 1;
         }
     }
     return ret;
 }
 // 3、销毁线程池
 int thread_pool_destroy(thread_pool_t *pool) {
     if (pool == NULL) {
         return -1;
     }
     // 阻止产生新的任务
     if (pthread_mutex_lock(&(pool->mutex)) != 0) {
         return -2;
     }
     // 判断是否已经退出，防止重复释放空间
     if (pool->closed) {
         thread_pool_free(pool);
         return -3;
     }
     // 标记线程池退出
     pool->closed = 1;
     // 让所有阻塞在cond上的线程唤醒，并释放互斥锁
     if (pthread_cond_broadcast(&(pool->condition)) != 0 || 
             pthread_mutex_unlock(&(pool->mutex)) != 0) {
         thread_pool_free(pool);
         return -4;
     }
     // 等待所有线程退出
     wait_all_done(pool);
     thread_pool_free(pool);
     return 0;
 }
 // 线程池中的线程（消费者）该干的事儿
 static void* thread_worker(void *thrd_pool) {
     thread_pool_t *pool = (thread_pool_t*)thrd_pool;
     task_queue_t *que;
     task_t task;
     for (;;) {
         pthread_mutex_lock(&(pool->mutex));
         que = &pool->task_queue;
         // 虚假唤醒问题
         // while 判断：没有任务而且线程池没有关闭
         while (que->count == 0 && pool->closed == 0) {
             // pthread_mutex_unlock(&(pool->mutex))
             // 阻塞在 condition
             // ===================================
             // 解除阻塞
             // pthread_mutex_lock(&(pool->mutex));
             pthread_cond_wait(&(pool->condition), &(pool->mutex));
         }
         // 线程池关闭
         if (pool->closed == 1) break;
         // 获取任务
         task = que->queue[que->head];
         que->head = (que->head + 1) % pool->queue_size;
         que->count--;
         pthread_mutex_unlock(&(pool->mutex));
         // 执行任务
         (*(task.func))(task.arg);
     }
     // 销毁该线程
     pool->started--;
     pthread_mutex_unlock(&(pool->mutex));
     pthread_exit(NULL);
     return NULL;
 }
 // main.c
 #include <stdio.h>
 #include <stdlib.h>
 #include <pthread.h>
 #include <unistd.h>
 #include "thrd_pool.h"
 int nums = 0;
 int done = 0;
 pthread_mutex_t lock;
 void do_task(void *arg) {
     usleep(10000);
     pthread_mutex_lock(&lock);
     done++;
     printf("doing %d task\n", done);
     pthread_mutex_unlock(&lock);
 }
 int main(int argc, char **argv) {
     int threads = 8;
     int queue_size = 256;
     if (argc == 2) {
         threads = atoi(argv[1]);
         if (threads <= 0) {
             printf("threads number error: %d\n", threads);
             return 1;
         }
     } else if (argc > 2) {
         threads = atoi(argv[1]);
         queue_size = atoi(argv[1]);
         if (threads <= 0 || queue_size <= 0) {
             printf("threads number or queue size error: %d,%d\n", threads, queue_size);
             return 1;
         }
     }
     thread_pool_t *pool = thread_pool_create(threads, queue_size);
     if (pool == NULL) {
         printf("thread pool create error!\n");
         return 1;
     }
     while (thread_pool_post(pool, &do_task, NULL) == 0) {
         pthread_mutex_lock(&lock);
         nums++;
         pthread_mutex_unlock(&lock);
     }
     printf("add %d tasks\n", nums);
     wait_all_done(pool);
     printf("did %d tasks\n", done);
     thread_pool_destroy(pool);
     return 0;
 }

3、线程池网络编程

3.1、reactor 的问题

reactor

在一个事件循环中，可以处理多个就绪事件。这些就绪事件在 reactor 模型中是串行执行的，一个事件处理若耗时较长，会延迟其他同时触发的事件的处理。

void eventloop_once(reactor_t * r, int timeout) {
     int n = epoll_wait(r->epfd, r->fire, MAX_EVENT_NUM, timeout);
     for (int i = 0; i < n; ++i) {
         struct epoll_event *e = &r->fire[i];
         int mask = e->events;
         // 用 io 函数捕获具体的错误信息
         if (e->events & EPOLLERR) mask |= EPOLLIN | EPOLLOUT;
         // 用 io 函数捕获断开的具体信息
         if (e->events & EPOLLHUP) mask |= EPOLLIN | EPOLLOUT;
         event_t *et = (event_t*) e->data.ptr;
         // 处理读事件
         if (mask & EPOLLIN) {
             if (et->read_fn) {
                 et->read_fn(et->fd, EPOLLIN, et);
             }    
         }
         // 处理写事件
         if (mask & EPOLLOUT) {
             if (et->write_fn) {
                 et->write_fn(et->fd, EPOLLOUT, et);
             }       
             else {
                 uint8_t * buf = buffer_write_atmost(&et->out);
                 event_buffer_write(et, buf, buffer_len(&et->out));
             }
         }
     }
 }

接下来，介绍 nginx, redis, skynet 线程池实现形式，都是对 reactor 模型的优化。

3.2、nginx 线程池

为什么使用线程池？

nginx 是磁盘 io 密集型。业务逻辑是处理文件缓冲，需要磁盘 io 操作，耗费大量的时间。

什么情况下使用线程池？

nginx 线程池作用 compute 阶段，将业务逻辑交给线程池处理。

怎样使用线程池？

nginx 的线程池有两个消息队列，当线程池处理完任务后，将处理的结果放入完成消息队列，以事件的方式通知主线程，主线程从完成消息队列中取出结果发送给客户端。

nginx线程池设计原理

值得注意的是，nginx 不推荐线程池的方式，可以采用替代的操作 sendfile, directio解决磁盘 io 的问题。

3.3、redis 线程池

为什么使用线程池？

redis 是网络 io 密集型，需要同时处理多条并发请求，存在读写 io 的问题（请求大量数据，写日志业务等）。

什么情况下使用线程池？

redis 线程池作用读写 io 阶段，即 read, decode 和 encode, send 阶段。主线程处理业务逻辑，之所以用单线程处理业务，是因为 redis 采用高效的数据结构，其业务逻辑处理较快。

redis 线程池作用阶段

怎样使用线程池？

主线程拥有两个全局队列clients_pending_read和clients_pending_write，每个 io 线程（主线程同时也是 io 线程）拥有一个专属队列 io_threads_list[id]。主线程既作为生产者，产生任务；又作为消费者，获取任务执行。

首先，主线程将一次循环的所有就绪的读事件收集到自己的全局任务队列clients_pending_read中，再把每个事件负载均衡地分配到每个 io 线程的专属任务队列中。一次事件循环中不会出现同名 fd，不同的 fd 分配到每个 io 线程各自的队列中，避免了多个 io 线程同时从全局队列中取数据，因此，不需要加锁操作。

接下来，io 线程从自己的专属队列中取出任务，（除主线程外）并发执行 read 和 decode 操作。主线程将解析后的任务做 compute 操作。最后，io 线程（包括主线程）并发执行 encode 和 send 操作。

redis 线程池设计原理

源码实现细节，见我的另一篇文章，redis 源码解析：IO 线程池。

3.4、skynet 线程池

为什么使用线程池？

skynet 是 cpu 密集型。处理读写 io 、数据包解压缩、业务逻辑处理等。特别地，当同一个 io 在多个线程处理时，将写 io 转由网络线程处理。

什么情况下使用线程池？

skynet 线程池作用于除 send 外的所有阶段。

怎样使用线程池？

skynet 的网络线程收集事件，将事件插入到对应的 actor 的专属消息队列中。全局消息队列存储有消息的 actor，即描述活跃的 actor，线程池负责检查该全局队列。

skynet线程池设计原理

线程池的实现