ES6 Generator函數(shù)的異步應(yīng)用

1. 傳統(tǒng)方法

ES6 誕生以前，異步編程的方法，大概有下面四種。

• 回調(diào)函數(shù)
• 事件監(jiān)聽
• 發(fā)布/訂閱
• Promise 對象

Generator 函數(shù)將 JavaScript 異步編程帶入了一個(gè)全新的階段。

2. 基本概念

異步

所謂"異步"，簡單說就是一個(gè)任務(wù)不是連續(xù)完成的，可以理解成該任務(wù)被人為分成兩段，先執(zhí)行第一段，然后轉(zhuǎn)而執(zhí)行其他任務(wù)，等做好了準(zhǔn)備，再回過頭執(zhí)行第二段。

比如，有一個(gè)任務(wù)是讀取文件進(jìn)行處理，任務(wù)的第一段是向操作系統(tǒng)發(fā)出請求，要求讀取文件。然后，程序執(zhí)行其他任務(wù)，等到操作系統(tǒng)返回文件，再接著執(zhí)行任務(wù)的第二段（處理文件）。這種不連續(xù)的執(zhí)行，就叫做異步。

相應(yīng)地，連續(xù)的執(zhí)行就叫做同步。由于是連續(xù)執(zhí)行，不能插入其他任務(wù)，所以操作系統(tǒng)從硬盤讀取文件的這段時(shí)間，程序只能干等著。

回調(diào)函數(shù)

JavaScript 語言對異步編程的實(shí)現(xiàn)，就是回調(diào)函數(shù)。所謂回調(diào)函數(shù)，就是把任務(wù)的第二段單獨(dú)寫在一個(gè)函數(shù)里面，等到重新執(zhí)行這個(gè)任務(wù)的時(shí)候，就直接調(diào)用這個(gè)函數(shù)?；卣{(diào)函數(shù)的英語名字 callback ，直譯過來就是"重新調(diào)用"。

讀取文件進(jìn)行處理，是這樣寫的。

fs.readFile('/etc/passwd', 'utf-8', function (err, data) {
  if (err) throw err;
  console.log(data);
});

上面代碼中， readFile 函數(shù)的第三個(gè)參數(shù)，就是回調(diào)函數(shù)，也就是任務(wù)的第二段。等到操作系統(tǒng)返回了 /etc/passwd 這個(gè)文件以后，回調(diào)函數(shù)才會(huì)執(zhí)行。

一個(gè)有趣的問題是，為什么 Node 約定，回調(diào)函數(shù)的第一個(gè)參數(shù)，必須是錯(cuò)誤對象 err （如果沒有錯(cuò)誤，該參數(shù)就是 null ）？

原因是執(zhí)行分成兩段，第一段執(zhí)行完以后，任務(wù)所在的上下文環(huán)境就已經(jīng)結(jié)束了。在這以后拋出的錯(cuò)誤，原來的上下文環(huán)境已經(jīng)無法捕捉，只能當(dāng)作參數(shù)，傳入第二段。

Promise

回調(diào)函數(shù)本身并沒有問題，它的問題出現(xiàn)在多個(gè)回調(diào)函數(shù)嵌套。假定讀取 A 文件之后，再讀取 B 文件，代碼如下。

fs.readFile(fileA, 'utf-8', function (err, data) {
  fs.readFile(fileB, 'utf-8', function (err, data) {
    // ...
  });
});

不難想象，如果依次讀取兩個(gè)以上的文件，就會(huì)出現(xiàn)多重嵌套。代碼不是縱向發(fā)展，而是橫向發(fā)展，很快就會(huì)亂成一團(tuán)，無法管理。因?yàn)槎鄠€(gè)異步操作形成了強(qiáng)耦合，只要有一個(gè)操作需要修改，它的上層回調(diào)函數(shù)和下層回調(diào)函數(shù)，可能都要跟著修改。這種情況就稱為"回調(diào)函數(shù)地獄"（callback hell）。

