基本概念

異步

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

相應(yīng)地，連續(xù)的執(zhí)行就叫做同步。由于是連續(xù)執(zhí)行，不能插入其他任務(wù)，所以這段時(shí)間，程序只能干等著。

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

JavaScript 語(yǔ)言對(duì)異步編程的實(shí)現(xiàn)，就是回調(diào)函數(shù)。所謂回調(diào)函數(shù)，就是把任務(wù)的第二段單獨(dú)寫(xiě)在一個(gè)函數(shù)里面，等到重新執(zhí)行這個(gè)任務(wù)的時(shí)候，就直接調(diào)用這個(gè)函數(shù)。

Promise

回調(diào)函數(shù)本身并沒(méi)有問(wèn)題，它的問(wèn)題出現(xiàn)在多個(gè)回調(diào)函數(shù)嵌套。

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

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

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

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í)行過(guò)程中拋出的錯(cuò)誤。

可以看到，Promise 的寫(xiě)法只是回調(diào)函數(shù)的改進(jìn)。Promise 的最大問(wèn)題是代碼冗余，原來(lái)的任務(wù)被 Promise 包裝了一下，不管什么操作，一眼看去都是一堆then，原來(lái)的語(yǔ)義變得很不清楚。

Generator 函數(shù)

協(xié)程

"協(xié)程"（coroutine），意思是多個(gè)線程互相協(xié)作，完成異步任務(wù)。

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

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

讀取文件的協(xié)程寫(xiě)法如下

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

協(xié)程遇到y(tǒng)ield命令就暫停，執(zhí)行權(quán)將交給其他協(xié)程，等到執(zhí)行權(quán)返回，再?gòu)臅和５牡胤嚼^續(xù)往后執(zhí)行。它的最大優(yōu)點(diǎn)，就是代碼的寫(xiě)法非常像同步操作，如果去除yield命令，簡(jiǎn)直一模一樣。yield命令是異步兩個(gè)階段的分界線

協(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ù)，或者說(shuō)是異步任務(wù)的容器。異步操作需要暫停的地方，都用yield語(yǔ)句注明。

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é)果，返回的是指針對(duì)象。調(diào)用指針g的next方法，會(huì)移動(dòng)內(nèi)部指針（即執(zhí)行異步任務(wù)的第一段），指向第一個(gè)遇到的yield語(yǔ)句，上例是執(zhí)行到x + 2為止。

換言之，next方法的作用是分階段執(zhí)行Generator函數(shù)。每次調(diào)用next方法，會(huì)返回一個(gè)對(duì)象，表示當(dāng)前階段的信息（value屬性和done屬性）。value屬性是yield語(yǔ)句后面表達(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ù)。輸入值將賦值給上一個(gè)yield表達(dá)式，不傳參即上一個(gè)yield表達(dá)式為undefined

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

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

異步任務(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);
}

var g = gen();
var result = g.next(); // fetch(url)返回一個(gè)promise

// 即result.value是一個(gè)promise
result.value.then(function(data){
  return data.json();
}).then(function(data){
  g.next(data);
});
// 第一個(gè)then 取到fetch(url)操作完成后的結(jié)果 并轉(zhuǎn)換為json格式
// 第一個(gè)then函數(shù)又返回一個(gè)promise 再次調(diào)用then方法 將轉(zhuǎn)化成json 的數(shù)據(jù) 通過(guò)next的參數(shù)傳回 gen函數(shù)的 result
// 執(zhí)行console.log(result.bio) 這里的result就是外面?zhèn)鬟M(jìn)來(lái)的json化之后的數(shù)據(jù)

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

Thunk 函數(shù)

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

參數(shù)的求值策略

• 傳值調(diào)用：即在進(jìn)入函數(shù)體之前，就計(jì)算參數(shù)的值，再將這個(gè)值傳入函數(shù)。
• 傳名調(diào)用：即直接將表達(dá)式傳入函數(shù)體，只在用到它的時(shí)候求值。

