Nginx（⼆）：worker进程处理流程框架解析 Nginx 启动起来之后，会有⼏个进程运⾏：1. master 进程接收⽤户命令并做出响应; 2. worker 进程负责处理各⽹络事件，并同时接收来⾃master的处理协调命令； master 主要是⼀控制命令，我们后⾯再说，⽽worker则是处理的nginx的核⼼任务，请求转发、反向代理、负载均衡等⼯作。
所以我们先来啃啃worker这块硬⾻头吧！0. worker 主循环 worker 的启动是被master 操作的，作为⼀个 fork 出来的进程，它拥有和master⼀样的内存数据信息。
但它的活动范围相对较⼩，所以它并不会替代master的位置。
// unix/ngx_process_cycle.cvoidngx_master_process_cycle(ngx_cycle_t *cycle){char *title;u_char *p;size_t size;ngx_int_t i;ngx_uint_t sigio;sigset_t set;struct itimerval itv;ngx_uint_t live;ngx_msec_t delay;ngx_core_conf_t *ccf;sigemptyset(&set);sigaddset(&set, SIGCHLD);sigaddset(&set, SIGALRM);sigaddset(&set, SIGIO);sigaddset(&set, SIGINT);sigaddset(&set, ngx_signal_value(NGX_RECONFIGURE_SIGNAL));sigaddset(&set, ngx_signal_value(NGX_REOPEN_SIGNAL));sigaddset(&set, ngx_signal_value(NGX_NOACCEPT_SIGNAL));sigaddset(&set, ngx_signal_value(NGX_TERMINATE_SIGNAL));sigaddset(&set, ngx_signal_value(NGX_SHUTDOWN_SIGNAL));sigaddset(&set, ngx_signal_value(NGX_CHANGEBIN_SIGNAL));if (sigprocmask(SIG_BLOCK, &set, NULL) == -1) {ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,"sigprocmask() failed");}sigemptyset(&set);size = sizeof(master_process);for (i = 0; i < ngx_argc; i++) {size += ngx_strlen(ngx_argv[i]) + 1;}title = ngx_pnalloc(cycle->pool, size);if (title == NULL) {/* fatal */exit(2);}p = ngx_cpymem(title, master_process, sizeof(master_process) - 1);for (i = 0; i < ngx_argc; i++) {*p++ = '';p = ngx_cpystrn(p, (u_char *) ngx_argv[i], size);}ngx_setproctitle(title);ccf = (ngx_core_conf_t *) ngx_get_conf(cycle->conf_ctx, ngx_core_module);// 启动之后会主动启动 worker 进程ngx_start_worker_processes(cycle, ccf->worker_processes,NGX_PROCESS_RESPAWN);ngx_start_cache_manager_processes(cycle, 0);ngx_new_binary = 0;delay = 0;sigio = 0;live = 1;for ( ;; ) {if (delay) {if (ngx_sigalrm) {sigio = 0;delay *= 2;ngx_sigalrm = 0;}ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0,"termination cycle: %M", delay);itv.it__sec = 0;itv.it__usec = 0;itv.it__sec = delay / 1000;itv.it__usec = (delay % 1000 ) * 1000;if (setitimer(ITIMER_REAL, &itv, NULL) == -1) {ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,"setitimer() failed");}}ngx_log_debug0(NGX_LOG_DEBUG_EVENT, cycle->log, 0, "sigsuspend");sigsuspend(&set);ngx_time_update();ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0,"wake up, sigio %i", sigio);ngx_log_debug0(NGX_LOG_DEBUG_EVENT, cycle->log, 0, "reap children"); live = ngx_reap_children(cycle);}if (!live && (ngx_terminate || ngx_quit)) {ngx_master_process_exit(cycle);}if (ngx_terminate) {if (delay == 0) {delay = 50;}if (sigio) {sigio--;continue;}sigio = ccf->worker_processes + 2/* cache processes */;if (delay > 1000) {ngx_signal_worker_processes(cycle, SIGKILL);} else {ngx_signal_worker_processes(cycle,ngx_signal_value(NGX_TERMINATE_SIGNAL));}continue;}if (ngx_quit) {ngx_signal_worker_processes(cycle,ngx_signal_value(NGX_SHUTDOWN_SIGNAL));ngx_close_listening_sockets(cycle);continue;}if (ngx_reconfigure) {ngx_reconfigure = 0;if (ngx_new_binary) {ngx_start_worker_processes(cycle, ccf->worker_processes,NGX_PROCESS_RESPAWN);ngx_start_cache_manager_processes(cycle, 0);ngx_noaccepting = 0;continue;}ngx_log_error(NGX_LOG_NOTICE, cycle->log, 0, "reconfiguring");cycle = ngx_init_cycle(cycle);if (cycle == NULL) {cycle = (ngx_cycle_t *) ngx_cycle;continue;}ngx_cycle = cycle;ccf = (ngx_core_conf_t *) ngx_get_conf(cycle->conf_ctx,ngx_core_module);// 收到reconfig命令时，重启worker 进程ngx_start_worker_processes(cycle, ccf->worker_processes,NGX_PROCESS_JUST_RESPAWN);ngx_start_cache_manager_processes(cycle, 1);/* allow new processes to start */ngx_msleep(100);live = 1;ngx_signal_worker_processes(cycle,ngx_signal_value(NGX_SHUTDOWN_SIGNAL));}if (ngx_restart) {ngx_restart = 0;// 收到重启命令时，传递消息给 workerngx_start_worker_processes(cycle, ccf->worker_processes,NGX_PROCESS_RESPAWN);ngx_start_cache_manager_processes(cycle, 0);live = 1;}if (ngx_reopen) {ngx_reopen = 0;ngx_log_error(NGX_LOG_NOTICE, cycle->log, 0, "reopening logs");ngx_reopen_files(cycle, ccf->user);ngx_signal_worker_processes(cycle,ngx_signal_value(NGX_REOPEN_SIGNAL));}if (ngx_change_binary) {ngx_change_binary = 0;ngx_log_error(NGX_LOG_NOTICE, cycle->log, 0, "changing binary");ngx_new_binary = ngx_exec_new_binary(cycle, ngx_argv);}if (ngx_noaccept) {ngx_noaccept = 0;ngx_noaccepting = 1;ngx_signal_worker_processes(cycle,ngx_signal_value(NGX_SHUTDOWN_SIGNAL));}}}static voidngx_start_worker_processes(ngx_cycle_t *cycle, ngx_int_t n, ngx_int_t type){ngx_log_error(NGX_LOG_NOTICE, cycle->log, 0, "start worker processes");ngx_memzero(&ch, sizeof(ngx_channel_t));mand = NGX_CMD_OPEN_CHANNEL;// n 代表worker的进程数，在 nginx.conf 中配置for (i = 0; i < n; i++) {// 依次启动 worker 进程，实际上就是通过fork进⾏⼦进程启动的ngx_spawn_process(cycle, ngx_worker_process_cycle,(void *) (intptr_t) i, "worker process", type);ch.pid = ngx_processes[ngx_process_slot].pid;ch.slot = ngx_process_slot;ch.fd = ngx_processes[ngx_process_slot].channel[0];ngx_pass_open_channel(cycle, &ch);}}ngx_pid_tngx_spawn_process(ngx_cycle_t *cycle, ngx_spawn_proc_pt proc, void *data,char *name, ngx_int_t respawn){u_long on;ngx_pid_t pid;ngx_int_t s;if (respawn >= 0) {s = respawn;} else {for (s = 0; s < ngx_last_process; s++) {if (ngx_processes[s].pid == -1) {break;}}if (s == NGX_MAX_PROCESSES) {ngx_log_error(NGX_LOG_ALERT, cycle->log, 0,"no more than %d processes can be spawned",NGX_MAX_PROCESSES);return NGX_INVALID_PID;}}if (respawn != NGX_PROCESS_DETACHED) {/* Solaris 9 still has no AF_LOCAL */if (socketpair(AF_UNIX, SOCK_STREAM, 0, ngx_processes[s].channel) == -1) {ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,"socketpair() failed while spawning \"%s\"", name);return NGX_INVALID_PID;}ngx_log_debug2(NGX_LOG_DEBUG_CORE, cycle->log, 0,"channel %d:%d",ngx_processes[s].channel[0],ngx_processes[s].channel[1]);if (ngx_nonblocking(ngx_processes[s].channel[0]) == -1) {ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,ngx_nonblocking_n " failed while spawning \"%s\"",name);ngx_close_channel(ngx_processes[s].channel, cycle->log);return NGX_INVALID_PID;}if (ngx_nonblocking(ngx_processes[s].channel[1]) == -1) {ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,ngx_nonblocking_n " failed while spawning \"%s\"",name);ngx_close_channel(ngx_processes[s].channel, cycle->log);return NGX_INVALID_PID;}on = 1;if (ioctl(ngx_processes[s].channel[0], FIOASYNC, &on) == -1) {ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,"ioctl(FIOASYNC) failed while spawning \"%s\"", name);ngx_close_channel(ngx_processes[s].channel, cycle->log);return NGX_INVALID_PID;}if (fcntl(ngx_processes[s].channel[0], F_SETOWN, ngx_pid) == -1) {ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,"fcntl(F_SETOWN) failed while spawning \"%s\"", name);ngx_close_channel(ngx_processes[s].channel, cycle->log);return NGX_INVALID_PID;}if (fcntl(ngx_processes[s].channel[0], F_SETFD, FD_CLOEXEC) == -1) {ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,"fcntl(FD_CLOEXEC) failed while spawning \"%s\"",name);ngx_close_channel(ngx_processes[s].channel, cycle->log);return NGX_INVALID_PID;}if (fcntl(ngx_processes[s].channel[1], F_SETFD, FD_CLOEXEC) == -1) {ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,"fcntl(FD_CLOEXEC) failed while spawning \"%s\"",name);ngx_close_channel(ngx_processes[s].channel, cycle->log);return NGX_INVALID_PID;}ngx_channel = ngx_processes[s].channel[1];} else {ngx_processes[s].channel[0] = -1;ngx_processes[s].channel[1] = -1;}ngx_process_slot = s;// fork 出⼦进程出来pid = fork();switch (pid) {case -1:ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,"fork() failed while spawning \"%s\"", name);ngx_close_channel(ngx_processes[s].channel, cycle->log);return NGX_INVALID_PID;case0:ngx_parent = ngx_pid;ngx_pid = ngx_getpid();// ⼦进程将调⽤传⼊的处理⽅法，worker 则会进⼊循环处理事件逻辑中// 即 ngx_worker_process_cycle 循环proc(cycle, data);break;default:break;}ngx_log_error(NGX_LOG_NOTICE, cycle->log, 0, "start %s %P", name, pid); ngx_processes[s].pid = pid;ngx_processes[s].exited = 0;if (respawn >= 0) {return pid;}ngx_processes[s].proc = proc;ngx_processes[s].data = data;ngx_processes[s].name = name;ngx_processes[s].exiting = 0;switch (respawn) {case NGX_PROCESS_NORESPAWN:ngx_processes[s].respawn = 0;ngx_processes[s].just_spawn = 0;ngx_processes[s].detached = 0;break;case NGX_PROCESS_JUST_SPAWN:ngx_processes[s].respawn = 0;ngx_processes[s].just_spawn = 1;ngx_processes[s].detached = 0;break;case NGX_PROCESS_RESPAWN:ngx_processes[s].respawn = 1;ngx_processes[s].just_spawn = 0;ngx_processes[s].detached = 0;break;case NGX_PROCESS_JUST_RESPAWN:ngx_processes[s].respawn = 1;ngx_processes[s].just_spawn = 1;ngx_processes[s].detached = 0;break;case NGX_PROCESS_DETACHED:ngx_processes[s].respawn = 0;ngx_processes[s].just_spawn = 0;ngx_processes[s].detached = 1;break;}if (s == ngx_last_process) {ngx_last_process++;}return pid;}// os/unix/ngx_process_cycle.c// worker 主循环服务static voidngx_worker_process_cycle(ngx_cycle_t *cycle, void *data){ngx_int_t worker = (intptr_t) data;ngx_process = NGX_PROCESS_WORKER;ngx_worker = worker;ngx_worker_process_init(cycle, worker);// 进程标题 worker processngx_setproctitle("worker process");// 死循环处理 worker 事务for ( ;; ) {// ⼤部分逻辑在接受 master 传递过来折命令if (ngx_exiting) {if (ngx_event_no_timers_left() == NGX_OK) {ngx_log_error(NGX_LOG_NOTICE, cycle->log, 0, "exiting");ngx_worker_process_exit(cycle);}}ngx_log_debug0(NGX_LOG_DEBUG_EVENT, cycle->log, 0, "worker cycle");// 这是其核⼼任务，检测事件、处理事件ngx_process_events_and_timers(cycle);// ⼤部分逻辑在接受 master 传递过来折命令if (ngx_terminate) {ngx_log_error(NGX_LOG_NOTICE, cycle->log, 0, "exiting");ngx_worker_process_exit(cycle);}// 退出事件if (ngx_quit) {ngx_quit = 0;ngx_log_error(NGX_LOG_NOTICE, cycle->log, 0,"gracefully shutting down");ngx_setproctitle("worker process is shutting down");if (!ngx_exiting) {ngx_exiting = 1;ngx_set_shutdown_timer(cycle);ngx_close_listening_sockets(cycle);ngx_close_idle_connections(cycle);}}// reopen 事件if (ngx_reopen) {ngx_reopen = 0;ngx_log_error(NGX_LOG_NOTICE, cycle->log, 0, "reopening logs");ngx_reopen_files(cycle, -1);}}} 上⾯就是nginx worker的主要功能体现, 使⽤⼀个死循环提供服务. 有很多是接⼝master命令进⾏响应的逻辑, 咱们忽略其对master命令的响应，观其业务核⼼: ngx_process_events_and_timers .// event/ngx_event.c// nginx worker 处理io事件和超时队列流程voidngx_process_events_and_timers(ngx_cycle_t *cycle){ngx_uint_t flags;ngx_msec_t timer, delta;if (ngx_timer_resolution) {timer = NGX_TIMER_INFINITE;flags = 0;} else {// 获取timertimer = ngx_event_find_timer();flags = NGX_UPDATE_TIME;#if (NGX_WIN32)/* handle signals from master in case of network inactivity */if (timer == NGX_TIMER_INFINITE || timer > 500) {timer = 500;}#endif}// 使⽤锁进⾏ tcp 监听// 该锁基于 shm 实现，多进程共享内存if (ngx_use_accept_mutex) {// disabled ⽤于优化监听锁竞争，直到 ngx_accept_disabled ⼩于0if (ngx_accept_disabled > 0) {ngx_accept_disabled--;} else {// 通过 shm 获取⼀个进程锁，没抢到锁则直接返回了// 获取到accept锁之后，其会注册 read 事件监听，所以，当其返回后，则意味着数据就绪if (ngx_trylock_accept_mutex(cycle) == NGX_ERROR) {return;}// 获取到锁，设置 flagsif (ngx_accept_mutex_held) {flags |= NGX_POST_EVENTS;} else {if (timer == NGX_TIMER_INFINITE|| timer > ngx_accept_mutex_delay){timer = ngx_accept_mutex_delay;}}}}// post 事件队列不为空，则触发事件处理if (!ngx_queue_empty(&ngx_posted_next_events)) {ngx_event_move_posted_next(cycle);timer = 0;}delta = ngx_current_msec;// 处理事件 ngx_event_actions.process_events, 将会进⾏阻塞等待// 此处的 ngx_event_actions 由系统决定如何初始化，如 linux 下// 使⽤ event/modules/ngx_epoll_module.c 中的定义 ngx_event_actions = ngx_epoll_module_ctx.actions;// ⽽其他系统则另外决定, 总体来说可能有以下⼏种可能// ngx_devpoll_module_ctx.actions;// ngx_epoll_module_ctx.actions;// ngx_eventport_module_ctx.actions;// ngx_iocp_module_ctx.actions;// ngx_kqueue_module_ctx.actions;// ngx_select_module_ctx.actions;// ngx_poll_module_ctx.actions;/*** 其定义样例如下:static ngx_event_module_t ngx_select_module_ctx = {&select_name,NULL, /* create configuration */ngx_select_init_conf, /* init configuration */{ngx_select_add_event, /* add an event */ngx_select_del_event, /* delete an event */ngx_select_add_event, /* enable an event */ngx_select_del_event, /* disable an event */NULL, /* add an connection */NULL, /* delete an connection */NULL, /* trigger a notify */ngx_select_process_events, /* process the events */ngx_select_init, /* init the events */ngx_select_done /* done the events */}};*/(void) ngx_process_events(cycle, timer, flags);// 计算耗时delta = ngx_current_msec - delta;ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0,"timer delta: %M", delta);// 处理 posted 事件，它存放在 ngx_posted_accept_events 队列中ngx_event_process_posted(cycle, &ngx_posted_accept_events);// 处理完事件后，释放锁if (ngx_accept_mutex_held) {ngx_shmtx_unlock(&ngx_accept_mutex);}// 处理超时的任务if (delta) {ngx_event_expire_timers();}// 读写事件将会被添加到 ngx_posted_events 队列中ngx_event_process_posted(cycle, &ngx_posted_events);} 以上也就是nginx worker的主要功能框架了: 1. 先通过shm获取tcp的监听锁, 避免socket惊群; 2. 获取到锁的worker进程, 将会注册accept的read事件，没有抢到锁的进程不会⽴即返回，因为他还可以继续处理其他事件，以及在之前被监听到的socket（此处io事件处理决定了worker不会进⾏空转）; 3. 如果有 ngx_posted_next_events 队列, 则先处理其队列请求; 4. 根据系统类型调⽤⽹络io模块, select 机制接收io事件; 5. 接⼊accept事件后, 释放accept锁(基于shm); 6. 处理过期超时队列; 7. 处理普通的已接⼊的socket的读写事件; ⼀次处理往往只会处理部分事件, ⽐如可能只是处理了 accept, read 则需要在下⼀次或n次之后才会处理, 这也是异步机制⾮阻塞的体现.1. worker 时序图 下⾯我先给到⼀个整个worker的⼯作时序图, 以便有个整体的认知. 接下来我们从⼏个点依次简单看看 nginx 是如何处理各细节的.2. 获取accept锁及注册accept事件 由于nginx是基于多进程实现的并发处理, 那么各进程必然都需要监听相同的端⼝数据, 如果没有锁控制, 则当有事件到达时, 必然导致各进程同时被唤醒, 即所谓的惊群. 所以, nginx 提供了⼀个锁机制, 使同⼀时刻只有⼀个进程在监听某端⼝, 从⽽避免竞争. 实现⽅式是基于共享内存 shm 实现.(如果是多线程⽅式会更简单哟)// event/ngx_event_accept.cngx_int_tngx_trylock_accept_mutex(ngx_cycle_t *cycle){// ⾸先获取shm锁, 通过 shm 实现进程数据共享if (ngx_shmtx_trylock(&ngx_accept_mutex)) {ngx_log_debug0(NGX_LOG_DEBUG_EVENT, cycle->log, 0,"accept mutex locked");// 如果上⼀次就是⾃⼰执⾏的accept操作, 则直接返回// 否则需要重新注册accept监听if (ngx_accept_mutex_held && ngx_accept_events == 0) {return NGX_OK;}// 注册 accept 事件if (ngx_enable_accept_events(cycle) == NGX_ERROR) {ngx_shmtx_unlock(&ngx_accept_mutex);return NGX_ERROR;}ngx_accept_events = 0;ngx_accept_mutex_held = 1;return NGX_OK;}ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0,"accept mutex lock failed: %ui", ngx_accept_mutex_held);if (ngx_accept_mutex_held) {// 如果没有获取到锁，则将之前注册的 accept 事件取消，避免惊群if (ngx_disable_accept_events(cycle, 0) == NGX_ERROR) {return NGX_ERROR;}ngx_accept_mutex_held = 0;}// 不管有没有获取到锁, 都会执⾏后续的逻辑, 因为除了 accept 外, 还有read/write事件需要处理return NGX_OK;}// core/ngx_shmtx.c, 获取锁，锁的值为当前进程idngx_uint_tngx_shmtx_trylock(ngx_shmtx_t *mtx){return (*mtx->lock == 0 && ngx_atomic_cmp_set(mtx->lock, 0, ngx_pid));}// 注册 accept 事件监听// event/ngx_event_accept.cngx_int_tngx_enable_accept_events(ngx_cycle_t *cycle){ngx_uint_t i;ngx_listening_t *ls;ngx_connection_t *c;ls = cycle->listening.elts;for (i = 0; i < cycle->listening.nelts; i++) {c = ls[i].connection;if (c == NULL || c->read->active) {continue;}// 注册accept事件，READ ?// 交由 ngx_event_actions.add 处理, 实际运⾏由系统决定, 如 ngx_select_add_event if (ngx_add_event(c->read, NGX_READ_EVENT, 0) == NGX_ERROR) {return NGX_ERROR;}}return NGX_OK;}// event/module/ngx_select_module.c// 注册⼀个 io 事件监听, fd_setstatic ngx_int_tngx_select_add_event(ngx_event_t *ev, ngx_int_t event, ngx_uint_t flags){ngx_connection_t *c;c = ev->data;ngx_log_debug2(NGX_LOG_DEBUG_EVENT, ev->log, 0,"select add event fd:%d ev:%i", c->fd, event);if (ev->index != NGX_INVALID_INDEX) {ngx_log_error(NGX_LOG_ALERT, ev->log, 0,"select event fd:%d ev:%i is already set", c->fd, event);return NGX_OK;}if ((event == NGX_READ_EVENT && ev->write)|| (event == NGX_WRITE_EVENT && !ev->write)){ngx_log_error(NGX_LOG_ALERT, ev->log, 0,"invalid select %s event fd:%d ev:%i",ev->write ? "write" : "read", c->fd, event);return NGX_ERROR;}if (event == NGX_READ_EVENT) {FD_SET(c->fd, &master_read_fd_set);} else if (event == NGX_WRITE_EVENT) {FD_SET(c->fd, &master_write_fd_set);}if (max_fd != -1 && max_fd < c->fd) {max_fd = c->fd;}ev->active = 1;event_index[nevents] = ev;ev->index = nevents;nevents++;return NGX_OK;} 主要就是shm的应⽤，以及fd_set处理。
