惊群现象的处理与负载均衡的实现
2015-03-14 16:43:10 最后更新: 2015-03-14 16:43:10 访问数量：905
事件驱动函数 ngx_process_events_and_timers
初始化流程结束后，设置进程名为 nginx: worker process 后就进入主循环
在进入主循环之前，如果 nginx 是以线程模式启动的，则首先会进行线程初始化，这里我们不做讲解，后续讲解线程模式 nginx 的时候我们再重新会看这里的代码
// void ngx_process_events_and_timers(ngx_cycle_t *cycle)
// 事件驱动函数 {{{
void
ngx_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 {
timer = ngx_event_find_timer();
flags = NGX_UPDATE_TIME;
#if (NGX_THREADS)
if (timer == NGX_TIMER_INFINITE || timer > 500) {
timer = 500;
}
#endif
}
// 判断是否使用 accept 互斥体，用于避免惊群现象
if (ngx_use_accept_mutex) {
// 如果进程接受的链接太多，则放弃一次
if (ngx_accept_disabled > 0) {
ngx_accept_disabled--;
} else {
// 尝试获取 accept 锁，避免惊群现象
if (ngx_trylock_accept_mutex(cycle) == NGX_ERROR) {
return;
}
if (ngx_accept_mutex_held) {
flags |= NGX_POST_EVENTS;
} else {
// 没有获取到锁，则延迟一段时间后重新尝试获取锁
if (timer == NGX_TIMER_INFINITE
|| timer > ngx_accept_mutex_delay)
{
timer = ngx_accept_mutex_delay;
}
}
}
}
delta = ngx_current_msec;
// ngx_epoll_process_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);
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_event_process_posted(cycle, &ngx_posted_events);
} // }}}M<>
在 worker 的主循环中，主要工作都集中在这个函数中
在这个函数中，nginx 进行了事件驱动的完整过程
惊群现象的处理
// 判断是否使用 accept 互斥体，用于避免惊群现象
if (ngx_use_accept_mutex) {
// 如果进程接受的链接太多，则放弃一次
if (ngx_accept_disabled > 0) {
ngx_accept_disabled--;
} else {
// 尝试获取 accept 锁，避免惊群现象
if (ngx_trylock_accept_mutex(cycle) == NGX_ERROR) {
return;
}
if (ngx_accept_mutex_held) {
flags |= NGX_POST_EVENTS;
} else {
// 没有获取到锁，则延迟一段时间后重新尝试获取锁
if (timer == NGX_TIMER_INFINITE
|| timer > ngx_accept_mutex_delay)
{
timer = ngx_accept_mutex_delay;
}
}
}
}
这段代码首先判断是否开启了 accept_mutex 锁，这个锁正是用来解决惊群现象的，如果这个锁被关闭，则 nginx 不对惊群现象做处理
关于惊群现象可以参考：
尝试获取 accept 锁
程序调用 ngx_trylock_accept_mutex 函数尝试获取 accept 锁
// ngx_int_t ngx_trylock_accept_mutex(ngx_cycle_t *cycle)
// 尝试获取锁，返回是否获取锁成功 {{{
ngx_int_t
ngx_trylock_accept_mutex(ngx_cycle_t *cycle)
{
if (ngx_shmtx_trylock(&ngx_accept_mutex)) {
ngx_log_debug0(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"accept mutex locked");
if (ngx_accept_mutex_held
&& ngx_accept_events == 0
&& !(ngx_event_flags & NGX_USE_RTSIG_EVENT))
{
return NGX_OK;
}
// 将全部监听连接的读事件添加到当前 epoll 事件模块中
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) {
if (ngx_disable_accept_events(cycle) == NGX_ERROR) {
return NGX_ERROR;
}
ngx_accept_mutex_held = 0;
}
return NGX_OK;
} // }}}
// static ngx_int_t ngx_enable_accept_events(ngx_cycle_t *cycle)
// 将连接池中的所有连接的读事件全部添加到 epoll 模块中 {{{
static ngx_int_t
ngx_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->read->active) {
continue;
}
if (ngx_event_flags & NGX_USE_RTSIG_EVENT) {
if (ngx_add_conn(c) == NGX_ERROR) {
return NGX_ERROR;
}
} else {
if (ngx_add_event(c->read, NGX_READ_EVENT, 0) == NGX_ERROR) {
return NGX_ERROR;
}
}
}
return NGX_OK;
} // }}}
未获取锁处理
如果未能获取到锁，则该子进程需要等待一个 delay 后才能够重新尝试获取
他将只能处理已有连接上的事件
但是，如果锁被长期占用，那么新的连接总无法被该进程获取，这样当然也是不可取的
于是在全局变量 flags 中加入了 NGX_POST_EVENTS 标志
在接下来即将调用的 ngx_epoll_process_events 函数中，如果 flags 已经设置了该标志位，这就意味着有其他子进程正在等待获取锁，于是，该进程将不会立即调用事件的 handler 回调函数
这样大大减少了 ngx_accept_mutex 锁占用的时间
nginx 的负载均衡
上面代码中有一个 if 判断被我们忽略了
// 如果进程接受的链接太多，则放弃一次
if (ngx_accept_disabled > 0) {
ngx_accept_disabled--;
}
这个 if 判断是用来做什么呢？
如果 ngx_accept_disabled > 0 则不去尝试获取 accept 锁
在 ngx_event_module 初始化过程中，新连接事件的回调函数被初始化为 ngx_event_accept 函数
在这个函数中，每当新的连接进入，都会执行
ngx_accept_disabled = ngx_cycle->connection_n / 8
- ngx_cycle->free_connection_n;
这样，如果该子进程的连接池已经使用 7/8，则不再争抢互斥锁，保证了负载均衡的实现