Promise 對象就是為了解決這個(gè)問題而提出的。它不是新的語法功能，而是一種新的寫法，允許將回調(diào)函數(shù)的嵌套，改成鏈?zhǔn)秸{(diào)用。采用 Promise，連續(xù)讀取多個(gè)文件，寫法如下。

var readFile = require('fs-readfile-promise');


readFile(fileA)
.then(function (data) {
  console.log(data.toString());
})
.then(function () {
  return readFile(fileB);
})
.then(function (data) {
  console.log(data.toString());
})
.catch(function (err) {
  console.log(err);
});

上面代碼中，我使用了 fs-readfile-promise 模塊，它的作用就是返回一個(gè) Promise 版本的 readFile 函數(shù)。Promise 提供 then 方法加載回調(diào)函數(shù)， catch 方法捕捉執(zhí)行過程中拋出的錯(cuò)誤。

可以看到，Promise 的寫法只是回調(diào)函數(shù)的改進(jìn)，使用 then 方法以后，異步任務(wù)的兩段執(zhí)行看得更清楚了，除此以外，并無新意。

Promise 的最大問題是代碼冗余，原來的任務(wù)被 Promise 包裝了一下，不管什么操作，一眼看去都是一堆 then ，原來的語義變得很不清楚。

那么，有沒有更好的寫法呢？

3. Generator 函數(shù)

協(xié)程

傳統(tǒng)的編程語言，早有異步編程的解決方案（其實(shí)是多任務(wù)的解決方案）。其中有一種叫做"協(xié)程"（coroutine），意思是多個(gè)線程互相協(xié)作，完成異步任務(wù)。

協(xié)程有點(diǎn)像函數(shù)，又有點(diǎn)像線程。它的運(yùn)行流程大致如下。

• 第一步，協(xié)程 A 開始執(zhí)行。
• 第二步，協(xié)程 A 執(zhí)行到一半，進(jìn)入暫停，執(zhí)行權(quán)轉(zhuǎn)移到協(xié)程 B 。
• 第三步，（一段時(shí)間后）協(xié)程 B 交還執(zhí)行權(quán)。
• 第四步，協(xié)程 A 恢復(fù)執(zhí)行。

上面流程的協(xié)程 A ，就是異步任務(wù)，因?yàn)樗殖蓛啥危ɑ蚨喽危﹫?zhí)行。

舉例來說，讀取文件的協(xié)程寫法如下。

function* asyncJob() {
  // ...其他代碼
  var f = yield readFile(fileA);
  // ...其他代碼
}

上面代碼的函數(shù) asyncJob 是一個(gè)協(xié)程，它的奧妙就在其中的 yield 命令。它表示執(zhí)行到此處，執(zhí)行權(quán)將交給其他協(xié)程。也就是說， yield 命令是異步兩個(gè)階段的分界線。

協(xié)程遇到 yield 命令就暫停，等到執(zhí)行權(quán)返回，再從暫停的地方繼續(xù)往后執(zhí)行。它的最大優(yōu)點(diǎn)，就是代碼的寫法非常像同步操作，如果去除 yield 命令，簡直一模一樣。

協(xié)程的 Generator 函數(shù)實(shí)現(xiàn)

Generator 函數(shù)是協(xié)程在 ES6 的實(shí)現(xiàn)，最大特點(diǎn)就是可以交出函數(shù)的執(zhí)行權(quán)（即暫停執(zhí)行）。

整個(gè)Generator 函數(shù)就是一個(gè)封裝的異步任務(wù)，或者說是異步任務(wù)的容器。異步操作需要暫停的地方，都用 yield 語句注明。Generator 函數(shù)的執(zhí)行方法如下。

function* gen(x) {
  var y = yield x + 2;
  return y;
}


var g = gen(1);
g.next() // { value: 3, done: false }
g.next() // { value: undefined, done: true }

上面代碼中，調(diào)用 Generator 函數(shù)，會(huì)返回一個(gè)內(nèi)部指針（即遍歷器） g 。這是 Generator 函數(shù)不同于普通函數(shù)的另一個(gè)地方，即執(zhí)行它不會(huì)返回結(jié)果，返回的是指針對象。調(diào)用指針 g 的 next 方法，會(huì)移動(dòng)內(nèi)部指針（即執(zhí)行異步任務(wù)的第一段），指向第一個(gè)遇到的 yield 語句，上例是執(zhí)行到 x + 2 為止。

