node中的流程控制中,co,thunkify为什么return callback()可以做到流程控制？

前言

我在学习generator ,yield ,co，thunkify的时候，有许多费解的地方，经过了许多的实践，也慢慢学会用，慢慢的理解，前一阵子有个其他项目的同事过来我们项目组学习node，发现他问的问题和我学习node的时候，遇到的困难都一样，所以产生写篇blog记录下co，thunkify的运用和原理，和园子里的神仙们交流交流，不对之处，还请指正，谢谢。

我在node的编写中，认真敲着敲着代码，然后回过头来发现，代码变成像这样子了，

var fs = require('fs'); 
//TODO:读取3个文件，然后...
fs.readFile('file1.txt','utf-8', function(err,data1){ 
    fs.readFile('file1.txt','utf-8', function(err,data2){ 
        fs.readFile('file1.txt','utf-8', function(err,data3){ 
            ...
        })
    })
})

或者在前端的代码中也有，

$.get("/b.txt", function(r1){
    $.get("/b.txt", function(r2){
        $.get("/b.txt", function(r3){
            $("div").html(r1+r2+r3);
        });
    });
});

这样子的代码的维护和代码量成倍数上涨，江湖人称回调金字塔，或邪恶金字塔，或者你可能说fs模块和jq,都可以提供同步的方式去调用接口，fs.readFileSync, 设置 async:true,通过同步的执行代码，但是javascript作为一种异步编程的语言，若使用同步的等待进行执行代码，那麽也将失去javascript原有的魅力。所以智慧无穷的人们通过一些新的技术或者新的运用去发挥js的异步编程的特性。

• 事件监听的方法,node原生有EventEmitter对象，通过on,emit的方式把业务行为进行解耦，而社区有eventproxy等一些成熟的包支持。
• promise对象，在jq1.5版本中始露尖角，运用在ajax调用中，后得到developer的广泛认可，在es6中纳入规范，当今的大多浏览器内置了这个对象，nodez中有q，bluebird等包。
• 协程，微线程，es6中的generator,其实协程并不是在es6中的新概念，在Lua中已经普遍的使用,其实说白了也是一种函数调用的方式，只不过会保留执行前后的上下文，node有co,thunkify等支持流程控制。
• 其他像async包，为流程控制而衍生的串行，并行，瀑布流，等概念，在早期的时候也被许多人运用（我们公司项目的前几个大版本的代码就是Async,coffeeScript写的，看得晕坨坨）

等等。

这些方法都可以去避免 回调金字塔的问题，在开发的成本上参差不齐。

而我再对generator函数做流程控制的时候，用的co,thunkify，用return callback(err,res);进行然后上一级调用的函数表示费解，而且不止co中有return callback(err,res); 在async 包也有,所以想弄清楚co中，究竟是怎么样去做流程控制的。

什么是thunk？

co，thunkify 的普通调用方式：

Demo1:

co(function*(){
    //打印的顺序为 timeout!alen ，1，yield 会开启一个协程并开始等待返回
    yield thunkify(genFun)("timeout! alen");
    console.log("over!")
}).catch(function(e){
    console.log(e.stack);
})

function genFun (param,callback){
    co(function*(){
        setTimeout(function () {
            console.log(param);
            return callback(null,1);
        }, 4000);
    }).catch(function(e){
        console.log(e.stack);
    })
}
// 运行结果：
//timeout! alen
// 1

 这里开始运行时，注意：

• 所有的yield 的东西必须包含在co的匿名generator函数里面，在co后面返回的是一个prmise，需要catch Error,不然回调函数里面的错误无法捕捉，为什么无法捕捉呢？因为我们的函数不是我们自己去调用的，是由node的底层去调用的，比如读取一个文件 fs.readFile("a.txt",fun(err,res)),为什么底层知道读取完a.txt，需要把错误放到err,而结果放到res中？回到函数在外部我们捕捉不到，node底层的实现是用多线程去定时的轮询，判断内核中是否有已经完成的i/o调用，若是有，则执行回调函数，(node,异步IO模型实现原理，详情请看《深入浅出NodeJS》第三章异步I/O)，所以执行回调函数的环境决定了我们回调函数的第一个参数必须是错误，然后传回给调用者。   

 

• thunkify 的作用是把所有的函数封装成thunk函数，那麽什么是thunk函数呢？以下是维基百科的一段引用：

In computer programming, a thunk is a subroutine that is created, often automatically, to assist a call to another subroutine. Thunks are primarily used to represent an additional calculation that a subroutine needs to execute, or to call a routine that does not support the usual calling mechanism. They have a variety of other applications to compiler code generation and in modular programming.

     在计算机编程中，thunk 就是一个自动被创建去帮助调用另一个子进程的子进程。Thunk 主要是被用于表示一个子进程需要执行的额外计算，或者调用一个不支持普通调用机制的子进程。Thunk 在代码编译生成和模块化编程中有这广泛的运用。

     这段话说的有点晕，thunk最初产生是被用解决函数的 传值调用和传名调用，而用于js,其实就是把多个参数thunk化成只接受一个参数的函数.

　　

