Nginx基础教程（91）Nginx多线程机制之在模块里使用多线程：Nginx的多线程魔法：当单线程遇上阻塞操作

本以为Nginx是单线程的纯情少年，没想到它偷偷玩起了多线程魔法。

0. 写在前面：当单线程不再孤单

如果你一直以为Nginx是个坚守单线程的"纯情少年"，那么今天可能要颠覆你的认知了。是的，那个以单线程事件驱动模型闻名天下的Nginx，其实早就偷偷玩起了多线程的魔法！

想象一下：Nginx的主线程就像一名高效的快递员，平时骑着电动车（事件驱动）在小区里飞快派件。但当他遇到需要爬楼梯送大冰箱（阻塞操作）时，他会聪明地呼叫后勤团队（线程池）来帮忙，而自己继续派送小包裹。

1. 初识Nginx的多面人格：单线程为主，多线程为辅

1.1 为什么单线程模型如此出色

首先，让我们重新认识一下Nginx的基本架构。Nginx采用Master-Worker多进程架构，每个Worker进程内部是单线程、事件驱动、非阻塞I/O的设计。

这种设计有诸多优点：

无锁设计：单线程内处理所有事件，无需加锁，消除了锁竞争和同步开销避免线程上下文切换：线程切换约消耗1μs，高并发下累积成显著开销内存占用极低：不为每个连接创建独立线程和栈空间

用一个形象的比喻：Nginx的Worker进程就像一位高效的独立收银员，不需要和别人交接班（无上下文切换），没有交接文档（无锁），也没有人帮忙（单线程），但她用了一个智能排队系统（epoll），只有顾客准备好结账时才会处理（事件通知），而不是呆呆地等着每个顾客挑选商品（阻塞等待）。

1.2 单线程的软肋：当阻塞操作来袭

再完美的设计也有软肋。Nginx的单线程模型虽然高效，但当遇到阻塞操作时，整个事件循环就会像堵车一样被卡住。

常见的阻塞操作包括：

读取未缓存的大文件复杂的数据库查询调用外部API接口耗时的CPU密集型计算

这就好比那位高效的快递员，突然遇到一个需要爬楼梯送大冰箱的订单，如果他亲自去送，那么车上所有小包裹都得等着，整个派件系统陷入瘫痪。

在Nginx中，这种情况更糟糕——一个工作进程被阻塞，会导致所有分配给它的连接都无法处理，即使系统有其他可用资源也无济于事。

2. 线程池：Nginx的御用后勤部队

2.1 线程池的救赎

从Nginx 1.7.11版本开始，Nginx引入了线程池机制，专门用来解决阻塞操作的问题。

线程池的工作原理很简单：

主线程（生产者）将阻塞任务放入队列后台线程（消费者）从队列中取出任务并执行任务完成后，通过管道（pipe）发送通知给主线程主线程继续处理后续工作

这就像餐厅的厨师和服务员：服务员（主线程）接单后，不等厨师（线程池）做完菜再去接待下一桌客人，而是把做菜任务交给厨师后立即去接待新客人，菜做好后厨师通过铃声（通知机制）告诉服务员来取餐。

2.2 线程池的适用场景

线程池并非万能，它主要适用于以下场景：

文件操作：读取或写入大型文件时数据库查询：执行复杂或耗时的数据库查询时API请求：与外部API通信时

需要注意的是，网络事件处理仍保持单线程模型不变，线程池仅用于处理阻塞IO。

3. 实战开始：配置Nginx线程池

3.1 基础配置

让我们从线程池的基础配置开始。在nginx.conf中添加以下配置：

# 创建名为file_io的线程池，包含8个线程
thread_pool file_io threads=8 max_queue=1024;
 
events {
    worker_connections 10240;
    use epoll;
    multi_accept on;
}
 
http {
    # 开启异步文件IO，使用file_io线程池
    aio threads=file_io;
    sendfile on;
    
    server {
        listen 80;
        
        location /download {
            root /storage;
            # 对大文件使用直接IO
            directio 8m;
        }
    }
}

关键参数解释：

threads：线程池中的线程数量，通常设置为CPU核心数max_queue：任务队列的最大长度，当队列满时新任务会被拒绝aio threads：启用异步文件IO并使用指定线程池

3.2 性能优化配置

对于高并发场景，我们还需要进行一些优化配置：

worker_processes auto;
worker_rlimit_nofile 65535;
 
events {
    worker_connections 10240;
    use epoll;
    multi_accept on;
}
 
