剖析nodejs的事件循环

时间 2019-11-06

标签剖析 nodejs 事件循环繁體版

原文原文链接

本文首发在github，感兴趣请点击此处javascript

nodejs是单线程执行的，同时它又是基于事件驱动的非阻塞IO编程模型。这就使得咱们不用等待异步操做结果返回，就能够继续往下执行代码。当异步事件触发以后，就会通知主线程，主线程执行相应事件的回调。java

以上是众所周知的内容。今天咱们从源码入手，分析一下nodejs的事件循环机制。node

nodejs架构

首先，咱们先看下nodejs架构，下图所示：linux

如上图所示，nodejs自上而下分为

用户代码 ( js 代码 )

用户代码即咱们编写的应用程序代码、npm包、nodejs内置的js模块等，咱们平常工做中的大部分时间都是编写这个层面的代码。git

binding代码或者三方插件（js 或 C/C++ 代码）

胶水代码，可以让js调用C/C++的代码。能够将其理解为一个桥，桥这头是js，桥那头是C/C++，经过这个桥可让js调用C/C++。
在nodejs里，胶水代码的主要做用是把nodejs底层实现的C/C++库暴露给js环境。
三方插件是咱们本身实现的C/C++库，同时须要咱们本身实现胶水代码，将js和C/C++进行桥接。github

底层库

nodejs的依赖库，包括大名鼎鼎的V八、libuv。
V8：咱们都知道，是google开发的一套高效javascript运行时，nodejs可以高效执行 js 代码的很大缘由主要在它。
libuv：是用C语言实现的一套异步功能库，nodejs高效的异步编程模型很大程度上归功于libuv的实现，而libuv则是咱们今天重点要分析的。
还有一些其余的依赖库
http-parser：负责解析http响应
openssl：加解密
c-ares：dns解析
npm：nodejs包管理器
...npm

关于nodejs再也不过多介绍，你们能够自行查阅学习，接下来咱们重点要分析的就是libuv。编程

libuv 架构

咱们知道，nodejs实现异步机制的核心即是libuv，libuv承担着nodejs与文件、网络等异步任务的沟通桥梁，下面这张图让咱们对libuv有个大概的印象： windows

这是libuv官网的一张图，很明显，nodejs的网络I/O、文件I/O、DNS操做、还有一些用户代码都是在 libuv 工做的。既然谈到了异步，那么咱们首先概括下nodejs里的异步事件：promise

非I/O：
- 定时器（setTimeout，setInterval）
- microtask（promise）
- process.nextTick
- setImmediate
- DNS.lookup
I/O：
- 网络I/O
- 文件I/O
- 一些DNS操做
...

网络I/O

对于网络I/O，各个平台的实现机制不同，linux 是 epoll 模型，类 unix 是 kquene 、windows 下是高效的 IOCP 完成端口、SunOs 是 event ports，libuv 对这几种网络I/O模型进行了封装。

文件I/O、异步DNS操做

libuv内部还维护着一个默认4个线程的线程池，这些线程负责执行文件I/O操做、DNS操做、用户异步代码。当 js 层传递给 libuv 一个操做任务时，libuv 会把这个任务加到队列中。以后分两种状况：

一、线程池中的线程都被占用的时候，队列中任务就要进行排队等待空闲线程。
二、线程池中有可用线程时，从队列中取出这个任务执行，执行完毕后，线程归还到线程池，等待下个任务。同时以事件的方式通知event-loop，event-loop接收到事件执行该事件注册的回调函数。

固然，若是以为4个线程不够用，能够在nodejs启动时，设置环境变量UV_THREADPOOL_SIZE来调整，出于系统性能考虑，libuv 规定可设置线程数不能超过128个。

nodejs源码

先简要介绍下nodejs的启动过程：

一、调用platformInit方法，初始化 nodejs 的运行环境。
二、调用 performance_node_start 方法，对 nodejs 进行性能统计。
三、openssl设置的判断。
四、调用v8_platform.Initialize，初始化 libuv 线程池。
五、调用 V8::Initialize，初始化 V8 环境。
六、建立一个nodejs运行实例。
七、启动上一步建立好的实例。
八、开始执行js文件，同步代码执行完毕后，进入事件循环。
九、在没有任何可监听的事件时，销毁 nodejs 实例，程序执行完毕。

以上就是 nodejs 执行一个js文件的全过程。接下来着重介绍第八个步骤，事件循环。

咱们看几处关键源码：

一、core.c，事件循环运行的核心文件。

