標籤：

Node.js 非同步異聞錄

03-06

提到 Node.js, 我們腦海就會浮現非同步、非阻塞、單線程等關鍵詞，進一步我們還會想到 buffer、模塊機制、事件循環、進程、V8、libuv 等知識點。本文起初旨在理順 Node.js 以上易混淆概念，然而一入非同步深似海，本文嘗試基於 Node.js 的非同步展開討論，其他的主題只能日後慢慢補上了。(附：亦可以把本文當作是朴靈老師所著的《深入淺出 Node.js》一書的小結)。

非同步 I/O

Node.js 正是依靠構建了一套完善的高性能非同步 I/O 框架，從而打破了 JavaScript 在伺服器端止步不前的局面。

非同步 I/O VS 非阻塞 I/O

聽起來非同步和非阻塞，同步和阻塞是相互對應的，從實際效果而言，非同步和非阻塞都達到了我們並行 I/O 的目的，但是從計算機內核 I/O 而言，非同步/同步和阻塞/非阻塞實際上是兩回事。

注意，操作系統內核對於 I/O 只有兩種方式：阻塞與非阻塞。

調用阻塞 I/O 的過程：

調用非阻塞 I/O 的過程：

在此先引人一個叫作輪詢的技術。輪詢不同於回調，舉個生活例子，你有事去隔壁寢室找同學，發現人不在，你怎麼辦呢？方法1，每隔幾分鐘再去趟隔壁寢室，看人在不；方法2，拜託與他同寢室的人，看到他回來時叫一下你；那麼前者是輪詢，後者是回調。

再回到主題，阻塞 I/O 造成 CPU 等待浪費，非阻塞 I/O 帶來的麻煩卻是需要輪詢去確認是否完全完成數據獲取。從操作系統的這個層面上看，對於應用程序而言，不管是阻塞 I/O 亦或是 非阻塞 I/O，它們都只能是一種同步，因為儘管使用了輪詢技術，應用程序仍然需要等待 I/O 完全返回。

Node 的非同步 I/O

完成整個非同步 I/O 環節的有事件循環、觀察者、請求對象以及 I/O 線程池。

事件循環

在進程啟動的時候，Node 會創建一個類似於 whlie(true) 的循環，每一次執行循環體的過程我們稱為 Tick。

每個 Tick 的過程就是查看是否有事件待處理，如果有，就取出事件及其相關的回調函數。如果存在相關的回調函數，就執行他們。然後進入下一個循環，如果不再有事件處理，就退出進程。

偽代碼如下：

