关于node.js:彻底搞懂nodejs事件循环

3次阅读

共计 6243 个字符，预计需要花费 16 分钟才能阅读完成。

nodejs 是单线程执行的，同时它又是基于事件驱动的非阻塞 IO 编程模型。这就使得咱们不必期待异步操作后果返回，就能够持续往下执行代码。当异步事件触发之后，就会告诉主线程，主线程执行相应事件的回调。

以上是家喻户晓的内容。明天咱们从源码动手，剖析一下 nodejs 的事件循环机制。

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

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

用户代码 (js 代码)

用户代码即咱们编写的利用程序代码、npm 包、nodejs 内置的 js 模块等，咱们日常工作中的大部分工夫都是编写这个层面的代码。

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

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

底层库

nodejs 的依赖库，包含赫赫有名的 V8、libuv。
V8：咱们都晓得，是 google 开发的一套高效 javascript 运行时，nodejs 可能高效执行 js 代码的很大起因次要在它。
libuv：是用 C 语言实现的一套异步性能库，nodejs 高效的异步编程模型很大水平上归功于 libuv 的实现，而 libuv 则是咱们明天重点要剖析的。
还有一些其余的依赖库
http-parser：负责解析 http 响应
openssl：加解密
c-ares：dns 解析
npm：nodejs 包管理器
…

对于 nodejs 不再过多介绍，大家能够自行查阅学习，接下来咱们重点要剖析的就是 libuv。

咱们晓得，nodejs 实现异步机制的外围便是 libuv，libuv 承当着 nodejs 与文件、网络等异步工作的沟通桥梁，上面这张图让咱们对 libuv 有个大略的印象：

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

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

对于网络 I /O，各个平台的实现机制不一样，linux 是 epoll 模型，类 unix 是 kquene、windows 下是高效的 IOCP 实现端口、SunOs 是 event ports，libuv 对这几种网络 I / O 模型进行了封装。参考 nodejs 进阶视频解说：进入学习

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

1、线程池中的线程都被占用的时候，队列中工作就要进行排队期待闲暇线程。
2、线程池中有可用线程时，从队列中取出这个工作执行，执行结束后，线程偿还到线程池，期待下个工作。同时以事件的形式告诉 event-loop，event-loop 接管到事件执行该事件注册的回调函数。

当然，如果感觉 4 个线程不够用，能够在 nodejs 启动时，设置环境变量 UV_THREADPOOL_SIZE 来调整，出于零碎性能思考，libuv 规定可设置线程数不能超过 128 个。

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

1、调用 platformInit 办法，初始化 nodejs 的运行环境。
2、调用 performance_node_start 办法，对 nodejs 进行性能统计。
3、openssl设置的判断。
4、调用v8_platform.Initialize，初始化 libuv 线程池。
5、调用 V8::Initialize，初始化 V8 环境。
6、创立一个 nodejs 运行实例。
7、启动上一步创立好的实例。
8、开始执行 js 文件，同步代码执行结束后，进入事件循环。
9、在没有任何可监听的事件时，销毁 nodejs 实例，程序执行结束。

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

咱们看几处要害源码：

1、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;
}

2、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);
  }
}

3、轮询阶段 源码，源码文件：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 阶段。看到这里，你可能会想了，会永远阻塞在此处吗？

1、首先呢，在 poll 阶段执行的时候，会传入一个 timeout 超时工夫，该超时工夫就是 poll 阶段的最大阻塞工夫。
2、其次呢，在 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 等事件句柄）。
  - 如果有，持续下一轮循环。
  - 如果没有，完结事件循环，退出程序。