int uv_run(uv_loop_t* loop, uv_run_mode mode) {
  int timeout;
  int r;
  int ran_pending;
//判断事件循环是否存活。
  r = uv__loop_alive(loop);
  //若是没有存活，更新时间戳
  if (!r)
    uv__update_time(loop);
//若是事件循环存活，而且事件循环没有中止。
  while (r != 0 && loop->stop_flag == 0) {
    //更新当前时间戳
    uv__update_time(loop);
    //执行 timers 队列
    uv__run_timers(loop);
    //执行因为上个循环未执行完，并被延迟到这个循环的I/O 回调。
    ran_pending = uv__run_pending(loop); 
    //内部调用，用户不care，忽略
    uv__run_idle(loop); 
    //内部调用，用户不care，忽略
    uv__run_prepare(loop); 
    
    timeout = 0; 
    if ((mode == UV_RUN_ONCE && !ran_pending) || mode == UV_RUN_DEFAULT)
    //计算距离下一个timer到来的时间差。
      timeout = uv_backend_timeout(loop);
   //进入 轮询 阶段，该阶段轮询I/O事件，有则执行，无则阻塞，直到超出timeout的时间。
    uv__io_poll(loop, timeout);
    //进入check阶段，主要执行 setImmediate 回调。
    uv__run_check(loop);
    //进行close阶段，主要执行 **关闭** 事件
    uv__run_closing_handles(loop);

    if (mode == UV_RUN_ONCE) {
      
      //更新当前时间戳
      uv__update_time(loop);
      //再次执行timers回调。
      uv__run_timers(loop);
    }
    //判断当前事件循环是否存活。
    r = uv__loop_alive(loop); 
    if (mode == UV_RUN_ONCE || mode == UV_RUN_NOWAIT)
      break;
  }

  /* The if statement lets gcc compile it to a conditional store. Avoids * dirtying a cache line. */
  if (loop->stop_flag != 0)
    loop->stop_flag = 0;

  return r;
}
复制代码

二、timers 阶段，源码文件：timers.c。

void uv__run_timers(uv_loop_t* loop) {
  struct heap_node* heap_node;
  uv_timer_t* handle;

  for (;;) {
  //取出定时器堆中超时时间最近的定时器句柄
    heap_node = heap_min((struct heap*) &loop->timer_heap);
    if (heap_node == NULL)
      break;
    
    handle = container_of(heap_node, uv_timer_t, heap_node);
    // 判断最近的一个定时器句柄的超时时间是否大于当前时间，若是大于当前时间，说明还未超时，跳出循环。
    if (handle->timeout > loop->time)
      break;
    // 中止最近的定时器句柄
    uv_timer_stop(handle);
    // 判判定时器句柄类型是不是repeat类型，若是是，从新建立一个定时器句柄。
    uv_timer_again(handle);
    //执行定时器句柄绑定的回调函数
    handle->timer_cb(handle);
  }
}
复制代码

三、 轮询阶段 源码，源码文件：kquene.c