換言之， next 方法的作用是分階段執(zhí)行 Generator 函數(shù)。每次調(diào)用 next 方法，會(huì)返回一個(gè)對象，表示當(dāng)前階段的信息（ value 屬性和 done 屬性）。 value 屬性是 yield 語句后面表達(dá)式的值，表示當(dāng)前階段的值； done 屬性是一個(gè)布爾值，表示 Generator 函數(shù)是否執(zhí)行完畢，即是否還有下一個(gè)階段。

Generator 函數(shù)的數(shù)據(jù)交換和錯(cuò)誤處理

Generator 函數(shù)可以暫停執(zhí)行和恢復(fù)執(zhí)行，這是它能封裝異步任務(wù)的根本原因。除此之外，它還有兩個(gè)特性，使它可以作為異步編程的完整解決方案：函數(shù)體內(nèi)外的數(shù)據(jù)交換和錯(cuò)誤處理機(jī)制。

next 返回值的 value 屬性，是 Generator 函數(shù)向外輸出數(shù)據(jù)； next 方法還可以接受參數(shù)，向 Generator 函數(shù)體內(nèi)輸入數(shù)據(jù)。

function* gen(x){
  var y = yield x + 2;
  return y;
}


var g = gen(1);
g.next() // { value: 3, done: false }
g.next(2) // { value: 2, done: true }

上面代碼中，第一個(gè) next 方法的 value 屬性，返回表達(dá)式 x + 2 的值 3 。第二個(gè) next 方法帶有參數(shù) 2 ，這個(gè)參數(shù)可以傳入 Generator 函數(shù)，作為上個(gè)階段異步任務(wù)的返回結(jié)果，被函數(shù)體內(nèi)的變量 y 接收。因此，這一步的 value 屬性，返回的就是 2 （變量 y 的值）。

Generator 函數(shù)內(nèi)部還可以部署錯(cuò)誤處理代碼，捕獲函數(shù)體外拋出的錯(cuò)誤。

function* gen(x){
  try {
    var y = yield x + 2;
  } catch (e){
    console.log(e);
  }
  return y;
}


var g = gen(1);
g.next();
g.throw('出錯(cuò)了');
// 出錯(cuò)了

上面代碼的最后一行，Generator 函數(shù)體外，使用指針對象的 throw 方法拋出的錯(cuò)誤，可以被函數(shù)體內(nèi)的 try...catch 代碼塊捕獲。這意味著，出錯(cuò)的代碼與處理錯(cuò)誤的代碼，實(shí)現(xiàn)了時(shí)間和空間上的分離，這對于異步編程無疑是很重要的。

異步任務(wù)的封裝

下面看看如何使用 Generator 函數(shù)，執(zhí)行一個(gè)真實(shí)的異步任務(wù)。

var fetch = require('node-fetch');


function* gen(){
  var url = 'https://api.github.com/users/github';
  var result = yield fetch(url);
  console.log(result.bio);
}

上面代碼中，Generator 函數(shù)封裝了一個(gè)異步操作，該操作先讀取一個(gè)遠(yuǎn)程接口，然后從 JSON 格式的數(shù)據(jù)解析信息。就像前面說過的，這段代碼非常像同步操作，除了加上了 yield 命令。

執(zhí)行這段代碼的方法如下。

var g = gen();
var result = g.next();


result.value.then(function(data){
  return data.json();
}).then(function(data){
  g.next(data);
});

上面代碼中，首先執(zhí)行 Generator 函數(shù)，獲取遍歷器對象，然后使用 next 方法（第二行），執(zhí)行異步任務(wù)的第一階段。由于 Fetch 模塊返回的是一個(gè) Promise 對象，因此要用 then 方法調(diào)用下一個(gè) next 方法。

可以看到，雖然 Generator 函數(shù)將異步操作表示得很簡潔，但是流程管理卻不方便（即何時(shí)執(zhí)行第一階段、何時(shí)執(zhí)行第二階段）。

4. Thunk 函數(shù)

Thunk 函數(shù)是自動(dòng)執(zhí)行 Generator 函數(shù)的一種方法。

參數(shù)的求值策略

Thunk 函數(shù)早在上個(gè)世紀(jì) 60 年代就誕生了。

那時(shí)，編程語言剛剛起步，計(jì)算機(jī)學(xué)家還在研究，編譯器怎么寫比較好。一個(gè)爭論的焦點(diǎn)是"求值策略"，即函數(shù)的參數(shù)到底應(yīng)該何時(shí)求值。

