Future&Hope

co-parallel & co-gather源码解析

发表于  |    |   分类于   |  

最近看了Ma63d关于爬虫的这篇文章，正好自己也在做爬虫，看到他在文中提到了co-parallel和co-gather，就打算改一下自己的代码（本来代码就只是为了爬一些自己感兴趣的东西，现在还在改，地址在这里）。昨天也是好好的看了一下co-parallel的源码，今天打算自己来做一下解析。

co-parallel

源码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
var thread = require('co-thread');
module.exports = function *parallel(thunks, n){
  var n = Math.min(n || 5, thunks.length);
  var ret = [];
  var index = 0;
  function *next() {
    var i = index++;
    ret[i] = yield thunks[i];
    if (index < thunks.length) yield next;
  }
  yield thread(next, n);
  return ret;
};

这段代码真的是很短，但是方法真的很巧妙。因为两个方法用到了co-thread，这里把co-thread的源码也贴出来：

1
2
3
4
5
function thread(fn, n) {
  var gens = [];
  while (n--) gens.push(fn);
  return gens;
}

Run fn n times in parallel.

源码的描述就是为了parallel执行而创建的。由于在next外部创建了一个index变量，通过控制index的变化就可以使得每次执行的next函数都是不同的函数，在next中继续递归yield自己的话就是可以继续执行不同的next，最终把所有的thunks都yield了一遍，方法是不是很巧妙。

如果你觉得我说的话很混乱，那我们还是还是拿一个co-parallel中的例子来讲吧

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
var request = require('co-request');
var co = require('co');
var parallel = require('co-parallel');
var urls = [
  'http://google.com',
  'http://yahoo.com',
  'http://ign.com',
  'http://cloudup.com',
  'http://myspace.com',
  'http://facebook.com',
  'http://cuteoverload.com',
  'http://uglyoverload.com',
  'http://segment.io'
];
function *status(url) {
  console.log('GET %s', url);
  var s = (yield request(url)).statusCode;
  return s;
}
co(function *(){
  var start = Date.now();
  var reqs = urls.map(status);
  var res = yield parallel(reqs, 3);
  console.log(res);
  console.log('duration: %dms', Date.now() - start);
});

直接看到var reqs = urls.map(status);这句，由于传递给map的callback是一个Generator函数，所以最后的返回就是Generator的内部指针，也就是Iterator，也就是status()执行了一遍返回的结果。

再到var res = yield parallel(reqs, 3);这里，由于parallel是一个Generator，所以直接进入parallel中，因为n=3所以thread返回的数组内容应该是类似这个 [function*() {yield thunks[0]},function*(){yield thunks[1]}]。由于yield thread返回的结果是数组，在co中会对数组做Promise.all(obj.map(toPromise, this));因为obj中都是Generator，所以toPromise会直接对每一个Generator继续调用co(function*(){yield thunk[i]})。在next中最后又继续yield自己，所以当当一个thunk结束之后会继续下一个thunk。

pic

co-gather

co-gather实现的和co-parallel差不多的功能，只是增加了并行错误处理机制。因为Promise.all方法会在其中任何一个出问题的时候都把错误扔出来, co-gather是对all中每一个方法都做了错误处理，让Promise.all方法不会抛错，源码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
var thread = require('co-thread');
module.exports = function *gather(thunks, n){
  n = Math.min(n || 5, thunks.length);
  var ret = [];
  var index = 0;
  function *next() {
    var i = index++;
    ret[i] = {isError: false};
    try {
      ret[i].value = yield thunks[i];
    } catch (err) {
      ret[i].error = err;
      ret[i].isError = true;
    }
    if (index < thunks.length) yield next;
  }
  yield thread(next, n);
  return ret;
};

和co-parallel不同的地方就在于对yield thunks[i]做了一层try catch，然后返回的包含执行结果的对象。

总结

单单是看parallel的代码还是很好理解的，但是由于自己co的源码理解的不好，所以自己在捋清example的时候有点混乱了，后来又重新仔细的看了一遍了co的源码以及阮一峰的Generator的讲解，自己才弄明白。最后再次感谢Ma63d提供的思路。