ES6标准入门(第 3 版)
17.4 co模块
阅读(

ECMAScript 6 简介

let 和 const 命令

变量的解构赋值

字符串的扩展

正则的扩展

数值的扩展

函数的扩展

数组的扩展

对象的扩展

Symbol

Set 和 Map 数据结构

Proxy

Reflect对象

Promise对象

Iterator 和 for...of 循环

Generator函数语法

Generator函数的异步应用

基本概念

ES6 Generator函数实现协程coroutine

ES6 Thunk函数

co模块

基本用法

co模块是著名程序员 TJ Holowaychuk 于 2013 年 6 月发布的一个小工具,用于 Generator函数的自动执行。

下面是一个 Generator函数,用于依次读取两个文件。

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函数的执行器。

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

上面代码中,Generator函数只要传入co函数,就会自动执行。

co函数返回一个Promise对象,因此可以用then方法添加回调函数。

co(gen).then(function (){
  console.log('Generator函数执行完成');
});

上面代码中,等到 Generator函数执行结束,就会输出一行提示。

co模块的原理

为什么 co 可以自动执行 Generator函数?

前面说过,Generator 就是一个异步操作的容器。它的自动执行需要一种机制,当异步操作有了结果,能够自动交回执行权。

两种方法可以做到这一点。

(1)回调函数。将异步操作包装成 Thunk 函数,在回调函数里面交回执行权。

(2)Promise 对象。将异步操作包装成 Promise 对象,用then方法交回执行权。

co模块其实就是将两种自动执行器(Thunk 函数和 Promise 对象),包装成一个模块。使用 co 的前提条件是,Generator函数的yield命令后面,只能是 Thunk 函数或 Promise 对象。如果数组或对象的成员,全部都是 Promise 对象,也可以使用 co,详见后文的例子。

上一节已经介绍了基于 Thunk 函数的自动执行器。下面来看,基于 Promise 对象的自动执行器。这是理解 co模块必须的。

基于 Promise 对象的自动执行

还是沿用上面的例子。首先,把fs模块的readFile方法包装成一个 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());
};

然后,手动执行上面的 Generator函数。

var g = gen();

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

手动执行其实就是用then方法,层层添加回调函数。理解了这一点,就可以写出一个自动执行器。

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函数还没执行到最后一步,next函数就调用自身,以此实现自动执行。

co模块的源码

co 就是上面那个自动执行器的扩展,它的源码只有几十行,非常简单。

首先,co 函数接受 Generator函数作为参数,返回一个 Promise 对象。

function co(gen) {
  var ctx = this;

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

在返回的 Promise 对象里面,co 先检查参数gen是否为 Generator函数。如果是,就执行该函数,得到一个内部指针对象;如果不是就返回,并将 Promise 对象的状态改为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函数的内部指针对象的next方法,包装成onFulfilled函数。这主要是为了能够捕捉抛出的错误。

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);
    }
  });
}

最后,就是关键的next函数,它会反复调用自身。

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函数的内部代码,一共只有四行命令。

第一行,检查当前是否为 Generator函数的最后一步,如果是就返回。

第二行,确保每一步的返回值,是 Promise 对象。

第三行,使用then方法,为返回值加上回调函数,然后通过onFulfilled函数再次调用next函数。

第四行,在参数不符合要求的情况下(参数非 Thunk 函数和 Promise 对象),将 Promise 对象的状态改为rejected,从而终止执行。

处理并发的异步操作

co 支持并发的异步操作,即允许某些操作同时进行,等到它们全部完成,才进行下一步。

这时,要把并发的操作都放在数组或对象里面,跟在yield语句后面。

// 数组的写法
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);

下面是另一个例子。

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

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

上面的代码允许并发三个somethingAsync异步操作,等到它们全部完成,才会进行下一步。

实例:处理 Stream

Node 提供 Stream 模式读写数据,特点是一次只处理数据的一部分,数据分成一块块依次处理,就好像“数据流”一样。这对于处理大规模数据非常有利。Stream 模式使用 EventEmitter API,会释放三个事件。

  • data事件:下一块数据块已经准备好了。
  • end事件:整个“数据流”处理“完了。
  • error事件:发生错误。

使用Promise.race()函数,可以判断这三个事件之中哪一个最先发生,只有当data事件最先发生时,才进入下一个数据块的处理。从而,我们可以通过一个while循环,完成所有数据的读取。

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 模式读取《悲惨世界》的文本文件,对于每个数据块都使用stream.once方法,在dataenderror三个事件上添加一次性回调函数。变量res只有在data事件发生时才有值,然后累加每个数据块之中valjean这个词出现的次数。

如果本教程对您帮助很大,请随意打赏。您的支持,将鼓励我们提供更好的教程!

← 键盘方向键翻页 →
返回顶部 手机访问 关注微信 返回底部

扫码访问歪脖网

随时随地,想看就看

关注歪脖网微信

分享 web 知识、交流 web 经验