var x = 1;


function f(m) {
  return m * 2;
}


f(x + 5)

上面代碼先定義函數(shù) f ，然后向它傳入表達(dá)式 x + 5 。請問，這個(gè)表達(dá)式應(yīng)該何時(shí)求值？

一種意見是"傳值調(diào)用"（call by value），即在進(jìn)入函數(shù)體之前，就計(jì)算 x + 5 的值（等于 6），再將這個(gè)值傳入函數(shù) f 。C 語言就采用這種策略。

f(x + 5)
// 傳值調(diào)用時(shí)，等同于
f(6)

另一種意見是“傳名調(diào)用”（call by name），即直接將表達(dá)式 x + 5 傳入函數(shù)體，只在用到它的時(shí)候求值。Haskell 語言采用這種策略。

f(x + 5)
// 傳名調(diào)用時(shí)，等同于
(x + 5) * 2

傳值調(diào)用和傳名調(diào)用，哪一種比較好？

回答是各有利弊。傳值調(diào)用比較簡單，但是對參數(shù)求值的時(shí)候，實(shí)際上還沒用到這個(gè)參數(shù)，有可能造成性能損失。

function f(a, b){
  return b;
}


f(3 * x * x - 2 * x - 1, x);

上面代碼中，函數(shù) f 的第一個(gè)參數(shù)是一個(gè)復(fù)雜的表達(dá)式，但是函數(shù)體內(nèi)根本沒用到。對這個(gè)參數(shù)求值，實(shí)際上是不必要的。因此，有一些計(jì)算機(jī)學(xué)家傾向于"傳名調(diào)用"，即只在執(zhí)行時(shí)求值。

Thunk 函數(shù)的含義

編譯器的“傳名調(diào)用”實(shí)現(xiàn)，往往是將參數(shù)放到一個(gè)臨時(shí)函數(shù)之中，再將這個(gè)臨時(shí)函數(shù)傳入函數(shù)體。這個(gè)臨時(shí)函數(shù)就叫做Thunk 函數(shù)。

function f(m) {
  return m * 2;
}


f(x + 5);


// 等同于


var thunk = function () {
  return x + 5;
};


function f(thunk) {
  return thunk() * 2;
}

上面代碼中，函數(shù) f 的參數(shù) x + 5 被一個(gè)函數(shù)替換了。凡是用到原參數(shù)的地方，對 Thunk 函數(shù)求值即可。

這就是 Thunk 函數(shù)的定義，它是“傳名調(diào)用”的一種實(shí)現(xiàn)策略，用來替換某個(gè)表達(dá)式。

JavaScript 語言的 Thunk 函數(shù)

JavaScript 語言是傳值調(diào)用，它的 Thunk 函數(shù)含義有所不同。在 JavaScript 語言中，Thunk 函數(shù)替換的不是表達(dá)式，而是多參數(shù)函數(shù)，將其替換成一個(gè)只接受回調(diào)函數(shù)作為參數(shù)的單參數(shù)函數(shù)。

// 正常版本的readFile（多參數(shù)版本）
fs.readFile(fileName, callback);


// Thunk版本的readFile（單參數(shù)版本）
var Thunk = function (fileName) {
  return function (callback) {
    return fs.readFile(fileName, callback);
  };
};


var readFileThunk = Thunk(fileName);
readFileThunk(callback);

上面代碼中， fs 模塊的 readFile 方法是一個(gè)多參數(shù)函數(shù)，兩個(gè)參數(shù)分別為文件名和回調(diào)函數(shù)。經(jīng)過轉(zhuǎn)換器處理，它變成了一個(gè)單參數(shù)函數(shù)，只接受回調(diào)函數(shù)作為參數(shù)。這個(gè)單參數(shù)版本，就叫做 Thunk 函數(shù)。

任何函數(shù)，只要參數(shù)有回調(diào)函數(shù)，就能寫成 Thunk 函數(shù)的形式。下面是一個(gè)簡單的 Thunk 函數(shù)轉(zhuǎn)換器。

