Promise异步控制流模式

以前的博客在简书和github.io：https://chijitui.github.io/，但感受简书好像凉凉了就搬家来掘金了，以后会不按期更关于 Node.js 设计模式和微服务的东西，欢迎投喂点关注，爱您！

Node.js 风格函数的 promise 化

在 Javascript 中， 并不是全部的异步控制函数和库都支持开箱即用的promise,因此在大多数状况下都须要吧一个典型的基于回调的函数转换成一个返回promise的函数，好比：

function promisify(callbackBaseApi) {
    return function promisified() {
        const args = [].slice.call(arguments);
        return new Promise((resolve, reject) => {
            args.push((err, result) => {
                if(err) {
                    return reject(err);
                }
                if(arguments.length <= 2) {
                    resolve(result);
                } else {
                    resolve([].slice.call(arguments, 1));
                }
            }); 
            callbackBaseApi.apply(null, args);
        });
    }
}
复制代码

如今的 Node.js 核心实用工具库util里面已经支持(err, value) => ...回调函数是最后一个参数的函数, 返回一个返回值是一个promise版本的函数。

顺序执行流的迭代模式

在看异步控制流模式以前，先开始分析顺序执行流。按顺序执行一组任务意味着一次运行一个任务，一个接一个地运行。执行顺序很重要，必须保留，由于列表中任务运行的结果可能影响下一个任务的执行，好比：

start -> task1 -> task2 -> task3 -> end
复制代码

这种流程通常都有着几个特色：

• 按顺序执行一组已知任务，而没有连接或者传播结果；
• 一个任务的输出做为下一个的输入；
• 在每一个元素上运行异步任务时迭代一个集合，一个接一个。

这种执行流直接用在阻塞的 API 中并无太多问题，可是，在咱们使用非阻塞 API 编程的时候就很容易引发回调地狱。好比：

task1(err, (callback) => {
    task2(err, (callbakck) => {
        task3(err, (callback) => {
            callback();
        });
    });
});
复制代码

传统的解决方案是进行任务的拆解方法就是把每一个任务拆开，经过抽象出一个迭代器，在任务队列中去顺序执行任务：

class TaskIterator {
    constructor(tasks, callback) {
        this.tasks = tasks;
        this.index = 0;
        this.callback = callback;
    }
    do() {
        if(this.index === this.tasks.length) {
            return this.finish();
        }
        const task = tasks[index];
        task(() => {
            this.index++;
            this.do();
        })
    }
    finish() {
        this.callback();
    }
}

const tasks = [task1, task2, task3];
const taskIterator = new TaskIterator(tasks, callback);

taskIterator.do();
复制代码

须要注意的是, 若是task()是一个同步操做的话，这样执行任务就可能变成一个递归算法，可能会有因为再也不每个循环中释放堆栈而达到调用栈最大限制的风险。

顺序迭代 模式是很是强大的，由于它能够适应好几种状况。例如，能够映射数组的值，能够将操做的结果传递给迭代中的下一个，以实现 reduce 算法，若是知足特定条件，能够提早退出循环，甚至能够迭代无线数量的元素。 ------ Node.js 设计模式（第二版）

值得注意的是，在 ES6 里，引入了promise以后也能够更简便地抽象出顺序迭代的模式：

const tasks = [task1, task2, task3];

const didTask = tasks.reduce((prev, task) =>{
    return prev.then(() => {
        return task();
    })
}, Promise.resolve());

didTask.then(() => {
    // do callback
});
复制代码

并行执行流的迭代模式

在一些状况下，一组异步任务的执行顺序并不重要，咱们须要的仅仅是任务完成的时候获得通知，就能够用并行执行流程来处理，例如：

-> task1
     /
start -> task2     (allEnd callback)
     \
      -> task3
复制代码

不做要求的时候，在 Node.js 环境下编程就可开始放飞自我：

class AsyncTaskIterator {
    constructor(tasks, callback) {
        this.tasks = tasks;
        this.callback = callback;
        this.done = 0;
    }
    do() {
        this.tasks.forEach(task => {
            task(() => {
                if(++this.done === this.tasks.length) {
                    this.finish();
                }
            });
        });
    }
    finish() {
        this.callback();
    }
}

const tasks = [task1, task2, task3];
const asyncTaskIterator = new AsyncTaskIterator(tasks, callback);

asyncTaskIterator.do();
复制代码

使用promise也就能够经过Promise.all()来接受全部的任务而且执行：

const tasks = [task1, task2, task3];

const didTask = tasks.map(task => task());

Promise.all(didTask).then(() => {
    callback();
})
复制代码

限制并行执行流的迭代模式

并行编程放飞自我天然是很爽，可是在不少状况下咱们须要对并行队列的数量作限制，从而减小资源消耗，好比咱们限制并行队列最大数为2：

-> task1 -> task2 
     /
start                      (allEnd callback)
     \
      -> task3
复制代码

这时候，就须要抽象出一个并行队列，在使用的时候对其实例化：

class TaskQueque {
    constructor(max) {
        this.max = max;
        this.running = 0;
        this.queue = [];
    }
    push(task) {
        this.queue.push(task);
        this.next();
    }
    next() {
        while(this.running < this.max && this.queue.length) {
            const task = this.queue.shift();
            task(() => {
                this.running--;
                this.next();
            });
            this.running++;
        }
    }
}

const tasks = [task1, task2, task3];
const taskQueue = new TaskQueue(2);

let done = 0, hasErrors = false;
tasks.forEach(task => { 
    taskQueue.push(() => {
        task((err) => {
            if(err) {
                hasErrors = true;
                return callback(err);
            }
            if(++done === tasks.length && !hasError) {
                callback();
            }
        });
    });
});
复制代码

而用promise的处理方式也与之类似：

class TaskQueque {
    constructor(max) {
        this.max = max;
        this.running = 0;
        this.queue = [];
    }
    push(task) {
        this.queue.push(task);
        this.next();
    }
    next() {
        while(this.running < this.max && this.queue.length) {
            const task = this.queue.shift();
            task.then(() => {
                this.running--;
                this.next();
            });
            this.running++;
        }
    }
}

const tasks = [task1, task2, task3];
const taskQueue = new TaskQueue(2);

const didTask = new Promise((resolve, reject) => {
    let done = 0, hasErrors = true;
    tasks.forEach(task => {
        taskQueue.push(() => {
            return task().then(() => {
                if(++done === task.length) {
                    resolve();
                }
            }).catch(err => {
                if(!hasErrors) {
                    hasErrors = true;
                    reject();
                }
            });
        });
    });
});

didTask.then(() => {
    callback();
}).then(err => {
    callback(err);
});
复制代码

同时暴露两种类型的 API

那么问题来了，若是我须要封装一个库，使用者须要在callback和promise中灵活切换怎么办呢？让别人一直本身切换就会显得很难用，因此就须要同时暴露callback和promise的 API ，让使用者传入callback的时候使用callback，没有传入的时候返回一个promise：

function asyncDemo(args, callback) {
    return new Promise((resolve, reject) => {
        precess.nextTick(() => {
            // do something 
            // 报错产出 err, 没有则产出 result
            if(err) {
                if(callback) {
                    callback(err);
                }
                return resolve(err);
            }
            if(callback){
                callback(null, result);
            }
            resolve(result);
        });
    });
}
复制代码