http {
    thread_pool default threads=32 max_queue=65536;
    thread_pool file_io threads=8 max_queue=1024;
    
    aio threads=file_io;
    sendfile on;
    tcp_nopush on;
    tcp_nodelay on;
    keepalive_timeout 65;
    
    # 文件句柄缓存
    open_file_cache max=1000 inactive=20s;
    open_file_cache_valid 30s;
    open_file_cache_min_uses 2;
    open_file_cache_errors on;
    
    server {
        listen 80 reuseport;
        
        location /static {
            alias /data/static;
            aio threads=file_io;
        }
        
        location /api {
            proxy_pass http://backend;
            # API请求不使用线程池，保持事件驱动
        }
    }
}

3.3 验证线程池状态

配置完成后，我们可以验证线程池是否正常启用：

# 检查Nginx是否编译了threads模块
nginx -V 2>&1 | grep -- --with-threads
 
# 观察运行中的线程（如果线程池活跃，会看到多个线程）
ps -T -p $(pgrep -f "nginx: worker")

4. 深入源码：理解线程池内部机制

4.1 线程池的核心数据结构

根据Nginx线程池的源码分析，核心数据结构包括以下几个部分：

任务结构体：

typedef void (*cb_fun)(void *);
 
// 任务结构体
typedef struct task {
    void *argv; // 任务函数的参数
    cb_fun handler; // 任务函数
    struct task *next; // 任务链指针
} zoey_task_t;

任务队列：

typedef struct task_queue {
    zoey_task_t *head; // 队列头
    zoey_task_t **tail; // 队列尾
    unsigned int maxtasknum; // 最大任务限制
    unsigned int curtasknum; // 当前任务数
} zoey_task_queue_t;

线程池结构体：

typedef struct threadpool {
    pthread_mutex_t mutex; // 互斥锁
    pthread_cond_t cond; // 条件锁
    zoey_task_queue_t tasks; // 任务队列
    unsigned int threadnum; // 线程数
    unsigned int thread_stack_size; // 线程堆栈大小
} zoey_threadpool_t;

4.2 线程池的工作流程

线程池的工作流程可以分为以下几个步骤：

初始化线程池：创建指定数量的工作线程，初始化任务队列提交任务：主线程将任务添加到任务队列中通知工作线程：通过条件变量唤醒等待的工作线程处理任务：工作线程从队列中取出任务并执行任务完成：工作线程通过事件机制通知主线程

5. 实战进阶：在自定义模块中使用多线程

现在来到本文最精彩的部分——如何在自定义Nginx模块中使用多线程机制。

5.1 创建模块基础结构

首先，我们创建一个简单的模块框架：

config文件：

ngx_addon_name=ngx_http_my_threaded_module
HTTP_MODULES="$HTTP_MODULES ngx_http_my_threaded_module"
NGX_ADDON_SRCS="$NGX_ADDON_SRCS $ngx_addon_dir/ngx_http_my_threaded_module.c"

模块头文件： ngx_http_my_threaded_module.h

#ifndef _NGX_HTTP_MY_THREADED_MODULE_H_INCLUDED_
#define _NGX_HTTP_MY_THREADED_MODULE_H_INCLUDED_
 
#include <ngx_config.h>
#include <ngx_core.h>
#include <ngx_http.h>
 
typedef struct {
    ngx_str_t thread_pool_name;
    ngx_int_t max_task_size;
} ngx_http_my_threaded_loc_conf_t;
 
typedef struct {
    ngx_http_request_t *request;
    ngx_chain_t *in;
    ngx_chain_t *out;
    ngx_int_t (*handler)(ngx_http_request_t *r, ngx_chain_t *in, ngx_chain_t *out);
} ngx_http_my_threaded_ctx_t;
 
extern ngx_module_t ngx_http_my_threaded_module;
 
#endif

5.2 实现模块核心逻辑

模块源文件： ngx_http_my_threaded_module.c

#include "ngx_http_my_threaded_module.h"
 
static ngx_int_t ngx_http_my_threaded_handler(ngx_http_request_t *r);
static void ngx_http_my_threaded_task_handler(void *data, ngx_log_t *log);
static void ngx_http_my_threaded_event_handler(ngx_event_t *ev);
static void *ngx_http_my_threaded_create_loc_conf(ngx_conf_t *cf);
static char *ngx_http_my_threaded_merge_loc_conf(ngx_conf_t *cf, 
    void *parent, void *child);
 
static ngx_command_t ngx_http_my_threaded_commands[] = {
    {
        ngx_string("my_threaded"),
        NGX_HTTP_MAIN_CONF|NGX_HTTP_SRV_CONF|NGX_HTTP_LOC_CONF|NGX_CONF_TAKE1,
        ngx_conf_set_str_slot,
        NGX_HTTP_LOC_CONF_OFFSET,
        offsetof(ngx_http_my_threaded_loc_conf_t, thread_pool_name),
        NULL
    },
    {
        ngx_string("my_threaded_max_task_size"),
        NGX_HTTP_MAIN_CONF|NGX_HTTP_SRV_CONF|NGX_HTTP_LOC_CONF|NGX_CONF_TAKE1,
        ngx_conf_set_num_slot,
        NGX_HTTP_LOC_CONF_OFFSET,
        offsetof(ngx_http_my_threaded_loc_conf_t, max_task_size),
        NULL
    },
    ngx_null_command
};
 