// ES5版本
var Thunk = function(fn){
  return function (){
    var args = Array.prototype.slice.call(arguments);
    return function (callback){
      args.push(callback);
      return fn.apply(this, args);
    }
  };
};


// ES6版本
const Thunk = function(fn) {
  return function (...args) {
    return function (callback) {
      return fn.call(this, ...args, callback);
    }
  };
};

使用上面的轉(zhuǎn)換器，生成 fs.readFile 的 Thunk 函數(shù)。

var readFileThunk = Thunk(fs.readFile);
readFileThunk(fileA)(callback);

下面是另一個(gè)完整的例子。

function f(a, cb) {
  cb(a);
}
const ft = Thunk(f);


ft(1)(console.log) // 1

Thunkify 模塊

生產(chǎn)環(huán)境的轉(zhuǎn)換器，建議使用 Thunkify 模塊。

首先是安裝。

$ npm install thunkify

使用方式如下。

var thunkify = require('thunkify');
var fs = require('fs');


var read = thunkify(fs.readFile);
read('package.json')(function(err, str){
  // ...
});

Thunkify 的源碼與上一節(jié)那個(gè)簡單的轉(zhuǎn)換器非常像。

function thunkify(fn) {
  return function() {
    var args = new Array(arguments.length);
    var ctx = this;


    for (var i = 0; i < args.length; ++i) {
      args[i] = arguments[i];
    }


    return function (done) {
      var called;


      args.push(function () {
        if (called) return;
        called = true;
        done.apply(null, arguments);
      });


      try {
        fn.apply(ctx, args);
      } catch (err) {
        done(err);
      }
    }
  }
};