傳值調(diào)用比較簡(jiǎn)單，但是對(duì)參數(shù)求值的時(shí)候，實(shí)際上還沒(méi)用到這個(gè)參數(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;
}

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

JavaScript 語(yǔ)言的 Thunk 函數(shù)

在 JavaScript 語(yǔ)言中，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);
// 這里的readFileThunk 就是一個(gè)thunk函數(shù)，只接受回調(diào)函數(shù)作為參數(shù)的單參數(shù)函數(shù)

但是這里的Thunk函數(shù)，只能處理fs.readFile的Thunk化，不能處理其他的。

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

const Thunk = function(fn) {
  return function (...args) {
    return function (callback) {
      return fn.call(this, ...args, callback);
    }
  };
};
// 再包一層，將fs.readFile這個(gè)函數(shù)，也直接當(dāng)做參數(shù)傳遞進(jìn)來(lái)

var readFileThunk = Thunk(fs.readFile);
// 這里的readFileThunk就相當(dāng)于是上例中的Thunk
// readFileThunk(fileA) 返回一個(gè)thunk函數(shù)，即上例中的 readFileThunk
readFileThunk(fileA)(callback);

// 在參數(shù)有回調(diào)函數(shù) 上的使用

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

ft(1)(console.log) // 1
// 這里ft(1)就是一個(gè)thunk函數(shù)

Thunkify 模塊

生產(chǎn)環(huán)境的轉(zhuǎn)換器，建議使用 Thunkify 模塊，即使用成熟 Thunkify模塊來(lái)代替上例中的簡(jiǎn)單Thunk轉(zhuǎn)換器

安裝： npm install thunkify

使用

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

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

源碼：主要多了一個(gè)檢查機(jī)制，變量called確?；卣{(diào)函數(shù)只運(yù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];
    }
    // 這里的arg 針對(duì)上面使用過(guò)的來(lái)講就是 ['package.json']
    return function (done) {
      // 這里的done就是 上面 read('package.json')后傳入的回調(diào)
      var called;

      // 第一步：把done這個(gè)回調(diào)，再包一層，并設(shè)置called開(kāi)關(guān)，確保done這個(gè)回調(diào)只執(zhí)行一次
      // 這里的使用arg.push() 將一個(gè)包裝后的函數(shù)加入args，并在之后一起傳給fn
      args.push(function () {
        // 第三步：這個(gè)函數(shù) 就是 fs.readFile('package.json', cb)中的 cb
        // 在讀取完成，就會(huì)執(zhí)行這里的代碼

        // 設(shè)置開(kāi)關(guān) 讓回調(diào)只執(zhí)行一次
        if (called) return;
        called = true;

        // 這里的arguments 就是 [err, data] 傳遞給了回調(diào)函數(shù)并執(zhí)行
        done.apply(null, arguments);
      });

      try {
        // 第二步：下面這行代碼即相當(dāng)于執(zhí)行fs.readFile.apply(ctx, ['package.json', 包裝后的函數(shù)])
        // 所以包裝好的函數(shù) 就是 fs.readFile('package.json', cb)中的 cb
        fn.apply(ctx, args);
      } catch (err) {
        // 如果報(bào)錯(cuò) 把err對(duì)象傳遞給回調(diào)函數(shù)
        done(err);
      }
    }
  }
};

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

Thunk 函數(shù)有什么用？以前確實(shí)沒(méi)什么用，但是 ES6 有了 Generator 函數(shù)，Thunk 函數(shù)現(xiàn)在可以用于 Generator 函數(shù)的自動(dòng)流程管理。

Generator如何自動(dòng)執(zhí)行同步操作:

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

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

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

關(guān)于異步操作。如果必須保證前一步執(zhí)行完，才能執(zhí)行后一步，上面的自動(dòng)執(zhí)行就不可行。這時(shí)，Thunk 函數(shù)就能派上用處。

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