while(ture) { const event = eventQueue.pop() if (event && event.handler) { event.handler.execute() // execute the callback in Javascript thread } else { sleep() // sleep some time to release the CPU do other stuff }}

觀察者

每個 Tick 的過程中，如何判斷是否有事件需要處理，這裡就需要引入觀察者這個概念。

每個事件循環中有一個或多個觀察者，而判斷是否有事件需要處理的過程就是向這些觀察者詢問是否有要處理的事件。

在 Node 中，事件主要來源於網路請求、文件 I/O 等，這些事件都有對應的觀察者。

請求對象

對於 Node 中的非同步 I/O 而言，回調函數不由開發者來調用，在 JavaScript 發起調用到內核執行完 id 操作的過渡過程中，存在一種中間產物，它叫作請求對象。

請求對象是非同步 I/O 過程中的重要中間產物，所有狀態都保存在這個對象中，包括送入線程池等待執行以及 I/O 操作完後的回調處理

以 fs.open() 為例：

fs.open = function(path, flags, mode, callback) { bingding.open( pathModule._makeLong(path), stringToFlags(flags), mode, callback )}

fs.open 的作用就是根據指定路徑和參數去打開一個文件，從而得到一個文件描述符。

從前面的代碼中可以看到，JavaScript 層面的代碼通過調用 C++ 核心模塊進行下層的操作。

從 JavaScript 調用 Node 的核心模塊，核心模塊調用 C++ 內建模塊，內建模塊通過 libuv 進行系統調用，這是 Node 里經典的調用方式。

libuv 作為封裝層，有兩個平台的實現，實質上是調用了 uv_fs_open 方法，在 uv_fs_open 的調用過程中，會創建一個 FSReqWrap 請求對象，從 JavaScript 層傳入的參數和當前方法都被封裝在這個請求對象中。回調函數則被設置在這個對象的 oncomplete_sym 屬性上。

req_wrap -> object_ -> Set(oncomplete_sym, callback)

對象包裝完畢後，在 Windows 下，則調用 QueueUserWorkItem() 方法將這個 FSReqWrap 對象推人線程池中等待執行。

至此，JavaScript 調用立即返回，由 JavaScript 層面發起的非同步調用的第一階段就此結束(即上圖所注釋的非同步 I/O 第一部分)。JavaScript 線程可以繼續執行當前任務的後續操作，當前的 I/O 操作在線程池中等待執行，不管它是否阻塞 I/O，都不會影響到 JavaScript 線程的後續操作，如此達到了非同步的目的。

執行回調

組裝好請求對象、送入 I/O 線程池等待執行，實際上是完成了非同步 I/O 的第一部分，回調通知是第二部分。

線程池中的 I/O 操作調用完畢之後，會將獲取的結果儲存在 req -> result 屬性上，然後調用 PostQueuedCompletionStatus() 通知 IOCP，告知當前對象操作已經完成，並將線程歸還線程池。

在這個過程中，我們動用了事件循環的 I/O 觀察者，在每次 Tick 的執行過程中，它會調用 IOCP 相關的 GetQueuedCompletionStatus 方法檢查線程池中是否有執行完的請求，如果存在，會將請求對象加入到 I/O 觀察者的隊列中，然後將其當做事件處理。

I/O 觀察者回調函數的行為就是取出請求對象的 result 屬性作為參數，取出 oncomplete_sym 屬性作為方法，然後調用執行，以此達到調用 JavaScript 中傳入的回調函數的目的。

小結

通過介紹完整個非同步 I/O 後，有個需要重視的觀點是 JavaScript 是單線程的，Node 本身其實是多線程的，只是 I/O 線程使用的 CPU 比較少；還有個重要的觀點是，除了用戶的代碼無法並行執行外，所有的 I/O (磁碟 I/O 和網路 I/O) 則是可以並行起來的。

非同步編程

Node 是首個將非同步大規模帶到應用層面的平台。通過上文所述我們了解了 Node 如何通過事件循環實現非同步 I/O，有非同步 I/O 必然存在非同步編程。非同步編程的路經歷了太多坎坷，從回調函數、發布訂閱模式、Promise 對象，到 generator、asycn/await。趁著非同步編程這個主題剛好把它們串起來理理。

非同步 VS 回調

對於剛接觸非同步的新人，很大幾率會混淆回調 (callback) 和非同步 (asynchronous) 的概念。先來看看維基的 Callback 條目:

In computer programming, a callback is any executable code that is passed as an argument to other code

因此，回調本質上是一種設計模式，並且 jQuery (包括其他框架)的設計原則遵循了這個模式。

在 JavaScript 中，回調函數具體的定義為：函數 A 作為參數(函數引用)傳遞到另一個函數 B 中，並且這個函數 B 執行函數 A。我們就說函數 A 叫做回調函數。如果沒有名稱(函數表達式)，就叫做匿名回調函數。

因此 callback 不一定用於非同步，一般同步(阻塞)的場景下也經常用到回調，比如要求執行某些操作後執行回調函數。講了這麼多讓我們來看下同步回調和非同步回調的例子：

同步回調：

function f2() { console.log(f2 finished)}function f1(cb) { cb() console.log(f1 finished)}f1(f2) // 得到的結果是 f2 finished, f1 finished

非同步回調：

function f2() { console.log(f2 finished)}function f1(cb) { setTimeout(cb, 1000) // 通過 setTimeout() 來模擬耗時操作 console.log(f1 finished)}f1(f2) // 得到的結果是 f1 finished, f2 finished

小結：回調可以進行同步也可以非同步調用，但是 Node.js 提供的 API 大多都是非同步回調的，比如 buffer、http、cluster 等模塊。

發布/訂閱模式

事件發布/訂閱模式 (PubSub) 自身並無同步和非同步調用的問題，但在 Node 的 events 模塊的調用中多半伴隨事件循環而非同步觸發的，所以我們說事件發布/訂閱廣泛應用於非同步編程。它的應用非常廣泛，可以在非同步編程中幫助我們完成更松的解耦，甚至在 MVC、MVVC 的架構中以及設計模式中也少不了發布-訂閱模式的參與。

以 jQuery 事件監聽為例

$(#btn).on(myEvent, function(e) { // 觸發事件 console.log(I am an Event)})$(#btn).trigger(myEvent) // 訂閱事件

可以看到，訂閱事件就是一個高階函數的應用。事件發布/訂閱模式可以實現一個事件與多個回調函數的關聯，這些回調函數又稱為事件偵聽器。下面我們來看看發布/訂閱模式的簡易實現。

var PubSub = function() { this.handlers = {}}PubSub.prototype.subscribe = function(eventType, handler) { // 註冊函數邏輯 if (!(eventType in this.handlers)) { this.handlers[eventType] = [] } this.handlers[eventType].push(handler) // 添加事件監聽器 return this // 返回上下文環境以實現鏈式調用}PubSub.prototype.publish = function(eventType) { // 發布函數邏輯 var _args = Array.prototype.slice.call(arguments, 1) for (var i = 0, _handlers = this.handlers[eventType]; i < _handlers.length; i++) { // 遍歷事件監聽器 _handlers[i].apply(this, _args) // 調用事件監聽器 }}var event = new PubSub // 構造 PubSub 實例event.subscribe(name, function(msg) { console.log(my name is + msg) // my name is muyy})event.publish(name, muyy)

至此，一個簡易的訂閱發布模式就實現了。然而發布/訂閱模式也存在一些缺點，創建訂閱本身會消耗一定的時間與內存，也許當你訂閱一個消息之後，之後可能就不會發生。發布-訂閱模式雖然它弱化了對象與對象之間的關係，但是如果過度使用，對象與對象的必要聯繫就會被深埋，會導致程序難以跟蹤與維護。

Promise/Deferred 模式

想像一下，如果某個操作需要經過多個非阻塞的 IO 操作，每一個結果都是通過回調，程序有可能會看上去像這個樣子。這樣的代碼很難維護。這樣的情況更多的會發生在 server side 的情況下。代碼片段如下：

operation1(function(err, result1) { operation2(result1, function(err, result2) { operation3(result2, function(err, result3) { operation4(result3, function(err, result4) { callback(result4) // do something useful }) }) })})

這時候，Promise 出現了，其出現的目的就是為了解決所謂的回調地獄的問題。讓我們看下使用 Promise 後的代碼片段：

promise() .then(operation1) .then(operation2) .then(operation3) .then(operation4) .then(function(value4) { // Do something with value4 }, function (error) { // Handle any error from step1 through step4 }) .done()

可以看到，使用了第二種編程模式後能極大地提高我們的編程體驗，接著就讓我們自己動手實現一個支持序列執行的 Promise。(附：為了直觀的在瀏覽器上也能感受到 Promise，為此也寫了一段瀏覽器上的 Promise 用法示例)

在此之前，我們先要了解 Promise/A 提議中對單個非同步操作所作的抽象定義，定義具體如下所示：

  • Promise 操作只會處在 3 種狀態的一種：未完成態、完成態和失敗態。
  • Promise 的狀態只會出現從未完成態向完成態或失敗態轉化，不能逆反。完成態和失敗態不能相互轉化。
  • Promise 的狀態一旦轉化，將不能被更改。

Promise 的狀態轉化示意圖如下：

除此之外，Promise 對象的另一個關鍵就是需要具備 then() 方法，對於 then() 方法，有以下簡單的要求:

  • 接受完成態、錯誤態的回調方法。在操作完成或出現錯誤時，將會調用對應方法。
  • 可選地支持 progress 事件回調作為第三個方法。
  • then() 方法只接受 function 對象，其餘對象將被忽略。
  • then() 方法繼續返回 Promise 對象，已實現鏈式調用。

then() 方法的定義如下：

then(fulfilledHandler, errorHandler, progressHandler)

有了這些核心知識，接著進入 Promise/Deferred 核心代碼環節：

var Promise = function() { // 構建 Promise 對象 // 隊列用於存儲執行的回調函數 this.queue = [] this.isPromise = true}Promise.prototype.then = function (fulfilledHandler, errorHandler, progressHandler) { // 構建 Progress 的 then 方法 var handler = {} if (typeof fulfilledHandler === function) { handler.fulfilled = fulfilledHandler } if (typeof errorHandler === function) { handler.error = errorHandler } this.queue.push(handler) return this}

如上 Promise 的代碼就完成了，但是別忘了 Promise/Deferred 中的後者 Deferred，為了完成 Promise 的整個流程，我們還需要觸發執行上述回調函數的地方，實現這些功能的對象就叫作 Deferred，即延遲對象。

Promise 和 Deferred 的整體關係如下圖所示，從中可知，Deferred 主要用於內部來維護非同步模型的狀態；而 Promise 則作用於外部，通過 then() 方法暴露給外部以添加自定義邏輯。

接著來看 Deferred 代碼部分的實現：

var Deferred = function() { this.promise = new Promise()}// 完成態Deferred.prototype.resolve = function(obj) { var promise = this.promise var handler while(handler = promise.queue.shift()) { if (handler && handler.fulfilled) { var ret = handler.fulfilled(obj) if (ret && ret.isPromise) { // 這一行以及後面3行的意思是：一旦檢測到返回了新的 Promise 對象，停止執行，然後將當前 Deferred 對象的 promise 引用改變為新的 Promise 對象，並將隊列中餘下的回調轉交給它 ret.queue = promise.queue this.promise = ret return } } }}// 失敗態Deferred.prototype.reject = function(err) { var promise = this.promise var handler while (handler = promise.queue.shift()) { if (handler && handler.error) { var ret = handler.error(err) if (ret && ret.isPromise) { ret.queue = promise.queue this.promise = ret return } } }}// 生成回調函數Deferred.prototype.callback = function() { var that = this return function(err, file) { if(err) { return that.reject(err) } that.resolve(file) }}

接著我們以兩次文件讀取作為例子，來驗證該設計的可行性。這裡假設第二個文件讀取依賴於第一個文件中的內容，相關代碼如下：

var readFile1 = function(file, encoding) { var deferred = new Deferred() fs.readFile(file, encoding, deferred.callback()) return deferred.promise}var readFile2 = function(file, encoding) { var deferred = new Deferred() fs.readFile(file, encoding, deferred.callback()) return deferred.promise}readFile1(./file1.txt, utf8).then(function(file1) { // 這裡通過 then 把兩個回調存進隊列中 return readFile2(file1, utf8)}).then(function(file2) { console.log(file2) // I am file2.})

最後可以看到控制台輸出 I am file2，驗證成功~，這個案例的完整代碼可以點這裡查看，並建議使用 node-inspector 進行斷點觀察，(這段代碼裡面有些邏輯確實很繞，通過斷點調試就能較容易理解了)。

從 Promise 鏈式調用可以清晰地看到隊列(先進先出)的知識，其有如下兩個核心步驟：

  • 將所有的回調都存到隊列中；
  • Promise 完成時，逐個執行回調，一旦檢測到返回了新的 Promise 對象，停止執行，然後將當前 Deferred 對象的 promise 引用改變為新的 Promise 對象，並將隊列中餘下的回調轉交給它；

至此，實現了 Promise/Deferred 的完整邏輯，Promise 的其他知識未來也會繼續探究。

Generator

儘管 Promise 一定程度解決了回調地獄的問題，但是對於喜歡簡潔的程序員來說，一大堆的模板代碼 .then(data => {...}) 顯得不是很友好。所以愛折騰的開發者們在 ES6 中引人了 Generator 這種數據類型。仍然以讀取文件為例，先上一段非常簡潔的 Generator + co 的代碼：

co(function* () { const file1 = yield readFile(./file1.txt) const file2 = yield readFile(./file2.txt) console.log(file1) console.log(file2)})

可以看到比 Promise 的寫法簡潔了許多。後文會給出 co 庫的實現原理。在此之前，先歸納下什麼是 Generator。可以把 Generator 理解為一個可以遍歷的狀態機，調用 next 就可以切換到下一個狀態，其最大特點就是可以交出函數的執行權（即暫停執行），讓我們看如下代碼：

function* gen(x) { yield (function() {return 1})() var y = yield x + 2 return y}// 調用方式一var g = gen(1)g.next() // { value: 1, done: false }g.next() // { value: 3, done: false }g.next() // { value: undefined, done: true }// 調用方式二var g = gen(1)g.next() // { value: 1, done: false }g.next() // { value: 3, done: false }g.next(10) // { value: 10, done: true }

由此我們歸納下 Generator 的基礎知識：

  1. Generator 生成迭代器後，等待迭代器的 next() 指令啟動。
  2. 啟動迭代器後，代碼會運行到 yield 處停止。並返回一個 {value: AnyType, done: Boolean} 對象，value 是這次執行的結果，done 是迭代是否結束。並等待下一次的 next() 指令。
  3. next() 再次啟動，若 done 的屬性不為 true，則可以繼續從上一次停止的地方繼續迭代。
  4. 一直重複 2，3 步驟，直到 done 為 true。
  5. 通過調用方式二，我們可看到 next 方法可以帶一個參數，該參數就會被當作上一個 yield 語句的返回值。

另外我們注意到，上述代碼中的第一種調用方式中的 y 值是 undefined，如果我們真想拿到 y 值，就需要通過 g.next(); g.next().value 這種方式取出。可以看出，Generator 函數將非同步操作表示得很簡潔，但是流程管理卻不方便。這時候用於 Generator 函數的自動執行的 co 函數庫 登場了。為什麼 co 可以自動執行 Generator 函數呢？我們知道，Generator 函數就是一個非同步操作的容器。它的自動執行需要一種機制，當非同步操作有了結果，能夠自動交回執行權。

兩種方法可以做到這一點：

  • Thunk 函數。將非同步操作包裝成 Thunk 函數，在回調函數裡面交回執行權。
  • Promise 對象。將非同步操作包裝成 Promise 對象，用 then 方法交回執行權。

co 函數庫其實就是將兩種自動自動執行器(Thunk 函數和 Promise 對象)，包裝成一個庫。使用 co 的前提條件是，Generator 函數的 yield 命令後面，只能是 Thunk 函數或者是 Promise 對象。下面分別用以上兩種方法對 co 進行一個簡單的實現。

基於 Thunk 函數的自動執行

在 JavaScript 中，Thunk 函數就是指將多參數函數替換成單參數的形式，並且其只接受回調函數作為參數的函數。Thunk 函數的例子如下：

// 正常版本的 readFile(多參數)fs.readFile(filename, utf8, callback)// Thunk 版本的 readFile(單參數)function readFile(filename) { return function(callback) { fs.readFile(filename, utf8, callback); };}

在基於 Thunk 函數和 Generator 的知識上，接著我們來看看 co 基於 Thunk 函數的實現。(附：代碼參考自co最簡版實現)

function co(generator) { return function(fn) { var gen = generator() function next(err, result) { if(err) { return fn(err) } var step = gen.next(result) if (!step.done) { step.value(next) // 這裡可以把它聯想成遞歸；將非同步操作包裝成 Thunk 函數，在回調函數裡面交回執行權。 } else { fn(null, step.value) } } next() }}

用法如下：

co(function* () { // 把 function*() 作為參數 generator 傳入 co 函數 var file1 = yield readFile(./file1.txt) var file2 = yield readFile(./file2.txt) console.log(file1) // Im file1 console.log(file2) // Im file2 return done})(function(err, result) { // 這部分的 function 作為 co 函數內的 fn 的實參傳入 console.log(result) // done})

上述部分關鍵代碼已進行注釋，下面對 co 函數里的幾個難點進行說明：

  • var step = gen.next(result), 前文提到的一句話在這裡就很有用處了：next方法可以帶一個參數，該參數就會被當作上一個yield語句的返回值;在上述代碼的運行中一共會經過這個地方 3 次，result 的值第一次是空值，第二次是 file1.txt 的內容 Im file1，第三次是 file2.txt 的內容 Im file2。根據上述關鍵語句的提醒，所以第二次的內容會作為 file1 的值(當作上一個yield語句的返回值)，同理第三次的內容會作為 file2 的值。
  • 另一處是 step.value(next), step.value 就是前面提到的 thunk 函數返回的 function(callback) {}, next 就是傳入 thunk 函數的 callback。這句代碼是條遞歸語句，是這個簡易版 co 函數能自動調用 Generator 的關鍵語句。

建議親自跑一遍代碼，多打斷點，從而更好地理解，代碼已上傳github。

基於 Promise 對象的自動執行

基於 Thunk 函數的自動執行中，yield 後面需跟上 Thunk 函數，在基於 Promise 對象的自動執行中，yield 後面自然要跟 Promise 對象了，讓我們先構建一個 readFile 的

Promise 對象：

function readFile(fileName) { return new Promise(function(resolve, reject) { fs.readFile(fileName, function(error, data) { if (error) reject(error) resolve(data) }) })}

在基於前文 Promise 對象和 Generator 的知識上，接著我們來看看 co 基於 Promise 函數的實現:

function co(generator) { var gen = generator() function next(data) { var result = gen.next(data) // 同上，經歷了 3 次，第一次是 undefined，第二次是 Im file1，第三次是 Im file2 if (result.done) return result.value result.value.then(function(data) { // 將非同步操作包裝成 Promise 對象，用 then 方法交回執行權 next(data) }) } next()}

用法如下：

co(function* generator() { var file1 = yield readFile(./file1.txt) var file2 = yield readFile(./file2.txt) console.log(file1.toString()) // Im file1 console.log(file2.toString()) // Im file2})

這一部分的代碼上傳在這裡，通過觀察可以發現基於 Thunk 函數和基於 Promise 對象的自動執行方案的 co 函數設計思路幾乎一致，也因此呼應了它們共同的本質 —— 當非同步操作有了結果，自動交回執行權。

async

看上去 Generator 已經足夠好用了，但是使用 Generator 處理非同步必須得依賴 tj/co，於是 asycn 出來了。本質上 async 函數就是 Generator 函數的語法糖，這樣說是因為 async 函數的實現，就是將 Generator 函數和自動執行器，包裝進一個函數中。偽代碼如下，(註：其中 automatic 的實現可以參考 async 函數的含義和用法中的實現)

async function fn(args){ // ...}// 等同於function fn(args) { return automatic(function*() { // automatic 函數就是自動執行器，其的實現可以仿照 co 庫自動運行方案來實現，這裡就不展開了 // ... })}

接著仍然以上文的讀取文件為例，來比較 Generator 和 async 函數的寫法差異：

// Generatorvar genReadFile = co(function*() { var file1 = yield readFile(./file1.txt) var file2 = yield readFile(./file2.txt)})// 改用 async 函數var asyncReadFile = async function() { var file1 = await readFile(./file1.txt) var file2 = await 1 // 等同於同步操作（如果跟上原始類型的值）}

總體來說 async/await 看上去和使用 co 庫後的 generator 看上去很相似，不過相較於 Generator，可以看到 Async 函數更優秀的幾點：

  • 內置執行器。Generator 函數的執行必須依靠執行器，而 Aysnc 函數自帶執行器，調用方式跟普通函數的調用一樣；
  • 更好的語義。async 和 await 相較於 * 和 yield 更加語義化；
  • 更廣的適用性。前文提到的 co 模塊約定，yield 命令後面只能是 Thunk 函數或 Promise 對象，而 async 函數的 await 命令後面則可以是 Promise 或者原始類型的值；
  • 返回值是 Promise。async 函數返回值是 Promise 對象，比 Generator 函數返回的 Iterator 對象方便，因此可以直接使用 then() 方法進行調用；

參考資料

  • 深入淺出 Node.js
  • 理解回調函數
  • JavaScript之非同步編程簡述
  • 理解co執行邏輯
  • co 函數庫的含義和用法
  • async 函數的含義和用法

推薦閱讀：

TAG:Nodejs | 深入淺出Nodejs書籍 | 非同步 |