1. Nginx 频道
ngx_channel_t 频道是 Nginx master 进程与 worker 进程之间通信的常用工具，它是使用本机套接字实现的，即 socketpair 方法，它用于创建父子进程间使用的套接字。
#include/* See NOTES */#include int socketpair(int domain, int type, int protocol, int sv[2]);
这个方法可以创建一对关联的套接字 sv[2]。
- domain：表示域，在 Linux 下通常取值为 AF_UNIX；
- type：取值为 SOCK_STREAM 或 SOCK_DGRAM，它表示在套接字上使用的是 TCP 还是 UDP；
- protocol：必须传递 0；
- sv[2]：是一个含有两个元素的整型数组，实际上就是两个套接字。
- 当 socketpair 返回 0 时，sv[2] 这两个套接字创建成功，否则 sockpair 返回 -1 表示失败.
当 socketpair 执行成功时，sv[2] 这两个套接字具备下列关系：
- 向 sv[0] 套接字写入数据，将可以从 sv[1] 套接字中读取到刚写入的数据；
同样，向 sv[1] 套接字写入数据，也可以从 sv[0] 中读取到写入的数据。
- 通常，在父、子进程通信前，会先调用 socketpair 方法创建这样一组套接字，在调用 fork 方法创建出子进程后，将会在父进程中关闭 sv[1] 套接字，仅使用 sv[0] 套接字用于向子进程发送数据以及接收子进程发送来的数据；
而在子进程中则关闭 sv[0] 套接字，仅使用 sv[1] 套接字既可以接收父进程发送来的数据，也可以向父进程发送数据。
ngx_channel_t 结构体是 Nginx 定义的 master 父进程与 worker 子进程间的消息格式，如下：
typedef struct { // 传递的 TCP 消息中的命令 ngx_uint_t command; // 进程 ID，一般是发送命令方的进程 ID ngx_pid_t pid; // 表示发送命令方在 ngx_processes 进程数组间的序号 ngx_int_t slot; // 通信的套接字句柄 ngx_fd_t fd;}ngx_channel_t; Nginx 针对 command 成员定义了如下命令：
// 打开频道，使用频道这种方式通信前必须发送的命令#define NGX_CMD_OPEN_CHANNEL 1// 关闭已经打开的频道，实际上也就是关闭套接字#define NGX_CMD_CLOSE_CHANNEL 2// 要求接收方正常地退出进程#define NGX_CMD_QUIT 3// 要求接收方强制地结束进程#define NGX_CMD_TERMINATE 4// 要求接收方重新打开进程已经打开过的文件#define NGX_CMD_REOPEN 5
问：master 是如何启动、停止 worker 子进程的？
答：正是通过 socketpair 产生的套接字发送命令的，即每次要派生一个子进程之前，都会先调用 socketpair 方法。在 Nginx 派生子进程的 ngx_spawn_process 方法中，会首先派生基于 TCP 的套接字，如下：
ngx_pid_t ngx_spawn_process(ngx_cycle_t *cycle, ngx_spawn_proc_pt proc, void *data, char *name, ngx_int_t respawn){ if (respawn != NGX_PROCESS_DETACHED) { /* Solaris 9 still has no AF_LOCAL */ // ngx_processes[s].channel 数组正是将要用于父、子进程间通信的套接字对 if (socketpair(AF_UNIX, SOCK_STREAM, 0, ngx_processes[s].channel) == -1) { return NGX_INVALID_PID; } // 将 channel 套接字对都设置为非阻塞模式 if (ngx_nonblocking(ngx_processes[s].channel[0]) == -1) { ngx_close_channel(ngx_processes[s].channel, cycle->log); return NGX_INVALID_PID; } if (ngx_nonblocking(ngx_processes[s].channel[1]) == -1) { ngx_close_channel(ngx_processes[s].channel, cycle->log); return NGX_INVALID_PID; } ...} ngx_processes 数组定义了 Nginx 服务中所有的进程，包括 master 进程和 worker 进程，如下：
#define NGX_MAX_PROCESSES 1024// 虽然定义了 NGX_MAX_PROCESSES 个成员，但是已经使用的元素仅与启动的进程个数有关ngx_processes_t ngx_processes[NGX_MAX_PROCESSES];
ngx_processes 数组的类型是 ngx_processes_t，对于频道来说，这个结构体只关心它的 channel 成员：
typedef struct { ... // socketpair 创建的套接字对 ngx_socket_t channel[2];}ngx_processes_t; 1. ngx_write_channel：使用频道发送 ngx_channel_t 消息
ngx_int_t ngx_write_channel(ngx_socket_t s, ngx_channel_t *ch, size_t size, ngx_log_t *log){ ssize_t n; ngx_err_t err; struct iovec iov[1]; struct msghdr msg; #if (NGX_HAVE_MSGHDR_MSG_CONTROL) union { struct cmsghdr cm; char space[CMSG_SPACE(sizeof(int))]; }cmsg; if (ch->fd == -1) { msg.msg_control = NULL; msg.msg_controllen = 0; } else { // 辅助数据 msg.msg_control = (caddr_t)&cmsg; msg.msg_controllen = sizeof(cmsg); ngx_memzero(&cmsg, sizeof(cmsg)); cmsg.cm.cmsg_len = CMSG_LEN(sizeof(int)); cmsg.cm.cmsg_level = SOL_SOCKET; cmsg.cm.cmsg_type = SCM_RIGHTS; /* * We have to use ngx_memcpy() instead of simple * *(int *) CMSG_DATA(&cmsg.cm) = ch->fd; * because some gcc 4.4 with -O2/3/s optimization issues the warning: * dereferencing type-punned pointer will break strict-aliasing rules * * Fortunately, gcc with -O1 compiles this ngx_memcpy() * in the same simple assignment as in the code above */ ngx_memcpy(CMSG_DATA(&cmsg.cm), &ch->fd, sizeof(int)); } msg.msg_flags = 0; #else if (ch->fd == -1) { msg.msg_accrights = NULL; msg.msg_accrightslen = 0; } else { msg.msg_accrights = (caddr_t) &ch->fd; msg.msg_accrightslen = sizeof(int); } #endif // 指向要发送的 ch 起始地址 iov[0].iov_base = (char *) ch; iov[0].iov_len = size; // msg_name 和 msg_namelen 仅用于未连接套接字(如UDP) msg.msg_name = NULL; msg.msg_namelen = 0; msg.msg_iov = iov; msg.msg_iovlen = 1; // 将该 ngx_channel_t 消息发出去 n = sendmsg(s, &msg, 0); if (n == -1) { err = ngx_errno; if (err == NGX_EAGAIN) { return NGX_AGAIN; } return NGX_ERROR; } return NGX_OK;} 2. ngx_read_channel: 读取消息
ngx_int_tngx_read_channel(ngx_socket_t s, ngx_channel_t *ch, size_t size, ngx_log_t *log){ ssize_t n; ngx_err_t err; struct iovec iov[1]; struct msghdr msg;#if (NGX_HAVE_MSGHDR_MSG_CONTROL) union { struct cmsghdr cm; char space[CMSG_SPACE(sizeof(int))]; } cmsg;#else int fd;#endif iov[0].iov_base = (char *)ch; iov[0].iov_len = size; msg.msg_name = NULL; msg.msg_namelen = 0; msg.msg_iov = iov; msg.msg_iovlen = 1; #if (NGX_HAVE_MSGHDR_MSG_CONTROL) msg.msg_control = (caddr_t) &cmsg; msg.msg_controllen = sizeof(cmsg);#else msg.msg_accrights = (caddr_t) &fd; msg.msg_accrightslen = sizeof(int);#endif // 接收命令 n = recvmsg(s, &msg, 0); if (n == -1) { err = ngx_errno; if (err == NGX_EAGAIN) { return NGX_AGAIN; } return NGX_ERROR; } if (n == 0) { return NGX_ERROR; } // 接收的数据不足 if ((size_t) n < sizeof(ngx_channel_t)) { return NGX_ERROR; } #if (NGX_HAVE_MSGHDR_MSG_CONTROL) // 若接收到的命令为"打开频道，使用频道这种方式通信前必须发送的命令" if (ch->command == NGX_CMD_OPEN_CHANNEL) { if (cmsg.cm.cmsg_len < (socklen_t) CMSG_LEN(sizeof(int))) { return NGX_ERROR; } if (cmsg.cm.cmsg_level != SOL_SOCKET || cmsg.cm.cmsg_type != SCM_RIGHTS) { return NGX_ERROR; } /* ch->fd = *(int *) CMSG_DATA(&cmsg.cm); */ ngx_memcpy(&ch->fd, CMSG_DATA(&cmsg.cm), sizeof(int)); } // 若接收到的消息是被截断的 if (msg.msg_flags & (MSG_TRUNC|MSG_CTRUNC)) { ngx_log_error(NGX_LOG_ALERT, log, 0, "recvmsg() truncated data"); } #else if (ch->command == NGX_CMD_OPEN_CHANNEL) { if (msg.msg_accrightslen != sizeof(int)) { return NGX_ERROR; } ch->fd = fd; }#endif return n;} 在 Nginx 中，目前仅存在 master 进程向 worker 进程发送消息的场景，这时对于 socketpair 方法创建的 channel[2] 套接字来说，master 进程会使用 channel[0] 套接字来发送消息，而 worker 进程会使用 channel[1] 套接字来接收消息。
3. ngx_add_channel_event: 把接收频道消息的套接字添加到 epoll 中
worker 进程调度 ngx_read_channel 方法接收频道消息是通过该 ngx_add_channel_event 函数将接收频道消息的套接字（对于 worker 即为channel[1]）添加到 epoll 中，当接收到父进程消息时子进程会通过 epoll 的事件回调相应的 handler 方法来处理这个频道消息，如下：
ngx_int_t ngx_add_channel_event(ngx_cycle_t *cycle, ngx_fd_t fd, ngx_int_t event, ngx_event_handler_pt handler){ ngx_event_t *ev, *rev, *wev; ngx_connection_t *c; // 获取一个空闲连接 c = ngx_get_connection(fd, cycle->log); if (c == NULL) { return NGX_ERROR; } c->pool = cycle->pool; rev = c->read; wev = c->write; rev->log = cycle->log; wev->log = cycle->log; rev->channel = 1; wev->channel = 1; ev = (event == NGX_READ_EVENT) ? rev : wev; // 初始化监听该 ev 事件时调用的回调函数 ev->handler = handler; // 将该接收频道消息的套接字添加到 epoll 中 if (ngx_add_conn && (ngx_event_flags && NGX_USE_EPOLL_EVENT) == 0) { // 这里是同时监听该套接字的读、写事件 if (ngx_add_conn(c) == NGX_ERROR) { ngx_free_connection(c); return NGX_ERROR; } } else { // 这里是仅监听 ev 事件 if (ngx_add_event(ev, event, 0) == NGX_ERROR) { ngx_free_connection(c); return NGX_ERROR; } } return NGX_OK;} 4. ngx_close_channel: 关闭这个频道通信方式
voidngx_close_channel(ngx_fd_t *fd, ngx_log_t *log){ if (close(fd[0]) == -1) { } if (close(fd[1]) == -1) { }}