微信公众号:郑尔多斯
关注可了解更多的Nginx
知识。任何问题或建议,请公众号留言;
关注公众号,有趣有内涵的文章第一时间送达!
事件驱动框架
我们前面分析过,在ngx_event_process_init()
中,会将每个监听端口的rev->handler
设置为ngx_event_accept()
函数,然后把对应的读事件
加入到epoll
驱动模块中。这样,在执行ngx_epoll_process_events()
的时候,如果有新连接事件出现,那么就会调用ngx_event_accept()
方法进行建立连接。
Nginx
为了解决惊群现象和负载均衡问题,使用了accept_mutex
以及post
事件处理机制,那么什么是post
事件处理机制?
所谓的post
事件处理机制就是允许事件延后执行。Nginx
设计了两个队列,一个是由被触发的监听连接的读事件构成的ngx_posted_accept_events
队列,另一个是由普通读/写事件构成的ngx_posted_events
队列。
将
epoll_wait()
产生的事件,分别存放到这两个队列中。让存放新连接事件的ngx_posted_accept_events
队列优先执行,存放普通事件的ngx_posted_events
队列最后执行,这样就可以保证建立连接的及时性,同时这也是解决惊群和负载均衡的关键所在。
ngx_event_accept
void
ngx_event_accept(ngx_event_t *ev)
{
socklen_t socklen;
ngx_err_t err;
ngx_log_t *log;
ngx_uint_t level;
ngx_socket_t s;
ngx_event_t *rev, *wev;
ngx_sockaddr_t sa;
ngx_listening_t *ls;
ngx_connection_t *c, *lc;
ngx_event_conf_t *ecf;
#if (NGX_HAVE_ACCEPT4)
static ngx_uint_t use_accept4 = 1;
#endif
if (ev->timedout) {
if (ngx_enable_accept_events((ngx_cycle_t *) ngx_cycle) != NGX_OK) {
return;
}
ev->timedout = 0;
}
ecf = ngx_event_get_conf(ngx_cycle->conf_ctx, ngx_event_core_module);
if (!(ngx_event_flags & NGX_USE_KQUEUE_EVENT)) {
ev->available = ecf->multi_accept;
}
// 事件关联的data,也即当前事件对应的connection
lc = ev->data;
// connection对应的监听端口结构
ls = lc->listening;
// ready=0,表示当前事件没有就绪,因为我们已经消费了读事件
ev->ready = 0;
do {
socklen = sizeof(ngx_sockaddr_t);
// 使用accept创建一个连接
s = accept(lc->fd, &sa.sockaddr, &socklen);
if (s == (ngx_socket_t) -1) {
/*accept()失败的时候处理逻辑,这里可以忽略*/
return;
}
#if (NGX_STAT_STUB)
/*这里是用于统计每个worker进程的统计数据,比如accept的数量等*/
(void) ngx_atomic_fetch_add(ngx_stat_accepted, 1);
#endif
ngx_accept_disabled = ngx_cycle->connection_n / 8
- ngx_cycle->free_connection_n;
c = ngx_get_connection(s, ev->log);
if (c == NULL) {
/*获取connection失败*/
return;
}
c->type = SOCK_STREAM;
c->pool = ngx_create_pool(ls->pool_size, ev->log);
if (c->pool == NULL) {
ngx_close_accepted_connection(c);
return;
}
if (socklen > (socklen_t) sizeof(ngx_sockaddr_t)) {
socklen = sizeof(ngx_sockaddr_t);
}
c->sockaddr = ngx_palloc(c->pool, socklen);
if (c->sockaddr == NULL) {
ngx_close_accepted_connection(c);
return;
}
ngx_memcpy(c->sockaddr, &sa, socklen);
log = ngx_palloc(c->pool, sizeof(ngx_log_t));
if (log == NULL) {
ngx_close_accepted_connection(c);
return;
}
/* set a blocking mode for iocp and non-blocking mode for others */
if (ngx_inherited_nonblocking) {
if (ngx_event_flags & NGX_USE_IOCP_EVENT) {
/*这里是windows的处理逻辑*/
}
} else {
if (!(ngx_event_flags & NGX_USE_IOCP_EVENT)) {
if (ngx_nonblocking(s) == -1) {
ngx_close_accepted_connection(c);
return;
}
}
}
*log = ls->log;
c->recv = ngx_recv;
c->send = ngx_send;
c->recv_chain = ngx_recv_chain;
c->send_chain = ngx_send_chain;
c->log = log;
c->pool->log = log;
c->socklen = socklen;
c->listening = ls;
c->local_sockaddr = ls->sockaddr;
c->local_socklen = ls->socklen;
#if (NGX_HAVE_UNIX_DOMAIN)
if (c->sockaddr->sa_family == AF_UNIX) {
c->tcp_nopush = NGX_TCP_NOPUSH_DISABLED;
c->tcp_nodelay = NGX_TCP_NODELAY_DISABLED;
#if (NGX_SOLARIS)
/* Solaris's sendfilev() supports AF_NCA, AF_INET, and AF_INET6 */
c->sendfile = 0;
#endif
}
#endif
rev = c->read;
wev = c->write;
wev->ready = 1;
if (ngx_event_flags & NGX_USE_IOCP_EVENT) {
rev->ready = 1;
}
if (ev->deferred_accept) {
rev->ready = 1;
#if (NGX_HAVE_KQUEUE || NGX_HAVE_EPOLLRDHUP)
rev->available = 1;
#endif
}
rev->log = log;
wev->log = log;
c->number = ngx_atomic_fetch_add(ngx_connection_counter, 1);
if (ls->addr_ntop) {
c->addr_text.data = ngx_pnalloc(c->pool, ls->addr_text_max_len);
if (c->addr_text.data == NULL) {
ngx_close_accepted_connection(c);
return;
}
c->addr_text.len = ngx_sock_ntop(c->sockaddr, c->socklen,
c->addr_text.data,
ls->addr_text_max_len, 0);
if (c->addr_text.len == 0) {
ngx_close_accepted_connection(c);
return;
}
}
if (ngx_add_conn && (ngx_event_flags & NGX_USE_EPOLL_EVENT) == 0) {
if (ngx_add_conn(c) == NGX_ERROR) {
ngx_close_accepted_connection(c);
return;
}
}
log->data = NULL;
log->handler = NULL;
ls->handler(c);
} while (ev->available);
}
上面的代码我做了删减,把一些错误判断和非Linux
平台代码删除,这个并不影响代码的功能和阅读,现在我们简单的分析一下这个源码:
- 首先这个是一个
do...while
循环,循环的条件就是ev-\>available
,该条件控制是否一次建立多个连接。 - 循环体首先使用
accept()
函数创建一个连接 - 设置
ngx_accept_disabled
变量,然后调用ngx_get_connection()
获取一个连接。 - 下面就是设置我们上一步获取的连接的各个属性。
最重要的一步是最后的代码:
ls->handler(c);
我们前面知道,ls->handler
其实就是ngx_http_init_connection
,从这个函数开始,我们就进入了http
初始化的过程了。
multi_accept的作用
Syntax: multi_accept on | off;
Default: multi_accept off;
Context: events
If multi_accept is disabled, a worker process will accept one new connection at a time. Otherwise, a worker process will accept all new connections at a time.
The directive is ignored if kqueue connection processing method is used, because it reports the number of new connections waiting to be accepted.
中文翻译:
如果multi_accept被禁止了,nginx一个工作进程只能同时接受一个新的连接。否则,一个工作进程可以同时接受所有的新连接。
如果nginx使用kqueue连接方法,那么这条指令会被忽略,因为这个方法会报告在等待被接受的新连接的数量。
如果启用了这个参数,那么当一个worker进程获取到accept_mutex之后,会一次尽可能的多的建立连接。
喜欢本文的朋友们,欢迎长按下图关注订阅号郑尔多斯,更多精彩内容第一时间送达