它的源碼主要多了一個(gè)檢查機(jī)制，變量 called 確?；卣{(diào)函數(shù)只運(yùn)行一次。這樣的設(shè)計(jì)與下文的 Generator 函數(shù)相關(guān)。請看下面的例子。

function f(a, b, callback){
  var sum = a + b;
  callback(sum);
  callback(sum);
}


var ft = thunkify(f);
var print = console.log.bind(console);
ft(1, 2)(print);
// 3

上面代碼中，由于 thunkify 只允許回調(diào)函數(shù)執(zhí)行一次，所以只輸出一行結(jié)果。

Generator 函數(shù)的流程管理

你可能會(huì)問， Thunk 函數(shù)有什么用？回答是以前確實(shí)沒什么用，但是 ES6 有了 Generator 函數(shù)，Thunk 函數(shù)現(xiàn)在可以用于 Generator 函數(shù)的自動(dòng)流程管理。

Generator 函數(shù)可以自動(dòng)執(zhí)行。

function* gen() {
  // ...
}


var g = gen();
var res = g.next();


while(!res.done){
  console.log(res.value);
  res = g.next();
}

上面代碼中，Generator 函數(shù) gen 會(huì)自動(dòng)執(zhí)行完所有步驟。

但是，這不適合異步操作。如果必須保證前一步執(zhí)行完，才能執(zhí)行后一步，上面的自動(dòng)執(zhí)行就不可行。這時(shí)，Thunk 函數(shù)就能派上用處。以讀取文件為例。下面的 Generator 函數(shù)封裝了兩個(gè)異步操作。

var fs = require('fs');
var thunkify = require('thunkify');
var readFileThunk = thunkify(fs.readFile);


var gen = function* (){
  var r1 = yield readFileThunk('/etc/fstab');
  console.log(r1.toString());
  var r2 = yield readFileThunk('/etc/shells');
  console.log(r2.toString());
};

上面代碼中， yield 命令用于將程序的執(zhí)行權(quán)移出 Generator 函數(shù)，那么就需要一種方法，將執(zhí)行權(quán)再交還給 Generator 函數(shù)。

這種方法就是 Thunk 函數(shù)，因?yàn)樗梢栽诨卣{(diào)函數(shù)里，將執(zhí)行權(quán)交還給 Generator 函數(shù)。為了便于理解，我們先看如何手動(dòng)執(zhí)行上面這個(gè) Generator 函數(shù)。

var g = gen();


var r1 = g.next();
r1.value(function (err, data) {
  if (err) throw err;
  var r2 = g.next(data);
  r2.value(function (err, data) {
    if (err) throw err;
    g.next(data);
  });
});

上面代碼中，變量 g 是 Generator 函數(shù)的內(nèi)部指針，表示目前執(zhí)行到哪一步。 next 方法負(fù)責(zé)將指針移動(dòng)到下一步，并返回該步的信息（ value 屬性和 done 屬性）。

仔細(xì)查看上面的代碼，可以發(fā)現(xiàn) Generator 函數(shù)的執(zhí)行過程，其實(shí)是將同一個(gè)回調(diào)函數(shù)，反復(fù)傳入 next 方法的 value 屬性。這使得我們可以用遞歸來自動(dòng)完成這個(gè)過程。

Thunk 函數(shù)的自動(dòng)流程管理

Thunk 函數(shù)真正的威力，在于可以自動(dòng)執(zhí)行 Generator 函數(shù)。下面就是一個(gè)基于 Thunk 函數(shù)的 Generator 執(zhí)行器。

function run(fn) {
  var gen = fn();


  function next(err, data) {
    var result = gen.next(data);
    if (result.done) return;
    result.value(next);
  }


  next();
}


function* g() {
  // ...
}


run(g);

上面代碼的 run 函數(shù)，就是一個(gè) Generator 函數(shù)的自動(dòng)執(zhí)行器。內(nèi)部的 next 函數(shù)就是 Thunk 的回調(diào)函數(shù)。 next 函數(shù)先將指針移到 Generator 函數(shù)的下一步（ gen.next 方法），然后判斷 Generator 函數(shù)是否結(jié)束（ result.done 屬性），如果沒結(jié)束，就將 next 函數(shù)再傳入 Thunk 函數(shù)（ result.value 屬性），否則就直接退出。

有了這個(gè)執(zhí)行器，執(zhí)行 Generator 函數(shù)方便多了。不管內(nèi)部有多少個(gè)異步操作，直接把 Generator 函數(shù)傳入 run 函數(shù)即可。當(dāng)然，前提是每一個(gè)異步操作，都要是 Thunk 函數(shù)，也就是說，跟在 yield 命令后面的必須是 Thunk 函數(shù)。

var g = function* (){
  var f1 = yield readFileThunk('fileA');
  var f2 = yield readFileThunk('fileB');
  // ...
  var fn = yield readFileThunk('fileN');
};


run(g);

上面代碼中，函數(shù) g 封裝了 n 個(gè)異步的讀取文件操作，只要執(zhí)行 run 函數(shù)，這些操作就會(huì)自動(dòng)完成。這樣一來，異步操作不僅可以寫得像同步操作，而且一行代碼就可以執(zhí)行。

Thunk 函數(shù)并不是 Generator 函數(shù)自動(dòng)執(zhí)行的唯一方案。因?yàn)樽詣?dòng)執(zhí)行的關(guān)鍵是，必須有一種機(jī)制，自動(dòng)控制 Generator 函數(shù)的流程，接收和交還程序的執(zhí)行權(quán)?；卣{(diào)函數(shù)可以做到這一點(diǎn)，Promise 對象也可以做到這一點(diǎn)。

5. co 模塊

基本用法

co 模塊是著名程序員 TJ Holowaychuk 于 2013 年 6 月發(fā)布的一個(gè)小工具，用于 Generator 函數(shù)的自動(dòng)執(zhí)行。

下面是一個(gè) Generator 函數(shù)，用于依次讀取兩個(gè)文件。

var gen = function* () {
  var f1 = yield readFile('/etc/fstab');
  var f2 = yield readFile('/etc/shells');
  console.log(f1.toString());
  console.log(f2.toString());
};

co 模塊可以讓你不用編寫 Generator 函數(shù)的執(zhí)行器。

var co = require('co');
co(gen);

上面代碼中，Generator 函數(shù)只要傳入 co 函數(shù)，就會(huì)自動(dòng)執(zhí)行。

co 函數(shù)返回一個(gè) Promise 對象，因此可以用 then 方法添加回調(diào)函數(shù)。

co(gen).then(function (){
  console.log('Generator 函數(shù)執(zhí)行完成');
});

上面代碼中，等到 Generator 函數(shù)執(zhí)行結(jié)束，就會(huì)輸出一行提示。

co 模塊的原理

為什么 co 可以自動(dòng)執(zhí)行 Generator 函數(shù)？

前面說過，Generator 就是一個(gè)異步操作的容器。它的自動(dòng)執(zhí)行需要一種機(jī)制，當(dāng)異步操作有了結(jié)果，能夠自動(dòng)交回執(zhí)行權(quán)。

兩種方法可以做到這一點(diǎn)。

（1）回調(diào)函數(shù)。將異步操作包裝成 Thunk 函數(shù)，在回調(diào)函數(shù)里面交回執(zhí)行權(quán)。

（2）Promise 對象。將異步操作包裝成 Promise 對象，用 then 方法交回執(zhí)行權(quán)。

co 模塊其實(shí)就是將兩種自動(dòng)執(zhí)行器（Thunk 函數(shù)和 Promise 對象），包裝成一個(gè)模塊。使用 co 的前提條件是，Generator 函數(shù)的 yield 命令后面，只能是 Thunk 函數(shù)或 Promise 對象。如果數(shù)組或?qū)ο蟮某蓡T，全部都是 Promise 對象，也可以使用 co，詳見后文的例子。

