理解 Node.js 事件驅動機制的原理

學習 Node.js 一定要理解的內容之一，文中主要涉及到了 EventEmitter 的使用和一些異步情況的處理，比較偏基礎，值得一讀。

大多數 Node.js 對象都依賴了 EventEmitter 模塊來監(jiān)聽和響應事件，比如我們常用的 HTTP requests, responses, 以及 streams。

const EventEmitter = require('events');

事件驅動機制的最簡單形式，是在 Node.js 中十分流行的回調函數，例如 fs.readFile。 在回調函數這種形式中，事件每被觸發(fā)一次，回調就會被觸發(fā)一次。

我們先來探索下這個最基本的方式。

你準備好了就叫我哈，Node！

很久很久以前，在 js 里還沒有原生支持 Promise，async/await 還只是一個遙遠的夢想，回調函數是處理異步問題的最原始的方式。

回調從本質上講是傳遞給其他函數的函數，在 JavaScript 中函數是第一類對象，這也讓回調的存在成為可能。

一定要搞清楚的是，回調在代碼中的并不表示異步調用。 回調既可以是同步調用的，也可以是異步調用的。

舉個例子，這里有一個宿主函數 fileSize，它接受一個回調函數 cb，并且可以通過條件判斷來同步或者異步地調用該回調函數：

function fileSize (fileName, cb) {
 if (typeof fileName !== 'string') {
 // Sync
 return cb(new TypeError('argument should be string')); 
 } 
 fs.stat(fileName, (err, stats) => {
 if (err) { 
 // Async
 return cb(err); 
 } 
 // Async
 cb(null, stats.size);
 });
}

這其實也是個反例，這樣寫經常會引起一些意外的錯誤，在設計宿主函數的時候，應當盡可能的使用同一種風格，要么始終都是同步的使用回調，要么始終都是異步的。

我們來研究下一個典型的異步 Node 函數的簡單示例，它用回調樣式編寫：

const readFileAsArray = function(file, cb) {
 fs.readFile(file, function(err, data) {
 if (err) {
 return cb(err);
 }
 const lines = data.toString().trim().split('\n');
 cb(null, lines);
 });
};

readFileAsArray 函數接受兩個參數：一個文件路徑和一個回調函數。它讀取文件內容，將其拆分成行數組，并將該數組作為回調函數的參數傳入，調用回調函數。

現在設計一個用例，假設我們在同一目錄中的文件 numbers.txt 包含如下內容：

10
11
12
13
14
15

如果我們有一個需求，要求統(tǒng)計該文件中的奇數數量，我們可以使用 readFileAsArray 來簡化代碼：

readFileAsArray('./numbers.txt', (err, lines) => {
 if (err) throw err;
 const numbers = lines.map(Number);
 const oddNumbers = numbers.filter(n => n%2 === 1);
 console.log('Odd numbers count:', oddNumbers.length);
});

這段代碼將文件內容讀入字符串數組中，回調函數將其解析為數字，并計算奇數的個數。

