非同步程式碼之霧:Node.js 的事件迴圈與 EventEmitter

身為一個 Node.js 工程師,怎麼可以不夠了解「非同步程式碼」的行為?我希望能綜合自己的一點心得與經驗,寫一篇探討 Node.js Event Loop 與 Event Pattern 的文章,而且還不能只是泛泛之談,必須稍微有點深度,然後還期待大家能夠很容易地讀懂。

這篇文章是我為這個想法所作的努力,它花了我好幾個晚上,寫了將近 20 個小時左右 (天吶~~)。雖然極力想要用更短的篇幅把一切說明清楚,卻發現這實在沒辦法用短短的幾句話就講完。然而,即便寫得夠多了,但難免還是有疏漏之處,也要請大家有發現錯誤之處,踴躍提出糾正!讓這篇文章能夠呈現最正確的內容!


導讀:您知道現在已經不能再使用 process.nextTick() 來拆分你的 long-running task 了嗎?假使您對 Node.js 的非同步程式行為已經有很好的認識,您可直接跳至本文章的「警告!」之處,直接了解 Node.js 官方給開發者的提醒。

接下來的文章會有點長,但其實是因為貼上程式碼的關係。在您要閱讀之前,請您先靜下心,請您給我 20 分鐘的耐心與時間,和我一起撥雲見日。

先來一段小程式,猜猜看 console.log 的列印順序

為了刺激一下你,請先看看底下的程式碼,那些訊息將被列印的順序?測試一下自己對 Node.js 非同步行為的認知。


console.log('<0> schedule with setTimeout in 1-sec');

setTimeout(function () {

    console.log('[0] setTimeout in 1-sec boom!');

}, 1000);


console.log('<1> schedule with setTimeout in 0-sec');

setTimeout(function () {

    console.log('[1] setTimeout in 0-sec boom!');

}, 0);


console.log('<2> schedule with setImmediate');

setImmediate(function () {

    console.log('[2] setImmediate boom!');

});


console.log('<3> A immediately resolved promise');

aPromiseCall().then(function () {

    console.log('[3] promise resolve boom!');

});


console.log('<4> schedule with process.nextTick');

process.nextTick(function () {

    console.log('[4] process.nextTick boom!');

});


function aPromiseCall () {

    return new Promise(function(resolve, reject) {

        return resolve();

    });

}


執行結果:


<0> schedule with setTimeout in 1-sec
<1> schedule with setTimeout in 0-sec
<2> schedule with setImmediate
<3> A immediately resolved promise
<4> schedule with process.nextTick
[4] process.nextTick boom!
[3] promise resolve boom![1] setTimeout in 0-sec boom![2] setImmediate boom!
[0] setTimeout in 1-sec boom!


(注意:在你的電腦上,[1] [2] 發生的順序可能會與我的不同)

好的,如果您猜對了。讓我們再來看看,如果這些事情發生在 I/O Callback 中又是如何?同樣的程式碼,整包塞進 readFile() 這支非同步 I/O API 的 Callback 中:

var fs = require('fs');


fs.readFile('./file.txt', 'utf8', function (err, data) {

   if (!err) {

       console.log('[I/O Callback get called] ' + data  + '\n');


       console.log('<0> schedule with setTimeout in 1-sec');

       setTimeout(function () {

           console.log('[0] setTimeout in 1-sec boom!');

       }, 1000);


       console.log('<1> schedule with setTimeout in 0-sec');

       setTimeout(function () {

           console.log('[1] setTimeout in 0-sec boom!');

       }, 0);


       console.log('<2> schedule with setImmediate');

       setImmediate(function () {

           console.log('[2] setImmediate boom!');

       });


       console.log('<3> A immediately resolved promise');

       aPromiseCall().then(function () {

           console.log('[3] promise resolve boom!');

       });


       console.log('<4> schedule with process.nextTick');

       process.nextTick(function () {

           console.log('[4] process.nextTick boom!');

       });

   }

});