上一節(jié)已經(jīng)介紹了基于 Thunk 函數(shù)的自動(dòng)執(zhí)行器。下面來看，基于 Promise 對象的自動(dòng)執(zhí)行器。這是理解 co 模塊必須的。

基于 Promise 對象的自動(dòng)執(zhí)行

還是沿用上面的例子。首先，把 fs 模塊的 readFile 方法包裝成一個(gè) Promise 對象。

var fs = require('fs');


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


var gen = function* (){
  var f1 = yield readFile('/etc/fstab');
  var f2 = yield readFile('/etc/shells');
  console.log(f1.toString());
  console.log(f2.toString());
};

然后，手動(dòng)執(zhí)行上面的 Generator 函數(shù)。

var g = gen();


g.next().value.then(function(data){
  g.next(data).value.then(function(data){
    g.next(data);
  });
});

手動(dòng)執(zhí)行其實(shí)就是用 then 方法，層層添加回調(diào)函數(shù)。理解了這一點(diǎn)，就可以寫出一個(gè)自動(dòng)執(zhí)行器。

function run(gen){
  var g = gen();


  function next(data){
    var result = g.next(data);
    if (result.done) return result.value;
    result.value.then(function(data){
      next(data);
    });
  }


  next();
}


run(gen);

上面代碼中，只要 Generator 函數(shù)還沒執(zhí)行到最后一步， next 函數(shù)就調(diào)用自身，以此實(shí)現(xiàn)自動(dòng)執(zhí)行。

co 模塊的源碼

co 就是上面那個(gè)自動(dòng)執(zhí)行器的擴(kuò)展，它的源碼只有幾十行，非常簡單。

首先，co 函數(shù)接受 Generator 函數(shù)作為參數(shù)，返回一個(gè) Promise 對象。

function co(gen) {
  var ctx = this;


  return new Promise(function(resolve, reject) {
  });
}

在返回的 Promise 對象里面，co 先檢查參數(shù) gen 是否為 Generator 函數(shù)。如果是，就執(zhí)行該函數(shù)，得到一個(gè)內(nèi)部指針對象；如果不是就返回，并將 Promise 對象的狀態(tài)改為 resolved 。

function co(gen) {
  var ctx = this;


  return new Promise(function(resolve, reject) {
    if (typeof gen === 'function') gen = gen.call(ctx);
    if (!gen || typeof gen.next !== 'function') return resolve(gen);
  });
}

接著，co 將 Generator 函數(shù)的內(nèi)部指針對象的 next 方法，包裝成 onFulfilled 函數(shù)。這主要是為了能夠捕捉拋出的錯(cuò)誤。

function co(gen) {
  var ctx = this;


  return new Promise(function(resolve, reject) {
    if (typeof gen === 'function') gen = gen.call(ctx);
    if (!gen || typeof gen.next !== 'function') return resolve(gen);


    onFulfilled();
    function onFulfilled(res) {
      var ret;
      try {
        ret = gen.next(res);
      } catch (e) {
        return reject(e);
      }
      next(ret);
    }
  });
}

最后，就是關(guān)鍵的 next 函數(shù)，它會(huì)反復(fù)調(diào)用自身。