static ngx_http_module_t ngx_http_my_threaded_module_ctx = {
    NULL, /* preconfiguration */
    NULL, /* postconfiguration */
    NULL, /* create main configuration */
    NULL, /* init main configuration */
    NULL, /* create server configuration */
    NULL, /* merge server configuration */
    ngx_http_my_threaded_create_loc_conf, /* create location configuration */
    ngx_http_my_threaded_merge_loc_conf /* merge location configuration */
};
 
ngx_module_t ngx_http_my_threaded_module = {
    NGX_MODULE_V1,
    &ngx_http_my_threaded_module_ctx, /* module context */
    ngx_http_my_threaded_commands, /* module directives */
    NGX_HTTP_MODULE, /* module type */
    NULL, /* init master */
    NULL, /* init module */
    NULL, /* init process */
    NULL, /* init thread */
    NULL, /* exit thread */
    NULL, /* exit process */
    NULL, /* exit master */
    NGX_MODULE_V1_PADDING
};
 
/* 创建location配置 */
static void *
ngx_http_my_threaded_create_loc_conf(ngx_conf_t *cf)
{
    ngx_http_my_threaded_loc_conf_t *conf;
    
    conf = ngx_pcalloc(cf->pool, sizeof(ngx_http_my_threaded_loc_conf_t));
    if (conf == NULL) {
        return NULL;
    }
    
    conf->thread_pool_name = ngx_string("default");
    conf->max_task_size = 1024; /* 默认最大任务大小 */
    
    return conf;
}
 
/* 合并location配置 */
static char *
ngx_http_my_threaded_merge_loc_conf(ngx_conf_t *cf, void *parent, void *child)
{
    ngx_http_my_threaded_loc_conf_t *prev = parent;
    ngx_http_my_threaded_loc_conf_t *conf = child;
    
    ngx_conf_merge_str_value(conf->thread_pool_name, 
                            prev->thread_pool_name, "default");
    ngx_conf_merge_value(conf->max_task_size, prev->max_task_size, 1024);
    
    return NGX_CONF_OK;
}
 