var gen = function* (){
  var r1 = yield readFileThunk('fileA');
  console.log(r1.toString());
  var r2 = yield readFileThunk('fileB');
  console.log(r2.toString());
};

// readFileThunk('fileA') 返回的就是一個(gè)thunk函數(shù)，接受一個(gè)回調(diào)函數(shù)作為參數(shù)
// 該回調(diào)最終也是fs.readFile('fileA',cb)中的cb

// 手動(dòng)執(zhí)行上面的gen函數(shù) 
var g = gen();

var r1 = g.next();
// r1.value就是一個(gè)thunk函數(shù) 因?yàn)閞eadFileThunk('fileA') 返回的就是一個(gè)thunk函數(shù)。
// 所以可以接受一個(gè)回調(diào)作為參數(shù)，這個(gè)回調(diào)會(huì)在 讀取完fileA后執(zhí)行
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);
  });
});

// 如果讀取fileA報(bào)錯(cuò)就拋錯(cuò)
// 讀取正常，就再將執(zhí)行權(quán)交換給gen函數(shù)，并把讀取后得到的data交給gen函數(shù)做讀取后操作
// 即 gen函數(shù)中的r1 就是 讀取fileA得到的data，這時(shí)候gen函數(shù)就像是同步執(zhí)行的代碼
// console.log(r1.toString())就相當(dāng)于是文件讀取后的操作
// 然后接著執(zhí)行g(shù)en函數(shù)，會(huì)再次遇到y(tǒng)ield，這時(shí)候又把執(zhí)行權(quán)交出來(lái)
// 執(zhí)行r2.value(cb)，r2.value也是thunk ... 依次循環(huán)上面操作

yield命令用于將程序的執(zhí)行權(quán)移出 Generator 函數(shù)，那么就需要一種方法，將執(zhí)行權(quán)再交還給 Generator 函數(shù)。（手動(dòng)執(zhí)行就是調(diào)用g.next）

這種方法就是 Thunk 函數(shù)，因?yàn)樗梢栽?strong>回調(diào)函數(shù)里，將執(zhí)行權(quán)交還給 Generator 函數(shù)。

使用thunk來(lái)自動(dòng)執(zhí)行 Generator 函數(shù)

核心就是在回調(diào)中將 執(zhí)行權(quán)傳入gen函數(shù)，并獲取到上一個(gè)異步操作的thunk函數(shù)將回調(diào)傳入，在上一個(gè)異步操作完成后會(huì)觸發(fā)回調(diào)，如此反復(fù)交換執(zhí)行權(quán)。

當(dāng)然，前提是每一個(gè)異步操作，都要是 Thunk 函數(shù)，也就是說(shuō)，跟在yield命令后面的必須是 Thunk 函數(shù)。

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

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

  cb();
}

function* g() {
  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)完成。
// 這樣一來(lái)，異步操作不僅可以寫(xiě)得像同步操作，而且一行代碼就可以執(zhí)行。

co 模塊

co 模塊用于 Generator 函數(shù)的自動(dòng)執(zhí)行。co 模塊可以讓你不用編寫(xiě) Generator 函數(shù)的執(zhí)行器。

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

var co = require('co');
// Generator 函數(shù)只要傳入co函數(shù)，就會(huì)自動(dòng)執(zhí)行。
co(gen).then(function (){
  console.log('Generator 函數(shù)執(zhí)行完成');
});
// co函數(shù)返回一個(gè)Promise對(duì)象，因此可以用then方法添加回調(diào)函數(shù)。

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

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

• 回調(diào)函數(shù)。將異步操作包裝成 Thunk 函數(shù)，在回調(diào)函數(shù)里面交回執(zhí)行權(quán)。見(jiàn)：使用thunk來(lái)自動(dòng)執(zhí)行 Generator 函數(shù)
• Promise 對(duì)象。將異步操作包裝成 Promise 對(duì)象，用then方法交回執(zhí)行權(quán)。見(jiàn)：基于 Promise 對(duì)象的自動(dòng)執(zhí)行

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

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

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í)行
var g = gen();

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