void uv__io_poll(uv_loop_t* loop, int timeout) {
  /*一连串的变量初始化*/
  //判断是否有事件发生 
  if (loop->nfds == 0) {
    //判断观察者队列是否为空，若是为空，则返回
    assert(QUEUE_EMPTY(&loop->watcher_queue));
    return;
  }
  
  nevents = 0;
  // 观察者队列不为空
  while (!QUEUE_EMPTY(&loop->watcher_queue)) {
    /* 取出队列头的观察者对象 取出观察者对象感兴趣的事件并监听。 */
    ....省略一些代码
    w->events = w->pevents;
  }

  
  assert(timeout >= -1);
  //若是有超时时间，将当前时间赋给base变量
  base = loop->time;
  // 本轮执行监听事件的最大数量
  count = 48; /* Benchmarks suggest this gives the best throughput. */
  //进入监听循环
  for (;; nevents = 0) {
  // 有超时时间的话，初始化spec
    if (timeout != -1) {
      spec.tv_sec = timeout / 1000;
      spec.tv_nsec = (timeout % 1000) * 1000000;
    }
    
    if (pset != NULL)
      pthread_sigmask(SIG_BLOCK, pset, NULL);
    // 监听内核事件，当有事件到来时，即返回事件的数量。
    // timeout 为监听的超时时间，超时时间一到即返回。
    // 咱们知道，timeout是传进来得下一个timers到来的时间差，因此，在timeout时间内，event-loop会一直阻塞在此处，直到超时时间到来或者有内核事件触发。
    nfds = kevent(loop->backend_fd,
                  events,
                  nevents,
                  events,
                  ARRAY_SIZE(events),
                  timeout == -1 ? NULL : &spec);

    if (pset != NULL)
      pthread_sigmask(SIG_UNBLOCK, pset, NULL);

    /* Update loop->time unconditionally. It's tempting to skip the update when * timeout == 0 (i.e. non-blocking poll) but there is no guarantee that the * operating system didn't reschedule our process while in the syscall. */
    SAVE_ERRNO(uv__update_time(loop));
    //若是内核没有监听到可用事件，且本次监听有超时时间，则返回。
    if (nfds == 0) {
      assert(timeout != -1);
      return;
    }
    
    if (nfds == -1) {
      if (errno != EINTR)
        abort();

      if (timeout == 0)
        return;

      if (timeout == -1)
        continue;

      /* Interrupted by a signal. Update timeout and poll again. */
      goto update_timeout;
    }

    。。。
    //判断事件循环的观察者队列是否为空
    assert(loop->watchers != NULL);
    loop->watchers[loop->nwatchers] = (void*) events;
    loop->watchers[loop->nwatchers + 1] = (void*) (uintptr_t) nfds;
    // 循环处理内核返回的事件，执行事件绑定的回调函数
    for (i = 0; i < nfds; i++) {
        。。。。
    }
    
}
复制代码

uv__io_poll阶段源码最长，逻辑最为复杂，能够作个归纳，以下：当js层代码注册的事件回调都没有返回的时候，事件循环会阻塞在poll阶段。看到这里，你可能会想了，会永远阻塞在此处吗？

一、首先呢，在poll阶段执行的时候，会传入一个timeout超时时间，该超时时间就是poll阶段的最大阻塞时间。
二、其次呢，在poll阶段，timeout时间未到的时候，若是有事件返回，就执行该事件注册的回调函数。timeout超时时间到了，则退出poll阶段，执行下一个阶段。

因此，咱们不用担忧事件循环会永远阻塞在poll阶段。

以上就是事件循环的两个核心阶段。限于篇幅，timers阶段的其余源码和setImmediate、process.nextTick的涉及到的源码就不罗列了，感兴趣的童鞋能够看下源码。

最后，总结出事件循环的原理以下，以上你能够不care，记住下面的总结就行了。

事件循环原理

node 的初始化
- 初始化 node 环境。
- 执行输入代码。
- 执行 process.nextTick 回调。
- 执行 microtasks。
进入 event-loop
- 进入 timers 阶段
  - 检查 timer 队列是否有到期的 timer 回调，若是有，将到期的 timer 回调按照 timerId 升序执行。
  - 检查是否有 process.nextTick 任务，若是有，所有执行。
  - 检查是否有microtask，若是有，所有执行。
  - 退出该阶段。
- 进入IO callbacks阶段。
  - 检查是否有 pending 的 I/O 回调。若是有，执行回调。若是没有，退出该阶段。
  - 检查是否有 process.nextTick 任务，若是有，所有执行。
  - 检查是否有microtask，若是有，所有执行。
  - 退出该阶段。
- 进入 idle，prepare 阶段：
  - 这两个阶段与咱们编程关系不大，暂且按下不表。
- 进入 poll 阶段
  - 首先检查是否存在还没有完成的回调，若是存在，那么分两种状况。
    - 第一种状况：
      - 若是有可用回调（可用回调包含到期的定时器还有一些IO事件等），执行全部可用回调。
      - 检查是否有 process.nextTick 回调，若是有，所有执行。
      - 检查是否有 microtaks，若是有，所有执行。
      - 退出该阶段。
    - 第二种状况：
      - 若是没有可用回调。
      - 检查是否有 immediate 回调，若是有，退出 poll 阶段。若是没有，阻塞在此阶段，等待新的事件通知。
  - 若是不存在还没有完成的回调，退出poll阶段。
- 进入 check 阶段。
  - 若是有immediate回调，则执行全部immediate回调。
  - 检查是否有 process.nextTick 回调，若是有，所有执行。
  - 检查是否有 microtaks，若是有，所有执行。
  - 退出 check 阶段
- 进入 closing 阶段。
  - 若是有immediate回调，则执行全部immediate回调。
  - 检查是否有 process.nextTick 回调，若是有，所有执行。
  - 检查是否有 microtaks，若是有，所有执行。
  - 退出 closing 阶段
- 检查是否有活跃的 handles（定时器、IO等事件句柄）。
  - 若是有，继续下一轮循环。
  - 若是没有，结束事件循环，退出程序。

细心的童鞋能够发现，在事件循环的每个子阶段退出以前都会按顺序执行以下过程：

检查是否有 process.nextTick 回调，若是有，所有执行。
检查是否有 microtaks，若是有，所有执行。
退出当前阶段。

记住这个规律哦。

那么，按照以上公式，代入网上各类有关 nodejs 事件循环的测试代码，相信你已经可以解释为何会输出那样的结果了。若是不能，那就私信我吧~~