/* 实际的任务处理函数 - 模拟一个耗时操作 */
static ngx_int_t
ngx_http_my_threaded_heavy_task(ngx_http_request_t *r, 
                               ngx_chain_t *in, ngx_chain_t *out)
{
    ngx_chain_t *cl;
    u_char *p;
    size_t len;
    
    /* 模拟耗时处理 - 比如图像处理、复杂计算等 */
    ngx_msleep(100); /* 模拟100ms的处理时间 */
    
    /* 创建响应数据 */
    p = ngx_palloc(r->pool, 256);
    if (p == NULL) {
        return NGX_ERROR;
    }
    
    len = ngx_sprintf(p, "Heavy task completed for request %V
", &r->uri) - p;
    
    cl = ngx_alloc_chain_link(r->pool);
    if (cl == NULL) {
        return NGX_ERROR;
    }
    
    cl->buf = ngx_create_temp_buf(r->pool, len);
    if (cl->buf == NULL) {
        return NGX_ERROR;
    }
    
    ngx_memcpy(cl->buf->pos, p, len);
    cl->buf->last = cl->buf->pos + len;
    cl->next = NULL;
    
    *out = cl;
    
    return NGX_OK;
}
 
/* 线程池任务处理函数 */
static void
ngx_http_my_threaded_task_handler(void *data, ngx_log_t *log)
{
    ngx_http_my_threaded_ctx_t *ctx = data;
    ngx_int_t rc;
    
    /* 在线程池中执行耗时任务 */
    rc = ctx->handler(ctx->request, ctx->in, ctx->out);
    
    /* 保存结果到上下文中 */
    ctx->request->status = rc;
    
    /* 注意：这里不能直接操作请求，需要通过事件通知主线程 */
}
 
/* 事件处理函数 - 在主线程中执行 */
static void
ngx_http_my_threaded_event_handler(ngx_event_t *ev)
{
    ngx_http_my_threaded_ctx_t *ctx = ev->data;
    ngx_http_request_t *r = ctx->request;
    ngx_int_t rc;
    
    /* 从上下文中获取任务执行结果 */
    rc = r->status;
    
    if (rc == NGX_ERROR) {
        ngx_http_finalize_request(r, NGX_HTTP_INTERNAL_SERVER_ERROR);
        return;
    }
    
    /* 发送响应 */
    ngx_http_send_header(r);
    
    if (ctx->out) {
        ngx_http_output_filter(r, ctx->out);
    }
    
    ngx_http_finalize_request(r, NGX_HTTP_OK);
}
 
/* 主要的请求处理函数 */
static ngx_int_t
ngx_http_my_threaded_handler(ngx_http_request_t *r)
{
    ngx_http_my_threaded_loc_conf_t *conf;
    ngx_http_my_threaded_ctx_t *ctx;
    ngx_event_t *ev;
    ngx_thread_task_t *task;
    
    conf = ngx_http_get_module_loc_conf(r, ngx_http_my_threaded_module);
    
    /* 只处理GET请求 */
    if (r->method != NGX_HTTP_GET) {
        return NGX_HTTP_NOT_ALLOWED;
    }
    
    /* 创建上下文 */
    ctx = ngx_palloc(r->pool, sizeof(ngx_http_my_threaded_ctx_t));
    if (ctx == NULL) {
        return NGX_HTTP_INTERNAL_SERVER_ERROR;
    }
    
    ctx->request = r;
    ctx->in = NULL;
    ctx->out = NULL;
    ctx->handler = ngx_http_my_threaded_heavy_task;
    
    /* 创建线程任务 */
    task = ngx_thread_task_alloc(r->pool, sizeof(ngx_http_my_threaded_ctx_t));
    if (task == NULL) {
        return NGX_HTTP_INTERNAL_SERVER_ERROR;
    }
    
    task->handler = ngx_http_my_threaded_task_handler;
    task->ctx = ctx;
    task->event.data = ctx;
    task->event.handler = ngx_http_my_threaded_event_handler;
    
    /* 提交任务到线程池 */
    if (ngx_thread_task_post(conf->thread_pool_name, task) != NGX_OK) {
        return NGX_HTTP_INTERNAL_SERVER_ERROR;
    }
    
    /* 设置超时处理 */
    ngx_add_timer(&task->event, 30000); /* 30秒超时 */
    
    return NGX_AGAIN;
}
 
/* 模块注册 */
static ngx_int_t
ngx_http_my_threaded_init(ngx_conf_t *cf)
{
    ngx_http_handler_pt *h;
    ngx_http_core_main_conf_t *cmcf;
    
    cmcf = ngx_http_conf_get_module_main_conf(cf, ngx_http_core_module);
    
    h = ngx_array_push(&cmcf->phases[NGX_HTTP_CONTENT_PHASE].handlers);
    if (h == NULL) {
        return NGX_ERROR;
    }
    
    *h = ngx_http_my_threaded_handler;
    
    return NGX_OK;
}

5.3 配置和使用模块

在nginx.conf中配置我们新创建的模块：

thread_pool my_pool threads=8 max_queue=65536;
 
http {
    server {
        listen 8080;
        
        location /heavy-task {
            my_threaded my_pool;
            my_threaded_max_task_size 2048;
        }
    }
}

6. 性能测试与优化建议

6.1 性能对比测试

为了验证线程池的效果，我们可以进行一个简单的性能测试：

测试场景：

1000个并发连接每个请求需要处理一个耗时100ms的任务比较使用线程池前后的性能

预期结果：

不使用线程池：QPS约为10（1000/100）使用线程池（8线程）：QPS约为80（8 * 10）

6.2 优化建议

根据实际使用经验，以下是线程池使用的一些优化建议：

线程数量设置： CPU密集型任务：线程数 = CPU核心数IO密集型任务：线程数 = CPU核心数 * 2 队列大小调整： 根据任务特性和内存大小调整避免队列过大导致内存占用过高 监控与调试： 定期检查线程池状态监控任务队列长度，避免队列满导致任务拒绝 错误处理： 实现完善的错误处理机制对任务超时进行适当处理

7. 结语：多线程不是银弹，而是精妙的补充

Nginx的多线程机制并不是要抛弃其经典的事件驱动模型，而是作为一种精妙的补充，专门用于处理那些不可避免的阻塞操作。

就像一位聪明的厨师，大部分时间都在高效地炒菜（事件驱动），但当遇到需要长时间炖煮的菜品（阻塞操作）时，他会明智地使用炖锅（线程池）来处理，而不是守在锅前浪费时间。

通过本文的介绍，希望你能深刻理解Nginx多线程机制的精髓，并在实际项目中灵活运用。记住，技术没有好坏之分，只有适用与否。在合适的场景使用合适的技术，这才是工程师的真正智慧。

现在，就去为你的Nginx配置一个合适的"后勤部队"吧，让它在那单线程的高速公路上，也能畅通无阻地处理那些"大件货物"！