function aPromiseCall () {  // ... 略



執行結果
[I/O Callback get called] read file boom!
<0> schedule with setTimeout in 1-sec
<1> schedule with setTimeout in 0-sec
<2> schedule with setImmediate
<3> A immediately resolved promise
<4> schedule with process.nextTick
[4] process.nextTick boom!
[3] promise resolve boom!
[2] setImmediate boom!
[1] setTimeout in 0-sec boom!
[0] setTimeout 1-sec boom!
(注意:在你的電腦上,[2] [1] 發生的順序一定會跟我的相同)

接下來,請靜下心,讓我們好好地來了解 Node.js 的非同步機制。真的要靜下心來看哦!因為很希望讓你看過一次,就把它給搞懂!
熱身:JavaScript 的事件迴圈與非同步機制
對於 JavaScript 的事件迴圈與非同步行為,很多書或網路文章都做了很淺顯易懂的說明,加上從實作中累積的經驗,相信每個 JS 的開發者內心,都隱隱約約有概念。

如果您還不是那麼清楚,以下兩部 P. Roberts 的影片,您可以花一點時間先看一下。第一部長約 15 分鐘,講得非常好,他很清楚地說明了瀏覽器中 JS 的 Single Thread + Single Call Stack + Callback Queue 是怎麼樣搭配運行起來的,簡單易懂。第二部影片是他在 JSConfEU 的演講,內容跟第一部大同小異,但是多了一個展示用的 webapp,所以可以跳過不看。當然,JS 老手這兩部都可以直接跳過啦 XDDD....


在 Browser 上的情況很容易理解。相對於瀏覽器,作為執行在 Server-side 的 Node.js,事情會稍微複雜一點。Node.js 採用 Google V8 作為 JS 的解釋引擎,而在處理 I/O 方面則使用了自己設計的 libuv。libuv 幫你封裝了不同 OS 平台的 I/O 操作,往上提供一致的 asynchronous/non-blocking API 與事件迴圈建設。當我們在討論 Node.js 的事件迴圈時,將會與 libuv 有關。


Node.js 真的是單執行緒嗎?
對於 Node.js 的評論,最常聽見它「單一執行緒」的環境,但實際上它的底層是多執行緒的。Daniel Khan 在 "How to track down CPU issues in Node.js" 這篇文章中起了很好的頭,它直接將 node 執行一支 app.js 時,所跑起來的 process 都列出來給你看。我們直接看 Khan 先生怎麼說 (要我自己說,絕對不會比他說的好):
 The famous statement ‘Node.js runs in a single thread’ is only partly true. Actually only your ‘userland’ code runs in one thread. Starting a simple node application and looking at the processes reveals that Node.js in fact spins up a number of threads. This is because Node.js maintains a thread pool to delegate synchronous tasks to, while Google V8 creates its own threads for tasks like garbage collection. 




libuv 的 Event Loop 與 Loop Iteration

在 libuv 的核心程式碼中,我們會看到一支 uv_run() 的函式,他所接受的第一個參數是一個指向 uv_loop_t 結構體的指標。這裡,uv_loop_t 結構體即事件迴圈 (名詞),而每一次執行 uv_run() 則是進行一次事件迴圈的 iteration (執行 uv_run() 是動詞)。


uv_run() 這函式真的是寫得淺顯易懂,我們不需要太執著於細節,只需要知道這支函式一執行起來,將依序跑過 uv__update_time(), uv__run_timers(), uv__run_pendings(), ..., uv__run_closing_handles() 等函式,每支函式稱之為 event loop 的 phase (階段)。Event Loop 跑完一圈,總共會歷經這幾個階段。


libuv core.cc (原始碼 core.cc),底下的程式碼片段用眼睛稍微掃過即可:


int uv_run(uv_loop_t* loop, uv_run_mode mode) {

 // ... 略

 r = uv__loop_alive(loop);

 if (!r)

   uv__update_time(loop);


 while (r != 0 && loop->stop_flag == 0) {

   uv__update_time(loop);

   uv__run_timers(loop);

   ran_pending = uv__run_pending(loop);

   uv__run_idle(loop);

   uv__run_prepare(loop);


   timeout = 0;

   if ((mode == UV_RUN_ONCE && !ran_pending) || mode == UV_RUN_DEFAULT)

     timeout = uv_backend_timeout(loop);


   uv__io_poll(loop, timeout);

   uv__run_check(loop);

   uv__run_closing_handles(loop);


   if (mode == UV_RUN_ONCE) {

     // ... 略

     uv__update_time(loop);

     uv__run_timers(loop);

   }


   r = uv__loop_alive(loop);

   // ... 略

 return r;

}

Node.js 官方文件對事件迴圈的說明

Node.js 官方隨附在原始碼中有一份非常佛心的文件,很簡要地說明了 Event Loop 的運作方式,讓我們不需要苦讀原始碼,便能對 Event Loop 的行為略知一二。這份文件是今年 4 月(2016 年 4 月) 加上去的,熱騰騰呀!


這份文件給了一張圖,我把它重新畫了一次,並且跟上面 libuv core.cc 中看到的各個 phase 工作函式左右對照一下,這樣應該就夠簡單清楚了。我認為每個 Node.js 的開發者,都應該好好閱讀一下這份文件。那如果你真的很懶得看,我下面會作一些重點摘要。這裡先說明一下圖中右邊的「I/O Callbacks」,例如系統錯誤 (比如 socket 的錯誤, ECONNREFSED) 這一類的 callbacks 都會被 queue 在這裡,對應的是 uv__run_pending() 階段。如果是一般的 I/O 請求,callbacks 是在 poll 階段被執行。


Event Loop 特點摘要

  • 每個 phase 都有自己的 FIFO queue,裡面存放和自己相關的 callbacks
  • 進入一個 phase 後,該 phase 會將自己 queue 中的 callbacks 依序地同步執行,直到完全消化完畢時 (或達到最高數量限制) 再繼續往下個 phase 走
  • 「不要在 callback 中執行繁重的工作,否則事件迴圈將會被阻塞住」,原因在此
  • 當 Event Loop 繞完後,若檢查發現已無任何等待中的非同步 I/O 或 timers,事件迴圈即結束退出
  • 比如說你寫一支 app.js,裡面只有 console.log('Hello'),執行完一定馬上退出。如果你寫一個 http server.listen(3000, function () { ... }),執行起來之後,就一直執行著,因為底層開了一個 socket 一直在等待它的 I/O 事件,除非你把 socket 給 close 掉

各 Phase 的責任說明

  • timer:執行由 setTimeout() 及 setInterval() 排進來的 callbacks
  • I/O callbacks:有關系統錯誤等 Callbacks 將 queue 在此
  • idle, prepare:內部使用
  • poll:向系統取回新的 I/O 事件,執行對應的 I/O callbacks
  • check:執行由 setImmediate() 排進來的 callbacks
  • close callbacks:監聽 I/O 'close' 事件的 callbacks (如 socket.on('close', ...))

將工作 (或 callback) 排入事件迴圈中的方法

如何將工作排入事件迴圈的觀念非常非常重要,或許你覺得沒什麼,但這卻會關係到如何寫出行為正確地的非同步程式碼。


  • 使用了 timer 的 setTimeout(), setInterval()
  • callbacks 會被排進 timer phase 的 queue
  • 呼叫了使用 libuv non-blocking IO 的 API
  • 如 sockets, filesystem 相關 API,在 node 裡即如 fs.readFile() 這種非同步的 API
  • 使用 setImmediate()
  • callbacks 被排入 check phase 的 queue
  • 透過 process.nextTick()
  • 屬於 Node Event Loop 的一部分,但不屬於 libuv 的 phase。這後面我會說明。
  • 還有一個文件中沒有提到,就是使用了 Promise (microtask)
  • 屬於 node event loop 的一部分,但不屬於 libuv 的 phase。這後面我會說明。

一個 tick 到底是多長?

前面我們有看到 process.nextTick() 這個 API,您是否有想過,nextTick?那一個 tick 的時間到底是多長?Stackoverflow 上的這個回答很清楚,很簡短地說是這樣:

 一個 tick 的時間長度,是 Event Loop 繞完一圈,把所有 queues 中的 callbacks 依序且同步地執行完,所消耗的總時間。因此,一個 tick 的值是不固定的。可能很長,可能很短,但我們希望它能盡量地短。

所以再一次,所有 Node.js 開發者一再強調:「不要在 callback 中執行 long-running 的工作!」因為你會阻塞 Event Loop,當每一個 tick 的時間被你拉長,代表每單位時間 Event Loop 可以繞行而檢測出 I/O 事件的次數就會降低,非同步程式碼的效能因而折損。

執行順序

關於執行順序,請閱讀官方文件的 Phases in Detail 一節。我很快地總結一下幾個重點,同樣回到底下這張圖。雖然整個迴圈看起來是從 timers 開始執行起,在 libuv 看起來也是這樣子。這樣說好了,Event Loop 是一個閉迴路,它在第一次 kick off 時,確實是從 timers 那個 phase 跑起。


但是以長遠來看,一個閉迴路,你可以拿任意點當作起跑點。由於程式的目的大多與 I/O 有關,例如你開了一個 socket,所以 Event Loop 的重心可以視為圍繞在 poll phase 上,因為繞來繞去,你總是會在 poll phase 停留一下子,把該執行的執行完後,再東看看、西看看,看還有沒有其他事情要繼續做的。你在網路上會看到許多文章,或是這份官方文件,都是以 poll phase 作為討論的核心。並且注意官方文件的這句話:

 Technically, the poll phase controls when timers are executed.


官方的說明比較零碎一點,我依照我的理解整理一下,如果我們從 poll 開始看起,整個順序會像這樣:


  • poll phase:I/O 事件先處理,同時會關心即將逾期的 timer,都處理完後進 check phase 
  • check phase:處理 setImmedaite() 排進來的東西,如果沒有、或處理完了,就捲回 timers 看沒有要到期的
  • 然後繼續往下走回到 poll,先看有沒有 I/O 事件要處理同時關心即將逾期的 timer


一般原則


  • timer 快逾期,但 I/O 事件先發生,一律會先等 I/O 先處理完,再處理到期的 timers,也因此 timers 的 callbacks 不保證可以準時執行
  • 以官網的例子來講:例如有個 timer 在 100ms 後到期,但在即將到期之前來了一個 I/O 事件,則先處理 I/O,所以 timer 的 callback 可能會稍微延宕一下才被執行到,例如在第 105 ms 時才執行


最高原則


  • 所有以 process.nextTick() 所安排進來的 callbacks 都將在每一個 phase 結束,要轉換至下個 phase 之前,馬上被依序且同步地執行
  • 因此絕對不可在 process.nextTick 的 callback 中執行 long-running task
  • 不可以執行會遞迴呼叫 process.nextTick 的函式,因為那個 phase 永遠會檢測到還有 1 個 callback 要執行,因而造成 Event Loop 永遠被阻塞於該 phase(如果有人看了官方文件,認為我的理解有誤,請一定要讓我知道!)

setTimeout() 與 setImmediate()

  • setTimeout() 屬於 timers phase。被設計於逾時執行。
  • setImmediate() 屬於 check phase。被設計在每次 poll phase 之後執行。
  • setImmediate() 並不是以計時器來定時的,但 Node.js 仍將這個 API 歸類在 timers 核心模組
  • 這兩支方法,如果在 I/O cycle 被呼叫,setImmediate(cb) 者必定會先執行(因為下一個 phase 就是 check)。如果不是在 I/O cycle 被呼叫,setImmediate(cb) 與 zeo-second  setTimeout(cb, 0) 兩者被執行的次序為不可預測 (non-deterministic)
  • 請回到文章最開始的「猜猜看」,那裡的 [1] [2] 發生順序的問題在此處得到了說明

process.nextTick() 與 setImmediate()

  • process.nextTick 不屬於任何一個 phase (後面會提到)
  • 由 process.nextTick() 所排進 queue 的 callbacks 會在當下的那個 phase 結束前被拉出來,全部執行完。所以你若遞迴地呼叫 process.nextTick(),將造成 queue 永遠無法清空,該 phase 永遠無法轉換到下一個 phase,因而會造成 I/O starving(飢餓) 的問題 (無法再 poll)
  • 遞迴地呼叫 setImmediate() 所排進的下一個 callback,會被排到下一次 loop iteration 才執行,所以不會塞住 Event Loop
  • 神奇的 process.nextTick(),連大神 mafintosh 去年 7 月都在 twitter 上徵求:「Does anyone have a good code example of when to use setImmediate instead of nextTick?」 XDDDD....


警告!


你很可能在書上看到一些教你「使用 process.nextTick()」來拆分 long-running task 的做法!由於 Node.js 對 process.nextTick() 的行為已經調整過。請勿再使用書上介紹的方式,因為 Event Loop 仍然會被你的 long-running task 阻塞住!(拆開的 sub-tasks 仍是排在同一個 phase 中,同步地執行完,結果變成有拆跟沒拆一樣啊!哈哈~ 請改用 setImmediate() 去拆囉~)

從今天起,請勿被它的名字誤導,請不要再有「process.nextTick」可以將工作排到下一次 tick 的想法了!非常非常危險!官方文件這樣說:

 We recommend developers use setImmediate() in all cases because it's easier to reason about (and it leads to code that's compatible with a wider variety of environments, like browser JS.)

Node.js 的 Event Loop

Node 官方文件上有提到,它說 process.nextTick 並不算 libuv 的 event loop phases 的一部分。你可以這樣想,Node 的 Event Loop 是對底層 libuv 的一層包裹,在這一層包裹之內、libuv 之外,還有其他事情得處理,就是 process.nextTick 與 Promise 的 microtask。所以當我們談論 Node 的 Event Loop,指的是在 Node 層級的 Event Loop 整體,而不僅是單單 libuv 的 event loop 本身。

我在「從 node.js 原始碼看 exports 與 module.exports」這篇文章有提到 Node 核心是如何執行起來的,順序是這樣:

StartNodeInstance() -> CreateEnvironment() -> LoadEnvironment()

  • StartNodeInstance()


在 StartNodeInstance() 中 uv_run() 被呼叫了,而且是在一個 do ... while 迴圈之中。在 Node 層級上看到的 Event Loop 就在這裡:


{

     SealHandleScope seal(isolate);

     bool more;

     do {

       v8::platform::PumpMessageLoop(default_platform, isolate);

       more = uv_run(env->event_loop(), UV_RUN_ONCE);


       if (more == false) {

         v8::platform::PumpMessageLoop(default_platform, isolate);

         EmitBeforeExit(env);


         // Emit `beforeExit` if the loop became alive either after emitting

         // event, or after running some callbacks.

         more = uv_loop_alive(env->event_loop());

         if (uv_run(env->event_loop(), UV_RUN_NOWAIT) != 0)

           more = true;

       }

     } while (more == true);

   }


  • CreateEnvironment()


建立環境時,需要傳入一個指向 uv_loop_t 結構體的指標,這也告訴我們,每一個 node 的實例都將擁有自己的 Event Loop。建立過程的一部分程式碼即在初始化各個 phase。


Environment* CreateEnvironment(Isolate* isolate,

                              uv_loop_t* loop,

                              // ... 略

                              const char* const* exec_argv) {

 // ... 略


 uv_check_init(env->event_loop(), env->immediate_check_handle());

 uv_unref(reinterpret_cast<uv_handle_t*>(env->immediate_check_handle()));


 uv_idle_init(env->event_loop(), env->immediate_idle_handle());

 uv_prepare_init(env->event_loop(), env->idle_prepare_handle());

 uv_check_init(env->event_loop(), env->idle_check_handle());

 uv_unref(reinterpret_cast<uv_handle_t*>(env->idle_prepare_handle()));

 uv_unref(reinterpret_cast<uv_handle_t*>(env->idle_check_handle()));

 // ... 略


 return env;

}


  •  startup.processNextTick() (/src/node.js)


這裡我們只要關注一開始的 nextTickQueue,還有 process.nextTick(),這支方法僅僅是把註冊的 callbacks 安排進這個 queue 中。在 _tickCallback() 被抓出來執行時,就把 queue 中的每支 callbacks 撈出來執行,而這些處理完後,下一步則是 _runMicroTasks() 繼續處理 Promise 的事情。如果您想更進一步了解 microtasks,您可以看看這篇 Google 工程師 Jake Archibald 寫的「Tasks, microtasks, queues and schedules」,他在裡面也準備了小測驗讓你猜程式碼的執行順序 XDDD。

總地來說,我想表達的是:「process.nextTick() 與 microtasks 在非同步程式碼中的優先序是數一數二高的!每個 phase 結束之前都會被執行!(再次提醒,不是每個 tick!)」


startup.processNextTick = function() {

   var nextTickQueue = [];   // Callbacks 會排進這個 queue!!

   var pendingUnhandledRejections = [];

   var microtasksScheduled = false;

   var _runMicrotasks = {};

   // ... 略

   process.nextTick = nextTick;  // nextTick 函式在下面

   // ... 略

   // process._setupNextTick 在 node.cc 中, 我認為意思到了, 就不用再挖下去了

   const tickInfo = process._setupNextTick(_tickCallback, _runMicrotasks);

   _runMicrotasks = _runMicrotasks.runMicrotasks;

   // ... 略

   function _tickCallback() {

     var callback, args, tock;


     do {

       while (tickInfo[kIndex] < tickInfo[kLength]) {

       // callbacks 從 queue 中一個一個被挖出來執行

         tock = nextTickQueue[tickInfo[kIndex]++];

         callback = tock.callback;

         args = tock.args;


         if (args === undefined) {

           nextTickCallbackWith0Args(callback);

         } else {

           switch (args.length) {

             case 1:

               nextTickCallbackWith1Arg(callback, args[0]);

             // ...

           }

         }

         if (1e4 < tickInfo[kIndex])

           tickDone();

       }

       tickDone();

       // process.nextTick 的 callbacks 跑完, 接著跑 Promise 的 microtasks

       _runMicrotasks();

       emitPendingUnhandledRejections();

     } while (tickInfo[kLength] !== 0);

   }


   // ...略

   function nextTick(callback) {

     var args;

     if (arguments.length > 1) {

       args = [];

       for (var i = 1; i < arguments.length; i++)

         args.push(arguments[i]);

     }


     // 將 callback 連它的 arguments 用一個物件存起來推進 queue

     nextTickQueue.push(new TickObject(callback, args));

     tickInfo[kLength]++;

   }


   // ...

 };


原始碼看到這裡,大致上也拼湊出了一些圖像,因為原始碼實在很多,我想就留給有興趣的人繼續追下去吧!您可以看看 module.js 的 Module.runMain() 方法、node.cc 的 MakeCallback() 方法以及它所呼叫 env->tick_callback_function() 都是相關的。


這裡 nextTickQueue 的 nextTick 從字面上看也會造成誤會,以為是在下一個 loop iteration 執行,實際上這個 queue 中的 callbacks 會在 Event Loop 每次準備作 phase transition 之前執行。關於 nextTick 與 setImmediate 命名上的語義不清之處,Node 官方文件上也有提出說明:

 In essence, the names should be swapped. process.nextTick() fires more immediately than setImmediate() but this is an artifact of the past which is unlikely to change. Making this switch would break a large percentage of the packages on npm. 

看到這裡,假如您還沒睡著!我真的是佩服佩服!...先不要幹角我啊!我們終於要進入另一個主題 EventEmitter 了!相信我!這個題目會很快!


EventEmitter

Node.js 最著名的就是它的「非同步」以及「事件驅動」特性,看完我們上面對 Event Loop 的淺析,相信大家現在應該有點爽爽的感覺。在這邊,我們要再討論一個很重要的東西,就是 EventEmitter。這邊我先說一下我對它的總結:


 Node.js 給你 EventEmitter,是讓你在使用者空間創造 event pattern 的工具。它的本身與 Event Loop 毫無關係!

什麼????!!!!


當你看完這句話,或許你會帶點疑惑,又或者帶一點不服氣!很想質疑我到底懂不懂 Node.js!先別抓狂、先別暴怒、先不要摔電腦!讓我們繼續看下去~

EventEmitter 本身是「同步的」

關於 EventEmitter 的「同步」本質,在我讀過一些書或文章,很遺憾地,都沒有很明確地指出這一點。甚至有些說法比較囫圇吞棗一點、有些說的比較隱晦、又或者有書上以為 EventEmitter 是事件迴圈的抽象 (這完全不對呀,請不要問我哪本書了)!


當我還是 Node.js 新手時,我也曾經這樣相信了。直到我自己使用 Lua 實作一套符合 Node.js 介面的 EventEmittertimer 時,我才發現事情完全不是那麼樣子。也因為想著要符合 Node.js 的介面,也讓我「抄襲」了 Node.js 的作法 (呵呵~ 是向優秀偉大的開發者學習啦~)


以下一樣是以 node.js v4.5.0 LTS 原始碼為例。EventEmitter 的實作在 /lib/events.js,它總共不超過 450 行。事件模式有兩支很重要的方法,我們在實作上幾乎都是圍繞在 .emit(event, ...) 以及.on(listener) 這兩支方法。.on() 讓你註冊事件監聽器,而 .emit() 讓你發射事件。一旦事件發生,註冊監聽該事件的 callbacks 將被執行。我想這樣的模式,使用 JavaScript 的開發者應該都蠻熟悉的 (豈止熟悉?連想都不用想了....)。


EventEmitter 的 constructor 長這樣,它的構造真的很簡單,內部就是一個 protected member this._event = {},這個盒子裡,會以 event type (事件名稱) 當 key,而註冊進來的 listener 作為 value。如果一個事件有很多個 listeners,那麼 value 就會是一個按註冊順序來儲存這些 handlers 的陣列。


function EventEmitter() {

 EventEmitter.init.call(this);

}


EventEmitter.init = function() {

 // ... 略

 if (!this._events || this._events === Object.getPrototypeOf(this)._events) {

   this._events = {};      // 這個物件將用來管理註冊進來的 listeners

   this._eventsCount = 0;

 }


 this._maxListeners = this._maxListeners || undefined;

};


現在讓我們先來看 .on(),它是 addListener 的別名,所以我們直接看 addListener 方法:


EventEmitter.prototype.on = EventEmitter.prototype.addListener;


EventEmitter.prototype.addListener = function addListener(type, listener) {

 var events;

 var existing;

 // ... 略

 events = this._events;


 // ... 略

   existing = events[type];


 // 如果 event type 不存在, 就把以 type 當 key, 把 listener 當 value 塞進去

 if (!existing) {

   existing = events[type] = listener;

   ++this._eventsCount;

 } else {

   // 如果事件已存在, 它的值是函式, 現在要改用陣列來儲存

   // 如果已經是陣列, 帶有已經有 2 個以上的 listeners, 就繼續把新的 listener push 進去

   if (typeof existing === 'function') {

     // Adding the second element, need to change to array.

     existing = events[type] = [existing, listener];

   } else {

     // If we've already got an array, just append.

     existing.push(listener);

   }

   // ... 略

 }


 return this;

};


是不是很簡單啊!接著,我們來看 .emit():


EventEmitter.prototype.emit = function emit(type) {

 var er, handler, len, args, i, events, domain;

 // ... 略

 events = this._events;

 // ... 略


 handler = events[type];  // 找出 handler


 // 若沒有那個 type 的監聽器, 就直接 return

 if (!handler)

   return false;


 // ... 略

 // 接下的 cases, 只是 node 為了效能起見, 針對不同 args 數量寫了不同的呼叫方式

 // 我們就抓 emitOne 出來看吧

 switch (len) {

   // fast cases

   case 1:

     emitNone(handler, isFn, this);

     break;

   case 2:

     emitOne(handler, isFn, this, arguments[1]);

     // ... 略

 }


 // ... 略

 return true;

};


就拿 .emitOne() 當代表來說明:


function emitOne(handler, isFn, self, arg1) {

 // 如果 handler 是一支函式, 就直接執行

 if (isFn)

   handler.call(self, arg1);


 // 若非函式, 那就是一堆函式的陣列

 else {

   var len = handler.length;

   var listeners = arrayClone(handler, len);


   // 陣列中的 handlers 一支支依序被拉出來執行

   // 註: 這是同步的程式碼

   for (var i = 0; i < len; ++i)

     listeners[i].call(self, arg1);

 }

}


這告訴我們,每一次的 emit,都將跑起一段同步的程式碼,一個 callback 執行完再接著下一個,直到執行結束。是不是聽起來跟前面的 callback queue 感覺很像呀!是的,就是那一回事!可是,EventEmitter 本身的運作,壓根與 Node.js 的事件迴圈完全沒有關係!你可以把整份 event.js 讀一讀,你不會看到任何非同步的程式碼!


如果你曾經使用過 React flux 模式,它的 Dispatcher 也運用了相同的模式來完成 payload 的 broadcast (請見 register() 與 dispatch() 兩支方法是不是跟 on() 與 emit() 的感覺很像呢?只是裡面多了些狀態控制的東東啦!)。

使用 EventEmitter 要格外小心

我們在 node.js 中大部分會使用到 EventEmitter 的情況,除了用於協助工作流程控制外,最常見的場合就是用於通知某件事情的發生(或完成),而這大部分多用於通知某個非同步的工作完成了、發生了(讀檔完成、斷線了、socket 關閉了等等)。


例如:


fooEmitter.on('data', function (data) {

 console.log(data);

});


fs.readFile('/path/to/file', (err, data) => {

 if (!err)

   fooEmitter.emit('data', data);

});


因為使用情境常常都是像上面這樣,所以造成了「使用了 EventEmitter 就好像是寫了非同步程式碼的假象」。


看過狗追自己的尾巴嗎?來寫一個!


我們在一個 event1 handler 中 emit 了一個 'event2' 事件,而在 event2 handler 中又繼續 emit 一個 'event3' 事件,然後最後一個 event3  handler 發射 'event1' 事件:


var EventEmitter = require("events");


var crazy = new EventEmitter();


crazy.on('event1', function () {

   console.log('event1 fired!');

   crazy.emit('event2');

});


crazy.on('event2', function () {

   console.log('event2 fired!');

   crazy.emit('event3');


});


crazy.on('event3', function () {

   console.log('event3 fired!');

   crazy.emit('event1');

});


crazy.emit('event1');


執行看看!你將會得到 call stack 爆炸的例外 XDDD.... 狗狗因為過度暈眩就這樣死了。為什麼?因為所有 callback 的執行是同步的!一直遞迴地 call 下去,永遠不回頭!不要以為這種事不會發生,天底下就是會有那麼多巧合!

瘋狂旋轉的不死狗

那如果第一次 fire 使用 setImmediate() 推入事件迴圈呢(注意哦!很非同步 style 對不對)?你還是會得到一樣的結果!你只是把第一次的 fire 丟入事件迴圈,當事件一發生時,整個 EventEmitter 的觸發鏈是同步的,將把事件迴圈阻塞住,然後 callback 一直遞迴地呼叫下去,直到 stack 爆掉而當機。同樣的道理,如果我們沒有故意把事件兜成一個閉迴路,但是每一個 event handler 都是 long-running 的話,那麼同樣會使事件迴圈被阻塞的時間變長。


接下來,我們將剛剛的程式碼中的每個 emit() 都用 setImmediate() 丟入事件迴圈呢?你將得到一隻不死狗:


var EventEmitter = require('events');


var crazy = new EventEmitter();


crazy.on('event1', function () {

   console.log('event1 fired!');

   setImmediate(function () {

       crazy.emit('event2');

   });

});


crazy.on('event2', function () {

   console.log('event2 fired!');

   setImmediate(function () {

       crazy.emit('event3');

   });


});


crazy.on('event3', function () {

   console.log('event3 fired!');

   setImmediate(function () {

       crazy.emit('event1');

   });

});


crazy.emit('event1');


執行看看!這是真正的非同步程式碼!你會很開心!因為不再當機了!

那改用 process.nextTick 好了

現在,你看 process.nextTick 應該也夠眼熟了,如果我們把上面程式碼全部的 setImmediate() 換成 process.nextTick 呢?你猜結果會怎樣? (不要試!很恐怖!)


// ... 略

crazy.on('event1', function () {

   console.log('event1 fired!');

   // 將 全部的 setImmediate 換成 process.nextTick

   process.nextTick(function () {

       crazy.emit('event2');

   });

});


// ... 略

crazy.emit('event1');


它會卡住!你要等久一點.... 大概 30 秒左右,最後它會給你一個 process out of memory 的例外。現在不是 stack 爆掉,而是 GC 沒有辦法成功回收記憶體 (每個 handler 都有自己的 closure 去存取外層的那個 crazy,這個開銷會在 heap 上)。姑且不管最後那個 GC 為何無法成功回收的原因,但相信你應該也猜的到,我們的程式會一直鎖死在某個 phase,因為永遠有清除不完的下一個 process.nextTick 的 callback (所以事件迴圈完全被阻塞住了,啾咪~ Heap 爆掉可以說是意外的收穫阿... XDDD)。


所以,關於 EventEmitter,回到我前面說的:

 Node.js 給你 EventEmitter,是讓你在使用者空間創造 event pattern 的工具。它的本身與 Event Loop 毫無關係!

那麼我們應該要怎麼樣寫出「非同步的 event pattern」呢?現在你知道,你需要的只是將 EventEmitter 與那些可以將工作丟入 Event Loop 的 APIs 搭配使用即可!(setTimeout(), setInterval(), Async I/O APIs, setImmediate(), 以及 process.nextTick()。如果使用 process.nextTick() 要稍微小心一點,只要避免產生遞迴呼叫、避免在 callback 中執行 long-running task,一般是不會有什麼大問題。)

如果說到這裡,您還是沒辦法被說服,那麼請您試著執行以下程式碼,您認為程式會一直執行下去還是馬上結束呢:


var EventEmitter = require('events');

var server = new EventEmitter();


server.on('data', function () {

   console.log('Am I waiting for data incoming?');

});


如果您不用執行,就馬上能回答出來。您已經確確實實懂我的意思了!那裡根本沒有任何事情被安排進 Event Loop。

結語

這篇文章,我們把 Node.js 的「Event Loop」跟「EventEmitter」兩個概念完全切割開了,把它們各自梳理的很清楚,它們本來就不是天生就結合在一起的東西,EventEmitter 更不是 Event Loop 的抽象。一旦我們對這兩個概念不再模模糊糊,那麼把它們兩者結合起來運用,你一定會覺得更加得心應手!


很希望這篇文章,對於跟我一樣熱愛 JavaScript、熱愛 Node.js 的開發者,能夠對 Node.js 的非同步行為可以有很好的啟發與認識,然後能繼續寫出更棒的非同步程式碼。然後,我自己有個很不要臉的期待是,希望這篇文章可以成為大家探討 Node.js 非同步行為很好的範例,非常歡迎各界拿去修修改改當教材(因為我覺得正確認識它,真的非常非常重要)。同時也希望大家有發現錯誤的話,能告訴我,讓我們一起把它修改得更好、更正確!


還有還有,我很久沒在文章裡面請求大家支持我們的粉絲團啦!之前都覺得粉絲跟朋友一樣,不用多,死忠的有一個足矣。不過呢,如果您覺得這裡的文章真的寫得不錯,那麼就請您多多推薦給您的朋友。其實我是不知道這對 front-end 有沒有用,所以我只打算發布在 Node.js TW,不過還是很歡迎大家的轉載。


當然也別忘了給粉絲團按個讚,持續接受 E.E. 狂想曲的騷擾 XDDD。有大家的鼓勵,也會讓我更有動力繼續努力寫文章! (眼神死)






內文圖片來源:李健榮 Simen

封面圖片來源:pexels