function next(ret) {
  if (ret.done) return resolve(ret.value);
  var value = toPromise.call(ctx, ret.value);
  if (value && isPromise(value)) return value.then(onFulfilled, onRejected);
  return onRejected(
    new TypeError(
      'You may only yield a function, promise, generator, array, or object, '
      + 'but the following object was passed: "'
      + String(ret.value)
      + '"'
    )
  );
}

上面代碼中， next 函數(shù)的內(nèi)部代碼，一共只有四行命令。

第一行，檢查當(dāng)前是否為 Generator 函數(shù)的最后一步，如果是就返回。

第二行，確保每一步的返回值，是 Promise 對象。

第三行，使用 then 方法，為返回值加上回調(diào)函數(shù)，然后通過 onFulfilled 函數(shù)再次調(diào)用 next 函數(shù)。

第四行，在參數(shù)不符合要求的情況下（參數(shù)非 Thunk 函數(shù)和 Promise 對象），將 Promise 對象的狀態(tài)改為 rejected ，從而終止執(zhí)行。

處理并發(fā)的異步操作

co 支持并發(fā)的異步操作，即允許某些操作同時(shí)進(jìn)行，等到它們?nèi)客瓿桑胚M(jìn)行下一步。

這時(shí)，要把并發(fā)的操作都放在數(shù)組或?qū)ο罄锩?，跟? yield 語句后面。

// 數(shù)組的寫法
co(function* () {
  var res = yield [
    Promise.resolve(1),
    Promise.resolve(2)
  ];
  console.log(res);
}).catch(onerror);


// 對象的寫法
co(function* () {
  var res = yield {
    1: Promise.resolve(1),
    2: Promise.resolve(2),
  };
  console.log(res);
}).catch(onerror);

下面是另一個(gè)例子。

co(function* () {
  var values = [n1, n2, n3];
  yield values.map(somethingAsync);
});


function* somethingAsync(x) {
  // do something async
  return y
}

上面的代碼允許并發(fā)三個(gè) somethingAsync 異步操作，等到它們?nèi)客瓿桑艜?huì)進(jìn)行下一步。

實(shí)例：處理 Stream

Node 提供 Stream 模式讀寫數(shù)據(jù)，特點(diǎn)是一次只處理數(shù)據(jù)的一部分，數(shù)據(jù)分成一塊塊依次處理，就好像“數(shù)據(jù)流”一樣。這對于處理大規(guī)模數(shù)據(jù)非常有利。Stream 模式使用 EventEmitter API，會(huì)釋放三個(gè)事件。

• data 事件：下一塊數(shù)據(jù)塊已經(jīng)準(zhǔn)備好了。
• end 事件：整個(gè)“數(shù)據(jù)流”處理完了。
• error 事件：發(fā)生錯(cuò)誤。

使用 Promise.race()函數(shù)，可以判斷這三個(gè)事件之中哪一個(gè)最先發(fā)生，只有當(dāng) data 事件最先發(fā)生時(shí)，才進(jìn)入下一個(gè)數(shù)據(jù)塊的處理。從而，我們可以通過一個(gè) while 循環(huán)，完成所有數(shù)據(jù)的讀取。

const co = require('co');
const fs = require('fs');


const stream = fs.createReadStream('./les_miserables.txt');
let valjeanCount = 0;


co(function*() {
  while(true) {
    const res = yield Promise.race([
      new Promise(resolve => stream.once('data', resolve)),
      new Promise(resolve => stream.once('end', resolve)),
      new Promise((resolve, reject) => stream.once('error', reject))
    ]);
    if (!res) {
      break;
    }
    stream.removeAllListeners('data');
    stream.removeAllListeners('end');
    stream.removeAllListeners('error');
    valjeanCount += (res.toString().match(/valjean/ig) || []).length;
  }
  console.log('count:', valjeanCount); // count: 1120
});

上面代碼采用 Stream 模式讀取《悲慘世界》的文本文件，對于每個(gè)數(shù)據(jù)塊都使用 stream.once 方法，在 data 、 end 、 error 三個(gè)事件上添加一次性回調(diào)函數(shù)。變量 res 只有在 data 事件發(fā)生時(shí)才有值，然后累加每個(gè)數(shù)據(jù)塊之中 valjean 這個(gè)詞出現(xiàn)的次數(shù)。