// g.next().value
// .then(data => {
//   return g.next(data).value
// }).then(data => {
//   g.next(data);
// })

手動(dòng)執(zhí)行其實(shí)就是用then方法，層層添加回調(diào)函數(shù)。理解了這一點(diǎn)，就可以寫(xiě)出一個(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);

co 模塊的源碼

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

function co(gen) {
  var ctx = this;

  return new Promise(function(resolve, reject) {
    // 先檢查參數(shù)gen是否為 Generator 函數(shù)。如果是，就執(zhí)行該函數(shù)，得到一個(gè)內(nèi)部指針對(duì)象；
    // 如果不是就返回，并將 Promise 對(duì)象的狀態(tài)改為resolved。
    if (typeof gen === 'function') gen = gen.call(ctx);
    if (!gen || typeof gen.next !== 'function') return resolve(gen);

    // co 將 Generator 函數(shù)的內(nèi)部指針對(duì)象的next方法，包裝成onFulfilled函數(shù)
    // 這主要是為了能夠捕捉拋出的錯(cuò)誤。
    onFulfilled();
    function onFulfilled(data) {
      var result;
      try {
        result = gen.next(data); // 啟動(dòng)gen函數(shù)的執(zhí)行
      } catch (e) {
        return reject(e); // 如果錯(cuò)誤就拋出
      }
      next(result);
    }

    // 關(guān)鍵的next函數(shù)，它會(huì)反復(fù)調(diào)用自身。
    function next(result) {
      // 檢查當(dāng)前是否為 Generator 函數(shù)的最后一步，如果是就返回
      if (result.done) return resolve(result.value);

      // 確保每一步的返回值，是 Promise 對(duì)象。
      var value = toPromise.call(ctx, result.value);

      // 使用then方法，為返回值加上回調(diào)函數(shù)，然后通過(guò)onFulfilled函數(shù)再次調(diào)用next函數(shù)。
      if (value && isPromise(value)) return value.then(onFulfilled, onRejected);

      // 上面的if為false時(shí) 執(zhí)行下面代碼
      // 即參數(shù)不符合要求，拋出錯(cuò)誤
      return onRejected(
        new TypeError(
          'You may only yield a function, promise, generator, array, or object, '
          + 'but the following object was passed: "'
          + String(result.value)
          + '"'
        )
      );
    }
  });
}

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

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

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

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

// toPromise 可能會(huì)將這些異步操作轉(zhuǎn)為數(shù)組傳入Promise.all中

實(shí)例：處理 Stream

Node 提供 Stream 模式讀寫(xiě)數(shù)據(jù)，特點(diǎn)是一次只處理數(shù)據(jù)的一部分，數(shù)據(jù)分成一塊塊依次處理，就好像“數(shù)據(jù)流”一樣。這對(duì)于處理大規(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ù)塊的處理。從而，我們可以通過(guò)一個(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) {
    // 對(duì)于每個(gè)數(shù)據(jù)塊都使用stream.once方法，在data、end、error三個(gè)事件上添加 一次性 回調(diào)函數(shù)。
    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))
    ]);
    // res只有在data事件發(fā)生時(shí)才有值
    // 只有data事件會(huì)在回調(diào)中傳遞參數(shù)(數(shù)據(jù))給resolve，resolve作為回調(diào)函數(shù)即resolve(數(shù)據(jù))
    // 所以在co模塊中下一次g.next(data)中的data才有值，然后賦值給res, res才能為true。

    // end 事件中 回調(diào)事件沒(méi)有參數(shù)
    // error事件 直接就拋錯(cuò)處理了
    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
});

這個(gè)函數(shù)就是用來(lái)讀取 les_miserables.txt 中 valjean 出現(xiàn)了多少次。