這才是最純粹的 Node 回調風格?；卣{的第一個參數要遵循錯誤優(yōu)先的原則，err 可以為空，我們要將回調作為宿主函數的最后一個參數傳遞。你應該一直用這種方式這樣設計你的函數，因為用戶可能會假設。讓宿主函數把回調當做其最后一個參數，并讓回調函數以一個可能為空的錯誤對象作為其第一個參數。

回調在現代 JavaScript 中的替代品

在現代 JavaScript 中，我們有 Promise，Promise 可以用來替代異步 API 的回調。回調函數需要作為宿主函數的一個參數進行傳遞（多個宿主回調進行嵌套就形成了回調地獄），而且錯誤和成功都只能在其中進行處理。而 Promise 對象可以讓我們分開處理成功和錯誤，還允許我們鏈式調用多個異步事件。

如果 readFileAsArray 函數支持 Promise，我們可以這樣使用它，如下所示：

readFileAsArray('./numbers.txt')
 .then(lines => {
 const numbers = lines.map(Number);
 const oddNumbers = numbers.filter(n => n%2 === 1);
 console.log('Odd numbers count:', oddNumbers.length);
 })
 .catch(console.error);

我們在宿主函數的返回值上調用了一個函數來處理我們的需求，這個 .then 函數會把剛剛在回調版本中的那個行數組傳遞給這里的匿名函數。為了處理錯誤，我們在結果上添加一個 .catch 調用，當發(fā)生錯誤時，它會捕捉到錯誤并讓我們訪問到這個錯誤。

在現代 JavaScript 中已經支持了 Promise 對象，因此我們可以很容易的將其使用在宿主函數之中。下面是支持 Promise 版本的 readFileAsArray 函數（同時支持舊有的回調函數方式）：

const readFileAsArray = function(file, cb = () => {}) {
 return new Promise((resolve, reject) => {
 fs.readFile(file, function(err, data) {
 if (err) {
 reject(err);
 return cb(err);
 } 
 const lines = data.toString().trim().split('\n');
 resolve(lines);
 cb(null, lines);
 });
 });
};

我們使該函數返回一個 Promise 對象，該對象包裹了 fs.readFile 的異步調用。Promise 對象暴露了兩個參數，一個 resolve 函數和一個 reject 函數。

當有異常拋出時，我們可以通過向回調函數傳遞 error 來處理錯誤，也同樣可以使用 Promise 的 reject 函數。每當我們將數據交給回調函數處理時，我們同樣也可以用 Promise 的 resolve 函數。

在這種同時可以使用回調和 Promise 的情況下，我們需要做的唯一一件事情就是為這個回調參數設置默認值，防止在沒有傳遞回調函數參數時，其被執(zhí)行然后報錯的情況。 在這個例子中使用了一個簡單的默認空函數：（）=> {}。

通過 async/await 使用 Promise

當需要連續(xù)調用異步函數時，使用 Promise 會讓你的代碼更容易編寫。不斷的使用回調會讓事情變得越來越復雜，最終陷入回調地獄。

Promise 的出現改善了一點，Generator 的出現又改善了一點。 處理異步問題的最新解決方式是使用 async 函數，它允許我們將異步代碼視為同步代碼，使其整體上更加可讀。

以下是使用 async/await 版本的調用 readFileAsArray 的例子：

async function countOdd () {
 try {
 const lines = await readFileAsArray('./numbers');
 const numbers = lines.map(Number);
 const oddCount = numbers.filter(n => n%2 === 1).length;
 console.log('Odd numbers count:', oddCount);
 } catch(err) {
 console.error(err);
 }
}
countOdd();

首先，我們創(chuàng)建了一個 async 函數 —— 就是一個普通的函數聲明之前，加了個 async 關鍵字。在 async 函數內部，我們調用了 readFileAsArray 函數，就像把它的返回值賦值給變量 lines 一樣，為了真的拿到 readFileAsArray 處理生成的行數組，我們使用關鍵字 await。之后，我們繼續(xù)執(zhí)行代碼，就好像 readFileAsArray 的調用是同步的一樣。

要讓代碼運行，我們可以直接調用 async 函數。這讓我們的代碼變得更加簡單和易讀。為了處理異常，我們需要將異步調用包裝在一個 try/catch 語句中。

有了 async/await 這個特性，我們不必使用任何特殊的API（如 .then 和 .catch ）。我們只是把這種函數標記出來，然后使用純粹的 JavaScript 寫代碼。

我們可以把 async/await 這個特性用在支持使用 Promise 處理后續(xù)邏輯的函數上。但是，它無法用在只支持回調的異步函數上（例如setTimeout）。

EventEmitter 模塊

EventEmitter 是一個處理 Node 中各個對象之間通信的模塊。 EventEmitter 是 Node 異步事件驅動架構的核心。 Node 的許多內置模塊都繼承自 EventEmitter。

它的概念其實很簡單：emitter 對象會發(fā)出被定義過的事件，導致之前注冊的所有監(jiān)聽該事件的函數被調用。所以，emitter 對象基本上有兩個主要特征：

為了使用 EventEmitter，我們需要創(chuàng)建一個繼承自 EventEmitter 的類。

class MyEmitter extends EventEmitter {
}

我們從 EventEmitter 的子類實例化的對象，就是 emitter 對象：

const myEmitter = new MyEmitter();

在這些 emitter 對象的生命周期里，我們可以調用 emit 函數來觸發(fā)我們想要的觸發(fā)的任何被命名過的事件。

myEmitter.emit('something-happened');
emit 函數的使用表示發(fā)生某種情況發(fā)生了，讓大家去做該做的事情。 這種情況通常是某些狀態(tài)變化引起的。

我們可以使用 on 方法添加監(jiān)聽器函數，并且每次 emitter 對象觸發(fā)其關聯的事件時，將執(zhí)行這些監(jiān)聽器函數。

事件 !== 異步

先看看這個例子：

const EventEmitter = require('events');

class WithLog extends EventEmitter {
 execute(taskFunc) {
 console.log('Before executing');
 this.emit('begin');
 taskFunc();
 this.emit('end');
 console.log('After executing');
 }
}

const withLog = new WithLog();

withLog.on('begin', () => console.log('About to execute'));
withLog.on('end', () => console.log('Done with execute'));

withLog.execute(() => console.log('*** Executing task ***'));

WithLog 是一個事件觸發(fā)器，它有一個方法 —— execute，該方法接受一個參數，即具體要處理的任務函數，并在其前后包裹 log 以輸出其執(zhí)行日志。

為了看到這里會以什么順序執(zhí)行，我們在兩個命名的事件上都注冊了監(jiān)聽器，最后執(zhí)行一個簡單的任務來觸發(fā)事件。

下面是上面程序的輸出結果：

Before executing
About to execute
*** Executing task ***
Done with execute
After executing

這里我想證實的是以上的輸出都是同步發(fā)生的，這段代碼里沒有什么異步的成分。

就像普通的回調一樣，不要以為事件意味著同步或異步代碼。

跟之前的回調一樣，不要一提到事件就認為它是異步的或者同步的，還要具體分析。

如果我們傳遞 taskFunc 是一個異步函數，會發(fā)生什么呢？

// ...

withLog.execute(() => {
 setImmediate(() => {
 console.log('*** Executing task ***')
 });
});

輸出結果變成了這樣：

Before executing
About to execute
Done with execute
After executing
*** Executing task ***

這樣就有問題了，異步函數的調用導致 "Done with execute" 和 "After executing" 的輸出并不準確。

要在異步函數完成后發(fā)出事件，我們需要將回調（或 Promise）與基于事件的通信相結合。 下面的例子說明了這一點。

使用事件而不是常規(guī)回調的一個好處是，我們可以通過定義多個監(jiān)聽器對相同的信號做出多個不同的反應。如果使用回調來完成這件事，我們要在單個回調中寫更多的處理邏輯。事件是應用程序允許多個外部插件在應用程序核心之上構建功能的好辦法。你可以把它們當成鉤子來掛一些由于狀態(tài)變化而引發(fā)執(zhí)行的程序。

異步事件

我們把剛剛那些同步代碼的示例改成異步的：

const fs = require('fs');
const EventEmitter = require('events');

class WithTime extends EventEmitter {
 execute(asyncFunc, ...args) {
 this.emit('begin');
 console.time('execute');
 asyncFunc(...args, (err, data) => {
 if (err) {
 return this.emit('error', err);
 }

 this.emit('data', data);
 console.timeEnd('execute');
 this.emit('end');
 });
 }
}

const withTime = new WithTime();

withTime.on('begin', () => console.log('About to execute'));
withTime.on('end', () => console.log('Done with execute'));

withTime.execute(fs.readFile, __filename);

用 WithTime 類執(zhí)行 asyncFunc 函數，并通過調用 console.time 和 console.timeEnd 報告該asyncFunc 所花費的時間。它在執(zhí)行之前和之后都將以正確的順序觸發(fā)相應的事件，并且還會發(fā)出 error/data 事件作為處理異步調用的信號。

我們傳遞一個異步的 fs.readFile 函數來測試一下 withTime emitter。 我們現在可以直接通過監(jiān)聽 data 事件來處理讀取到的文件數據，而不用把這套處理邏輯寫到 fs.readFile 的回調函數中。

執(zhí)行這段代碼，我們以預期的順序執(zhí)行了一系列事件，并且得到異步函數的執(zhí)行時間，這些是十分重要的。

About to execute
execute: 4.507ms
Done with execute

請注意，我們是將回調與事件觸發(fā)器 emitter 相結合實現的這部分功能。 如果 asynFunc 支持Promise，我們可以使用 async/await 函數來做同樣的事情：

class WithTime extends EventEmitter {
 async execute(asyncFunc, ...args) {
 this.emit('begin');
 try {
 console.time('execute');
 const data = await asyncFunc(...args);
 this.emit('data', data);
 console.timeEnd('execute');
 this.emit('end');
 } catch(err) {
 this.emit('error', err);
 }
 }
}

我認為這段代碼比之前的回調風格的代碼以及使用 .then/.catch 風格的代碼更具可讀性。async/await 讓我們更加接近 JavaScript 語言本身（不必再使用 .then/.catch 這些 api）。

事件參數和錯誤

在之前的例子中，有兩個事件被發(fā)出時還攜帶了別的參數。

error 事件被觸發(fā)時會攜帶一個 error 對象。

this.emit('error', err);

data 事件被觸發(fā)時會攜帶一個 data 對象。

this.emit('data', data);

我們可以在 emit 函數中不斷的添加參數，當然第一個參數一定是事件的名稱，除去第一個參數之外的所有參數都可以在該事件注冊的監(jiān)聽器中使用。

例如，要處理 data 事件，我們注冊的監(jiān)聽器函數將訪問傳遞給 emit 函數的 data 參數，而這個 data 也正是由 asyncFunc 返回的數據。

withTime.on('data', (data) => {
 // do something with data
});

error 事件比較特殊。在我們基于回調的那個示例中，如果不使用監(jiān)聽器處理 error 事件，node 進程將會退出。

舉個由于錯誤使用參數而造成程序崩潰的例子：

class WithTime extends EventEmitter {
 execute(asyncFunc, ...args) {
 console.time('execute');
 asyncFunc(...args, (err, data) => {
 if (err) {
 return this.emit('error', err); // Not Handled
 }

 console.timeEnd('execute');
 });
 }
}

const withTime = new WithTime();

withTime.execute(fs.readFile, ''); // BAD CALL
withTime.execute(fs.readFile, __filename);

第一次調用 execute 將會觸發(fā) error 事件，由于沒有處理 error ，Node 程序隨之崩潰：

events.js:163
 throw er; // Unhandled 'error' event
 ^
Error: ENOENT: no such file or directory, open ''

第二次執(zhí)行調用將受到此崩潰的影響，并且可能根本不會被執(zhí)行。

如果我們?yōu)檫@個 error 事件注冊一個監(jiān)聽器函數來處理 error，結果將大不相同：

