Syntax: thread_pool name threads=number [max_queue=number];Default: thread_pool default threads=32 max_queue=65536;Context: main

/******************* nginx/src/core/ngx_thread_pool.c ************************///創(chuàng)建線程池所需的基礎(chǔ)結(jié)構(gòu)static void * ngx_thread_pool_create_conf(ngx_cycle_t *cycle){  ngx_thread_pool_conf_t *tcf;   //從cycle->pool指向的內(nèi)存池中申請一塊內(nèi)存  tcf = ngx_pcalloc(cycle->pool, sizeof(ngx_thread_pool_conf_t));  if (tcf == NULL) {    return NULL;  }      //先申請包含4個ngx_thread_pool_t指針類型元素的數(shù)組   //ngx_thread_pool_t結(jié)構(gòu)體中保存了一個線程池相關(guān)的信息  if (ngx_array_init(&tcf->pools, cycle->pool, 4,            sizeof(ngx_thread_pool_t *))    != NGX_OK)  {    return NULL;  }   return tcf;} //解析處理配置文件中thread_pool的配置，并將相關(guān)信息保存的ngx_thread_pool_t中static char * ngx_thread_pool(ngx_conf_t *cf, ngx_command_t *cmd, void *conf){  ngx_str_t     *value;  ngx_uint_t     i;  ngx_thread_pool_t *tp;   value = cf->args->elts;   //根據(jù)thread_pool配置中的name作為線程池的唯一標(biāo)識（如果重名，只有第一個有效）  //申請ngx_thread_pool_t結(jié)構(gòu)保存線程池的相關(guān)信息  //由此可見，nginx支持配置多個name不同的線程池  tp = ngx_thread_pool_add(cf, &value[1]);  .......  //處理thread_pool配置行的所有元素  for (i = 2; i < cf->args->nelts; i++) {    //檢查配置的線程數(shù)    if (ngx_strncmp(value[i].data, "threads=", 8) == 0) {     .......    }         //檢查配置的最大隊列長度    if (ngx_strncmp(value[i].data, "max_queue=", 10) == 0) {     .......    }  }  ......} //判斷包含多個線程池的數(shù)組中的各個線程池的配置是否正確static char * ngx_thread_pool_init_conf(ngx_cycle_t *cycle, void *conf){  ....  ngx_thread_pool_t **tpp;   tpp = tcf->pools.elts;  //遍歷數(shù)組中所有的線程池配置，并檢查其正確性  for (i = 0; i < tcf->pools.nelts; i++) {    .....  }   return NGX_CONF_OK;}

/******************* nginx/src/core/ngx_thread_pool.c ************************///創(chuàng)建線程池所需的基礎(chǔ)結(jié)構(gòu)static ngx_int_tngx_thread_pool_init_worker(ngx_cycle_t *cycle){  ngx_uint_t        i;  ngx_thread_pool_t    **tpp;  ngx_thread_pool_conf_t  *tcf;  //如果不是worker或者只有一個worker就不起用線程池  if (ngx_process != NGX_PROCESS_WORKER    && ngx_process != NGX_PROCESS_SINGLE)  {    return NGX_OK;  }     //初始化任務(wù)隊列  ngx_thread_pool_queue_init(&ngx_thread_pool_done);   tpp = tcf->pools.elts;  for (i = 0; i < tcf->pools.nelts; i++) {    //初始化各個線程池    if (ngx_thread_pool_init(tpp[i], cycle->log, cycle->pool) != NGX_OK) {      return NGX_ERROR;    }  }   return NGX_OK;} //線程池初始化static ngx_int_t ngx_thread_pool_init(ngx_thread_pool_t *tp, ngx_log_t *log, ngx_pool_t *pool){  .....  //初始化任務(wù)隊列  ngx_thread_pool_queue_init(&tp->queue);   //創(chuàng)建線程鎖  if (ngx_thread_mutex_create(&tp->mtx, log) != NGX_OK) {    return NGX_ERROR;  }   //創(chuàng)建線程條件變量  if (ngx_thread_cond_create(&tp->cond, log) != NGX_OK) {    (void) ngx_thread_mutex_destroy(&tp->mtx, log);    return NGX_ERROR;  }  ......  for (n = 0; n < tp->threads; n++) {    //創(chuàng)建線程池中的每個線程    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;    }  }  ......} //線程池中線程處理主函數(shù)static void *ngx_thread_pool_cycle(void *data){   ......   for ( ;; ) {    //阻塞的方式獲取線程鎖    if (ngx_thread_mutex_lock(&tp->mtx, tp->log) != NGX_OK) {      return NULL;    }     /* the number may become negative */    tp->waiting--;     //如果任務(wù)隊列為空，就cond_wait阻塞等待有新任務(wù)時調(diào)用cond_signal/broadcast觸發(fā)    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;      }    }    //從任務(wù)隊列中獲取task，并將其從隊列中移除    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的處理函數(shù)    task->handler(task->ctx, tp->log);    .....     ngx_spinlock(&ngx_thread_pool_done_lock, 1, 2048);     //將經(jīng)過預(yù)處理的任務(wù)添加到done隊列中等待調(diào)用event的回調(diào)函數(shù)繼續(xù)處理    *ngx_thread_pool_done.last = task;    ngx_thread_pool_done.last = &task->next;         //防止編譯器優(yōu)化，保證解鎖操作是在上述語句執(zhí)行完畢后再去執(zhí)行的    ngx_memory_barrier();     ngx_unlock(&ngx_thread_pool_done_lock);         (void) ngx_notify(ngx_thread_pool_handler);  }} //處理pool_done隊列上task中包含的每個event事件static void ngx_thread_pool_handler(ngx_event_t *ev){  .....  ngx_spinlock(&ngx_thread_pool_done_lock, 1, 2048);   //獲取任務(wù)鏈表的頭部  task = ngx_thread_pool_done.first;  ngx_thread_pool_done.first = NULL;  ngx_thread_pool_done.last = &ngx_thread_pool_done.first;   ngx_memory_barrier();   ngx_unlock(&ngx_thread_pool_done_lock);   while (task) {    ngx_log_debug1(NGX_LOG_DEBUG_CORE, ev->log, 0,            "run completion handler for task #%ui", task->id);    //遍歷隊列中的所有任務(wù)事件    event = &task->event;    task = task->next;     event->complete = 1;    event->active = 0;     //調(diào)用event對應(yīng)的處理函數(shù)有針對性的進(jìn)行處理    event->handler(event);  }} 

