Nginx线程池浅析
Nginx通过使用多路复用IO（如Linux的epoll、FreeBSD的kqueue等）技术很好的解决了c10k问题，但前提是Nginx的请求不能有阻塞操作，否则将会导致整个Nginx进程停止服务。
但很多时候阻塞操作是不可避免的，例如客户端请求静态文件时，由于磁盘IO可能会导致进程阻塞，所以将会导致Nginx的性能下降。为了解决这个问题，Nginx在1.7.11版本中实现了线程池机制。
下面我们将会分析Nginx是怎么通过线程池来解决阻塞操作问题。
启用线程池功能
要使用线程池功能，首先需要在配置文件中添加如下配置项：
location / {
root /html;
thread_pool default threads=32 max_queue=65536;
aio threads=default;
}上面定义了一个名为“default”，包含32个线程，任务队列最多支持65536个请求的线程池。如果任务队列过载，Nginx将输出如下错误日志并拒绝请求：
thread pool "default" queue overflow: N tasks waiting
如果出现上面的错误，说明线程池的负载很高，这是可以通过添加线程数来解决这个问题。当达到机器的最高处理能力之后，增加线程数并不能改善这个问题。
一切从“源”开始
下面主要通过剖析Nginx的源码来了解线程池机制实现原理。现在先来了解Nginx线程池的两个重要数据结构ngx_thread_pool_t和ngx_thread_task_t。
ngx_thread_pool_t结构体
struct ngx_thread_pool_s {
ngx_thread_mutex_t mtx;
ngx_thread_pool_queue_t queue;
ngx_int_t waiting;
ngx_thread_cond_t cond;
ngx_log_t *log;
ngx_str_t name;
ngx_uint_t threads;
ngx_int_t max_queue;
u_char *file;
ngx_uint_t line;
};下面解释下每个字段的用途：
1) mtx: 互斥锁，用于锁定任务队列，避免竞争状态。
2) queue: 任务队列。
3) waiting: 有多少个任务正在等待处理。
4) cond: 用于通知线程池有任务需要处理。
5) name: 线程池名称。
6) threads: 线程池由多少个线程组成（线程数）。
7) max_queue: 线程池最大能处理的任务数。
ngx_thread_task_t结构体
struct ngx_thread_task_s {
ngx_thread_task_t *next;
ngx_uint_t id;
void *ctx;
void (*handler)(void *data, ngx_log_t *log);
ngx_event_t event;
};下面解释下每个字段的用途：
1) next: 指向下一个任务。
2) id: 任务ID。
3) ctx: 任务的上下文。
4) handler: 处理任务的函数句柄。
5) event: 跟任务关联的事件对象（当线程池处理成任务之后将会由主线程调用event对象的handler回调函数）。
线程池初始化
下面介绍下线程池的初始化过程。
在Nginx启动的时候，首先会调用ngx_thread_pool_init_worker()函数来初始化线程池。ngx_thread_pool_init_worker()函数最终会调用ngx_thread_pool_init()，源码如下：
static ngx_int_t
ngx_thread_pool_init(ngx_thread_pool_t *tp, ngx_log_t *log, ngx_pool_t *pool)
{
...
for (n = 0; n < tp->threads; n++) {
err = pthread_create(&tid, &attr, ngx_thread_pool_cycle, tp);
if (err) {
ngx_log_error(NGX_LOG_ALERT, log, err,
"pthread_create() failed");
return NGX_ERROR;
}
}
...
return NGX_OK;
}
ngx_thread_pool_init()最终调用pthread_create()函数创建线程池中的工作线程，工作线程会从ngx_thread_pool_cycle()函数开始执行。
ngx_thread_pool_cycle()函数源码如下：
static void *
ngx_thread_pool_cycle(void *data)
{
...
for ( ;; ) {
if (ngx_thread_mutex_lock(&tp->mtx, tp->log) != NGX_OK) {
return NULL;
}
tp->waiting--;
while (tp->queue.first == NULL) {
if (ngx_thread_cond_wait(&tp->cond, &tp->mtx, tp->log)
!= NGX_OK)
{
(void) ngx_thread_mutex_unlock(&tp->mtx, tp->log);
return NULL;
}
}
// 获取一个任务对象
task = tp->queue.first;
tp->queue.first = task->next;
if (tp->queue.first == NULL) {
tp->queue.last = &tp->queue.first;
}
if (ngx_thread_mutex_unlock(&tp->mtx, tp->log) != NGX_OK) {
return NULL;
}
// 处理任务
task->handler(task->ctx, tp->log);
task->next = NULL;
ngx_spinlock(&ngx_thread_pool_done_lock, 1, 2048);
// 把处理完的任务放置到完成队列中
*ngx_thread_pool_done.last = task;
ngx_thread_pool_done.last = &task->next;
ngx_unlock(&ngx_thread_pool_done_lock);
(void) ngx_notify(ngx_thread_pool_handler); // 通知主线程
}
}ngx_thread_pool_cycle()函数的主要工作是从待处理的任务队列中获取一个任务，然后调用任务对象的handler()函数处理任务，完成后把任务放置到完成队列中，并通过ngx_notify()通知主线程。
添加任务到任务队列
通过上面的分析，我们知道了线程池是怎么从任务队列获取任务并处理。但任务队列的任务从哪里来的呢？因为Nginx的使命是处理客户端请求，所以可以知道任务是通过客户端请求产生的。也就是说，任务是主线程创建的（主线程负责处理客户端请求）。
主线程通过ngx_thread_task_post()函数向任务队列中添加一个任务，代码如下：
ngx_int_t
ngx_thread_task_post(ngx_thread_pool_t *tp, ngx_thread_task_t *task)
{
...
if (ngx_thread_mutex_lock(&tp->mtx, tp->log) != NGX_OK) {
return NGX_ERROR;
}
// 通知线程池有任务需要处理
if (ngx_thread_cond_signal(&tp->cond, tp->log) != NGX_OK) {
(void) ngx_thread_mutex_unlock(&tp->mtx, tp->log);
return NGX_ERROR;
}
// 把任务添加到任务队列中
*tp->queue.last = task;
tp->queue.last = &task->next;
tp->waiting++;
(void) ngx_thread_mutex_unlock(&tp->mtx, tp->log);
return NGX_OK;
}ngx_thread_task_post()函数首先调用ngx_thread_cond_signal()通知线程池的线程有任务需要处理，然后把任务添加到任务队列中。可能有人会问，先通知线程池在添加任务到任务队列中会不会有顺序问题。其实这样做是没问题的，这是因为只要主线程不调用ngx_thread_mutex_unlock()把互斥锁解开，线程池中的工作线程是不会从ngx_thread_cond_wait()返回的。
收尾工作
当线程池把任务处理完后会把其放置到完成队列中（ngx_thread_pool_done），然后调用ngx_notify()通知主线程有任务完成了。主线程收到通知后，会在事件模块中进行收尾工作：调用task.event.handler()。task.event.handler由任务创建者设置，例如在ngx_http_copy_filter模块的ngx_http_copy_thread_handler()函数：
static ngx_int_t
ngx_http_copy_thread_handler(ngx_thread_task_t *task, ngx_file_t *file)
{
...
if (tp == NULL) {
if (ngx_http_complex_value(r, clcf->thread_pool_value, &name)
!= NGX_OK)
{
return NGX_ERROR;
}
tp = ngx_thread_pool_get((ngx_cycle_t *) ngx_cycle, &name);
}
task->event.data = r;
// 设置event的回调函数
task->event.handler = ngx_http_copy_thread_event_handler;
if (ngx_thread_task_post(tp, task) != NGX_OK) {
return NGX_ERROR;
}
r->main->blocked++;
r->aio = 1;
return NGX_OK;
}task.event.handler被设置为ngx_http_copy_thread_event_handler，就是说当任务处理完成后，主线程将会调用ngx_http_copy_thread_event_handler来进行收尾工作。
哪些操作会使用线程池
那么哪些操作会使用线程池去处理。一般来说，磁盘IO会使用线程池来处理。在ngx_http_copy_filter模块中，会调用ngx_thread_read()读取文件的内容（当启用了线程池时），而ngx_thread_read()会把读取文件内容的操作让线程池去处理。ngx_thread_read()代码如下：
ssize_t
ngx_thread_read(ngx_thread_task_t **taskp, ngx_file_t *file, u_char *buf,
size_t size, off_t offset, ngx_pool_t *pool)
{
...
task = *taskp;
if (task == NULL) {
task = ngx_thread_task_alloc(pool, sizeof(ngx_thread_read_ctx_t));
if (task == NULL) {
return NGX_ERROR;
}
task->handler = ngx_thread_read_handler;
*taskp = task;
}
ctx = task->ctx;
...
ctx->fd = file->fd;
ctx->buf = buf;
ctx->size = size;
ctx->offset = offset;
if (file->thread_handler(task, file) != NGX_OK) {
return NGX_ERROR;
}
return NGX_AGAIN;
}从上面的代码看到，task的handler被设置为ngx_thread_read_handler，也就是说在线程池中将会调用ngx_thread_read_handler()去读取文件内容。而file->thread_handler()将会调用ngx_thread_task_post()，前面已经分析过，ngx_thread_task_post()会把任务添加到任务队列中。
图解
最后用一张图来解释Nginx线程池机制的原理吧。