thunkify 做了什么？

Demo2:

//demo1:
var thunkify = function(fn) {

    //thunkify 就是直接返回一个函数,在这个函数里面收集arguments这个类数组对象的参数,保存起来
    return function() {
        var args = new Array(arguments.length);
        var ctx = this;

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

        /**
         * 这里返回的函数是给co用的，把上一级收集的参数数组args的末尾加上一个function函数返回的回调函数,
         * 这个也是为什么 return callback(null,res); 可以返回到上一级调用的函数,因为每个yield 后的结果都会转成promise
         * ,在function thunkToPromise(fn)函数封装done
         */
        //TODO: 返回一个只有一个done的函数，其他的参数通过args变量用apply调用传入。
        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);
            }
        }
    };
}

　　所有将多个函数封装成只接受一个参数函数（详情请看 Thunk的含义），所以使用thunkify 封装函数的时候，

　　　　yield thunkify(genFun)("timeout! alen")

　　返回的是一个function(done){}, 可以看到

• thunkify(genFun) 这里返回一个函数，保存这里的上下文，这个返回的函数做的工作就是收集参数并封装在args中.
• 等到再次调用的时候,thunkify(genFun)("timeout! alen"),这里也还是返回一个函数，会在args参数的最后加上一个function,这个函数结合在co这个done是co中传入做错误收集的.
• 这里called 变量就是做的防止回调函数被调用两次。
• 上面保存的this很！有！用！，在默认的情况下我们包的引入的是这样的 var thunkify  = require("thunkify"); 调用的时候也是 thunkify(genFun)("timeout! alen"),若是没有指定this,如果也没有启用严格模式 use strict，所以我们普通的调用环境是global,但是有时候我们必须改变thunkify的this指向的应用，特别实在我们用到原型链的时候，比如：
  

  demo3:

  var co = require("co")
  var thunkify = require("thunkify")
  
  // 一个构造
  function Person(age){
      this.age = age;
  }
  // 一个获取通过名字获取姓名的方法
  Person.prototype.getAgeName = function(name,callback){
      var _this = this;
      co(function*(){
          setTimeout(function(){
              return callback(null,name+":"+_this.age);
          },2000);
      }).catch(function(e){
          console.log(e.stack);
          throw new Error(e.stack);
      })
  }
  
  // 开始调用
  co(function*(){
      var p = new Person(13);
      // 分别是两种不同的调用方法
      // var res = yield thunkify(p.getAgeName)("xiaoming");   # 1
       var res = yield thunkify(p.getAgeName).call(p,"xiaoming");  # 2
      console.log(res)
  }).catch(function(e){
      console.log(e.stack)
  })
  
  // 结果：
  // #1:undefined  
  // #2:xiaoming:13

  正是需要用call去改变thunkify返回的函数里的上下文，所以才需要在里面保存this的调用环境。

   

• 具体请看git-demo[有co,thunkify详细注解]

为什么return callback(err,res) 返回？

　　return callback(err,res); 可以做到返回，而且最初接触thunkify，co的时候，发现，callback 这个变量根本就是我们直接声明的！ 注意是声明，不是定义，因为我们仅仅是为了不让js解析器在语法检查的时候不抛出callback undefined 的错误，然后我们就自己在函数的参数，在实际的调用的是我们根本就没有传=.=,废话少说上图，

　　

　　是的，的确只是为了避免callback undefined 抛错而什么声明的，但是，它却是有赋值，在thunkfify中，下面:

　　

　　args.push(function{}) 这个push的元素就是我们的callback,在这里真正赋值。但是这个done参数，又是怎么工作的呢？

　　co函数：

/**
 * Execute the generator function or a generator
 * and return a promise.
 *
 * @param {Function} fn
 * @return {Promise}
 * @api public
 */

/**
 * 执行 generator 函数 或者 一个 generator 同时返回一个promise
 */

function co(gen) {
    var ctx = this;
    var args = slice.call(arguments, 1);

    // we wrap everything in a promise to avoid promise chaining,
    // which leads to memory leak errors.
    // see https://github.com/tj/co/issues/180
    return new Promise(function(resolve, reject) {
        if (typeof gen === 'function') gen = gen.apply(ctx, args);
        if (!gen || typeof gen.next !== 'function') return resolve(gen);

        onFulfilled();

        /**
         * @param {Mixed} res
         * @return {Promise}
         * @api private
         */
        function onFulfilled(res) {
            var ret;
            try {
                ret = gen.next(res);
            } catch (e) {
                return reject(e);
            }
            next(ret);
        }

        /**
         * @param {Error} err
         * @return {Promise}
         * @api private
         */

        function onRejected(err) {
            var ret;
            try {
                ret = gen.throw(err);
            } catch (e) {
                return reject(e);
            }
            next(ret);
        }

        /**
         * Get the next value in the generator,
         * return a promise.
         *
         * @param {Object} ret
         * @return {Promise}
         * @api private
         */
        /* 把在generator 中next()的值传进去,然后返回一个promise*/
        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) + '"'));
        }

    });
}