/*********************** nginx/src/os/unix/ngx_files.c **********************///file_cache模塊的處理函數(shù)（涉及到了線程池）static ssize_t ngx_http_file_cache_aio_read(ngx_http_request_t *r, ngx_http_cache_t *c){  .......#if (NGX_THREADS)   if (clcf->aio == NGX_HTTP_AIO_THREADS) {    c->file.thread_task = c->thread_task;    //這里注冊的函數(shù)在下面語句中的ngx_thread_read函數(shù)中被調(diào)用    c->file.thread_handler = ngx_http_cache_thread_handler;    c->file.thread_ctx = r;    //根據(jù)任務(wù)的屬性，選擇正確的線程池，并初始化task結(jié)構(gòu)體中的各個成員        n = ngx_thread_read(&c->file, c->buf->pos, c->body_start, 0, r->pool);     c->thread_task = c->file.thread_task;    c->reading = (n == NGX_AGAIN);     return n;  }#endif   return ngx_read_file(&c->file, c->buf->pos, c->body_start, 0);}  //task任務(wù)的處理函數(shù)static ngx_int_t ngx_http_cache_thread_handler(ngx_thread_task_t *task, ngx_file_t *file){  .......  tp = clcf->thread_pool;  .......     task->event.data = r;  //注冊thread_event_handler函數(shù)，該函數(shù)在處理pool_done隊列中event事件時被調(diào)用  task->event.handler = ngx_http_cache_thread_event_handler;   //將任務(wù)放到線程池的任務(wù)隊列中  if (ngx_thread_task_post(tp, task) != NGX_OK) {    return NGX_ERROR;  }  ......} /*********************** nginx/src/core/ngx_thread_pool.c **********************///添加任務(wù)到隊列中ngx_int_t ngx_thread_task_post(ngx_thread_pool_t *tp, ngx_thread_task_t *task){  //如果當(dāng)前的任務(wù)正在處理就退出  if (task->event.active) {    ngx_log_error(NGX_LOG_ALERT, tp->log, 0,           "task #%ui already active", task->id);    return NGX_ERROR;  }   if (ngx_thread_mutex_lock(&tp->mtx, tp->log) != NGX_OK) {    return NGX_ERROR;  }     //判斷當(dāng)前線程池等待的任務(wù)數(shù)量與最大隊列長度的關(guān)系  if (tp->waiting >= tp->max_queue) {    (void) ngx_thread_mutex_unlock(&tp->mtx, tp->log);     ngx_log_error(NGX_LOG_ERR, tp->log, 0,           "thread pool /"%V/" queue overflow: %i tasks waiting",           &tp->name, tp->waiting);    return NGX_ERROR;  }  //激活任務(wù)  task->event.active = 1;   task->id = ngx_thread_pool_task_id++;  task->next = NULL;     //通知阻塞的線程有新事件加入，可以解除阻塞  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);   ngx_log_debug2(NGX_LOG_DEBUG_CORE, tp->log, 0,          "task #%ui added to thread pool /"%V/"",          task->id, &tp->name);   return NGX_OK;}