withTime.on('error', (err) => {
 // do something with err, for example log it somewhere
 console.log(err)
});

如果我們執(zhí)行上述操作，將會報告第一次執(zhí)行 execute 時發(fā)送的錯誤，但是這次 node 進程不會崩潰退出，其他程序的調用也都能正常完成：

{ Error: ENOENT: no such file or directory, open '' errno: -2, code: 'ENOENT', syscall: 'open', path: '' }
execute: 4.276ms

需要注意的是，基于 Promise 的函數有些不同，它們暫時只是輸出一個警告:

UnhandledPromiseRejectionWarning: Unhandled promise rejection (rejection id: 1): Error: ENOENT: no such file or directory, open ''

DeprecationWarning: Unhandled promise rejections are deprecated. In the future, promise rejections that are not handled will terminate the Node.js process with a non-zero exit code.

另一種處理異常的方式是在監(jiān)聽全局的 uncaughtException 進程事件。 然而，使用該事件全局捕獲錯誤并不是一個好辦法。

關于 uncaughtException，一般都會建議你避免使用它，但是如果必須用它，你應該讓進程退出：

process.on('uncaughtException', (err) => {
 // something went unhandled.
 // Do any cleanup and exit anyway!

 console.error(err); // don't do just that.

 // FORCE exit the process too.
 process.exit(1);
});