　　co,做了什么事情？

• co接受一个generator function ，然后收集arguments参数，然后直接 return new promise()!!!  也就是co和co之间是非阻塞的。所以

  co(function*(){
      // TODO:1 
  });
  co(function*(){
      // TODO:2
  })

  上述代码 1,2两处，不能保证一定是1先于比2完成。（估计也没人这么写=。=）

• gen = gen.apply(ctx, args); 开始调用我们传入的co中的generator 函数，

•  onFulfilled() ,函数中，gen.next(res)，执行到下一个yield 然后等待子程序返回，这时他返回的是一个function 正是thunkfiy最底层的那个

　　

　　所以第一次调用onFulfilled()，ret = gen.next(res);返回ret.value 是一个function ,而这个function 正是接受一个done为参数的函数。

　　然后去到next(ret),

　　next函数：

/* 把在generator 中next()的值传进去,然后返回一个promise*/
        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) + '"'));
        }

　　在这里，会依次判断是否已经完成，done==true, 否则，res.done =false，ret.value = function(done){},

　　那麽将ret.value 转成一个promise,而this指向ctx，调用co的上下文，而在value也是一个Promise 之后，然后then(onFulfilled, onRejected)，唯有这次value响应了，value的状态不在是pendding了,那麽才执行gen.next(res)的回调，所以

• 　　每个yield 返回的promise 唯有onFulfilled了之后，才会再次开始gen.next(),因为gen.next()在上一个yield 返回的promise的onFulfilled回调之中。
  

　　toPromise函数：

 /* 把一个被yield 过的值 转成promise */
function toPromise(obj) {
    if (!obj) return obj;
    if (isPromise(obj)) return obj;
    if (isGeneratorFunction(obj) || isGenerator(obj)) return co.call(this, obj);
    if ('function' == typeof obj) return thunkToPromise.call(this, obj);
    if (Array.isArray(obj)) return arrayToPromise.call(this, obj);
    if (isObject(obj)) return objectToPromise.call(this, obj);
    return obj;
}

 　　在这里，obj是一个function，所以会 return thunkToPromise.call(this, obj); 转成promise，并且吧obj 传进去，这个obj正是我们有一个done为参数的函数，

　　thunkToPromise函数：　　

 /* 把thunk函数转成promise */
function thunkToPromise(fn) {
    var ctx = this;
    return new Promise(function (resolve, reject) {
        fn.call(ctx, function (err, res) {
            if (err) return reject(err);
            if (arguments.length > 2) res = slice.call(arguments, 1);
            resolve(res);
        });
    });
}

 　　到这里，终于看到了done被赋值什么了，是个function(resolve, reject),这里直接返回一个Promise，这个Promise 做了什么？fn被调用，上下文是co调用的环境，done参数是一个回调函数，第一个参数是err,那麽这个和我们上面只是声明却没有定义的有什么关系？

　　我们申明的callback，在我们thunkify的是时候，就已经赋值了，

　　A,B,C 三张图，总结一下

• A是我们在调用thunkify话的函数具有多少个参数，而最后一个callback是在thunkify 里面在自动帮我们补上,也就是B上的args.push(function)。
• C呢，做的事情就是，传递一个done函数给我们thunkify()返回的函数，然后执行他。在这里执行callback的时候done.apply(null,arguments),这个时候会把我们平时return callbakc(err,res),这两个err,res,传进入arguments,去执行done ，而这个done函数，会去判断这个函数是否超过2个参数，如果超过两个参数，那麽就把err参数除外的函数封装成一个数组返回。
• 因为在generator 中 return 关键字会直接导致 return 以下的yield 失效，即是

　　

　　结果：

　　

　　所以，看到如果函数有return 那么，这个generator 函数会直接返回{done:true,value=XX}这也是为什么return callback(err,res); 就直接返回上一级函数了。yield thunkify() 唯有遇到return，generator中的所有yield都完成了， 或在本执行函数的过程中遇到reject，才会返回。

　　

总结

• co是通过传入一个generator函数，这个generator函数中，每个yield thunkify 返回的都是一个promise, 每次gen.next();的运行依赖上一个yield 返回的promise是否 onFulfilled ，若没有执行则throw Error()终止。
• 每个yield thunkify 的返回，都是通过 return ，throw Error,或者所有的yield 都已经onFulfilled了，所以才返回；
• thunkify中可以通过call,apply来改变this，进行保存原型链中的对象的调用上下文。