剖析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等事件句柄）。
  - 若是有，繼續下一輪循環。
  - 若是沒有，結束事件循環，退出程序。