但是，假設在同一時間發(fā)生多個錯誤事件，這意味著上面的 uncaughtException 監(jiān)聽器將被多次觸發(fā)，這可能會引起一些問題。

EventEmitter 模塊暴露了 once 方法，這個方法發(fā)出的信號只會調用一次監(jiān)聽器。所以，這個方法常與 uncaughtException 一起使用。

監(jiān)聽器的順序

如果針對一個事件注冊多個監(jiān)聽器函數，當事件被觸發(fā)時，這些監(jiān)聽器函數將按其注冊的順序被觸發(fā)。

// first
withTime.on('data', (data) => {
 console.log(`Length: ${data.length}`);
});

// second
withTime.on('data', (data) => {
 console.log(`Characters: ${data.toString().length}`);
});

withTime.execute(fs.readFile, __filename);

上述代碼會先輸出 Length 信息，再輸出 Characters 信息，執(zhí)行的順序與注冊的順序保持一致。

如果你想定義一個新的監(jiān)聽函數，但是希望它能夠第一個被執(zhí)行，你還可以使用 prependListener 方法：

withTime.on('data', (data) => {
 console.log(`Length: ${data.length}`);
});

withTime.prependListener('data', (data) => {
 console.log(`Characters: ${data.toString().length}`);
});

withTime.execute(fs.readFile, __filename);

上述代碼中，Charaters 信息將首先被輸出。

最后，你可以用 removeListener 函數來刪除某個監(jiān)聽器函數。

聲明：本網頁內容旨在傳播知識，若有侵權等問題請及時與本網聯系，我們將在第一時間刪除處理。TEL:177 7030 7066 E-MAIL:11247931@qq.com