前言

Nodejs其他系列文章
Nodejs高性能原理与异步非阻塞事件驱动模型与非异步API浅谈(有需要就更新)
简单使用Nodejs+Socket.io2.0+boostrap4.0实现聊天室功能

之前看过Nodejs异步模型,现在顺着讲一下异步编程方面的东西
以下全部代码可以看Asynchronous-programming

nodejs原生方法

我们以读取文件为例,读取文件名依赖前一个读取文件内容决定,nodejs提供的fs模块可以这么写

(以上代码可以看Asynchronous-programming的nodejs.js.)

正如我前面介绍过Nodejs高性能原理与异步非阻塞事件驱动模型与非异步API浅谈,Nodejs单綫程事件驱动、非阻塞式 I/O的强大性能优势,但是异步编程也会带来一些问题,例如:
1, 地狱回调;
2, 外层无法捕获错误;
3, 无法利用多线程优势;
4, 编写习惯可能不符合部分人群;

发佈订閲模式

nodejs里提供了一个单独模块events,大多数 Node.js 核心 API 都采用惯用的异步事件驱动架构，其中某些类型的对象（触发器）会周期性地触发命名事件来调用函数对象（监听器）。

例如，net.Server 对象会在每次有新连接时触发事件；fs.ReadStream 会在文件被打开时触发事件；流对象会在数据可读时触发事件。

所有能触发事件的对象都是 EventEmitter 类的实例。这些对象开放了一个 eventEmitter.on() 函数，允许将一个或多个函数绑定到会被对象触发的命名事件上。

异步与同步

EventEmitter 会按照监听器注册的顺序同步地调用所有监听器。所以需要确保事件的正确排序且避免竞争条件或逻辑错误。监听器函数可以使用 setImmediate() 或 process.nextTick() 方法切换到异步操作模式：

错误事件

EventEmitter 有时候也会发生错误,如果没有监听"error"事件的话会抛出错误、打印堆栈跟踪、且退出 Node.js 进程,所以应该始终注册一个监听器.

动态监听每次注册/移除事件

EventEmitter 实例会在一个监听器被添加到其内部监听器数组之前触发自身的 'newListener' 事件,在'newListener' 回调函数中, 如果该监听器的名字和已有监听器名称相同, 则在被插入到EventEmitter实例的内部监听器数组时, 该监听器会被添加到它同名监听器的前面。

你可以试试把一次性监once听换成普通监听on方法看什麽效果.

removeListener事件在 listener 被移除后触发,返回移除的eventName和listener

其余API都没什么好讲的可以直接查看文档.
Node.js v10.5.0 文档
 Node.js v10.5.0 Documentation

修改版

虽然不是很适合的例子,但是上面的原生代码可以类似这么改

const EventEmitter = require('events');
const fs = require("fs");

//创建类
class MyEmitter extends EventEmitter {}
//实例化
const myEmitter = new MyEmitter();
//监听
myEmitter.on('data1', (file) => {
    fs.readFile(file, 'utf8', function(err, data) {
        if (err) {
            return console.error(err);
        }
        console.log("第一次异步读取: " + data.toString());
        myEmitter.emit('data2', data.toString().trim());
    });
}).on('data2', (file) => {
    fs.readFile(file, 'utf8', function(err, data) {
        if (err) {
            return console.error(err);
        }
        console.log("第二次异步读取: " + data.toString());
        myEmitter.emit('data3', data.toString().trim());
    });
}).on('data3', (file) => {
    fs.readFile(file, 'utf8', function(err, data) {
        if (err) {
            return console.error(err);
        }
        console.log("第三次异步读取: " + data.toString());
        myEmitter.emit('data3', data.toString().trim());
    });
}).on('error', (err) => {
    console.error('有错误');
});

//触发
myEmitter.emit('data1', 'data1.txt');

(以上代码可以看Asynchronous-programming的EventEmitter.js.)

Promise && Async Await

以前因爲还没纳入规范所以大家都是通过各种工具库实现,但是自从ES6标准出来之后,大家普遍使用这种更加直观舒适的写法,不仅代码简洁,而且能够满足绝大部分需求.
我之前写过一篇相关文章,里面介绍的相当详细,而且有点长,不清楚的可以看一下.
Javascript关于异步编程的发展

EventProxy.js

EventProxy 仅仅是一个很轻量的工具，但是能够带来一种事件式编程的思维变化。有几个特点：

利用事件机制解耦复杂业务逻辑
移除被广为诟病的深度callback嵌套问题
将串行等待变成并行等待，提升多异步协作场景下的执行效率
友好的Error handling
无平台依赖，适合前后端，能用于浏览器和Node.js
兼容CMD，AMD以及CommonJS模块环境

EventProxy中文
 EventProxy英文

解决嵌套问题

有种和上面等价的快速创建all函数的方法

异步协作

1, 多类型异步协作
我们可以使用all方法将handler注册到事件组合上。当注册的多个事件都触发后，将会调用handler执行，每个事件传递的数据，将会依照事件名顺序，传入handler作为参数。

2, 重复异步协作
after方法适合重复的操作,将handler注册到N次相同事件的触发上。达到指定的触发数，handler将会被调用执行，每次触发的数据，将会按触发顺序，存为数组作为参数传入。

3, 持续型异步协作
tail方法在指定事件都触发之后，如果事件依旧持续触发，将会在每次触发时调用handler。

但是这些方式都不适合依赖型需求,所以不适用我那个依赖读取文件内容的代码.

异常处理

fail方法侦听了error事件，默认处理卸载掉所有handler，并调用回调函数。

done方法,在Node的最佳实践中，回调函数第一个参数一定会是一个error对象。检测到异常后，将会触发error事件。剩下的参数，将触发事件，传递给对应handler处理。done方法除了接受事件名外，还接受回调函数。如果是函数时，它将剔除第一个error对象(此时为null)后剩余的参数，传递给该回调函数作为参数。该回调函数无需考虑异常处理。

group方法,在after的回调函数中，结果顺序是与用户emit的顺序有关。为了满足返回数据按发起异步调用的顺序排列，EventProxy提供了group方法,当回调函数的数据还需要进行加工时，可以给group带上回调函数，只要在操作后将数据返回即可

异步事件触发: emitLater && doneLater

看不太懂没关系,我直接跟你说,尽管node的约定是所有的callback都是需要异步返回的，但是如果这个方法是由第三方提供的，我们没有办法保证db.check的callback一定会异步执行，如果db.check('key')在ep.once('check')监听之前执行,会无法执行后续操作,所以我们需要用异步事件触发

这样,就算db.check的回调函数被同步执行了，事件的触发也还是异步的，ep在当前事件循环中监听了所有的事件，之后的事件循环中才会去触发check事件。代码顺序将和逻辑顺序保持一致。
也可以像ep.done()一样通过doneLater来解决.

async.js

async.js是一个强大实用的异步库,提供多达70+API,包括一些常用的功能函数 (map, reduce, filter, each…)和流程控制函数(parallel, series, waterfall…),所有方法都假定你遵循nodejs约定:
1, 提供一个回调函数作为最后入参;
2, 错误作为返回值第一个参数;
3, 只执行一次回调;

因为本身太大,可以根据需要按需加载.就拿两个方法来说说

parallel(tasks, callback)

并行运行任务不需要等待顺序执行,一旦发生错误就作为参数传递执行回调函数,否则执行完之后全部结果传递给回调函数.
注意: parallel是并行启动I/O任务，而不是并行执行代码。如果您的任务不使用任何计时器或执行任何I/O，它们实际上将以串行方式执行。

waterfall(tasks, callback)

串行执行代码并把当前函数结果作为下一个函数的入参,如果发生错误会终止执行后续函数把错误作为参数传递给回调函数执行.

auto(tasks, concurrencyopt, callback)

根据任务的异步函数依赖需求决定最佳执行顺序,每个函数都能选择依赖其他先完成的函数并且满足条件之后立即执行.如果当中任何一个异步函数发生错误auto方法就会终止执行然后传递错误信息给回调函数执行.
有依赖性的话异步函数会接收包含已完成的函数结果作为第一个参数,否则只传递回调函数.

async.auto({
    // this function will just be passed a callback
    readData: async.apply(fs.readFile, 'data.txt', 'utf-8'),
    showData: [
        'readData',
        function(results, cb) {
            // results.readData is the file's contents
            // ...
        }
    ]
}, callback);

async.auto({
    get_data: function(callback) {
        console.log('in get_data');
        // async code to get some data
        callback(null, 'data', 'converted to array');
    },
    make_folder: function(callback) {
        console.log('in make_folder');
        // async code to create a directory to store a file in
        // this is run at the same time as getting the data
        callback(null, 'folder');
    },
    write_file: [
        'get_data',
        'make_folder',
        function(results, callback) {
            console.log('in write_file', JSON.stringify(results));
            // once there is some data and the directory exists,
            // write the data to a file in the directory
            callback(null, 'filename');
        }
    ],
    email_link: [
        'write_file',
        function(results, callback) {
            console.log('in email_link', JSON.stringify(results));
            // once the file is written let's email a link to it...
            // results.write_file contains the filename returned by write_file.
            callback(null, {
                'file': results.write_file,
                'email': 'user@example.com'
            });
        }
    ]
}, function(err, results) {
    console.log('err = ', err);
    console.log('results = ', results);
});

mapLimit(coll, limit, iteratee, callbackopt)

提供一个遍历任务集合进行并行操作,执行顺序不保证,但是返回结果不管集合是数组还是对象,统一返回任务顺序的结果数组.并且提供最大并发数,但是iteratee任何一次错误都会终止执行并且传输err给回调函数执行

修改版

我们可以使用waterfall或者auto方法修改

var waterfall = require('async/waterfall'),
    auto = require('async/auto'),
    fs = require('fs');

waterfall([
    function(callback) {
        fs.readFile('data1.txt', 'utf8', function(err, data) {
            if (err) {
                return console.error(err);
            }
            console.log("第一次异步读取: " + data.toString());
            callback(null, data.toString().trim());
        });
    },
    function(file, callback) {
        fs.readFile(file, 'utf8', function(err, data) {
            if (err) {
                return console.error(err);
            }
            console.log("第二次异步读取: " + data.toString());
            callback(null, data.toString().trim());
        });
    },
    function(file) {
        fs.readFile(file, 'utf8', function(err, data) {
            if (err) {
                return console.error(err);
            }
            console.log("第三次异步读取: " + data.toString());
        });
    }
], function(err, result) {
    // result now equals 'done'
});

//OR

auto({
    data1: function(callback) {
        fs.readFile('data1.txt', 'utf8', function(err, data) {
            if (err) {
                return console.error(err);
            }
            console.log("第一次异步读取: " + data.toString());
            callback(null, data.toString().trim());
        });
    },
    data2: [
        'data1',
        function(file, callback) {
            fs.readFile(file.data1, 'utf8', function(err, data) {
                if (err) {
                    return console.error(err);
                }
                console.log("第二次异步读取: " + data.toString());
                callback(null, data.toString().trim());
            });
        }
    ],
    data3: [
        'data2',
        function(file) {
            fs.readFile(file.data2, 'utf8', function(err, data) {
                if (err) {
                    return console.error(err);
                }
                console.log("第三次异步读取: " + data.toString());
            });
        }
    ]
}, function(err, result) {
    // result now equals 'done'
});

(以上代码可以看Asynchronous-programmingasync.js.)

异步并发控制

Bagpipe

Bagpipe通过push将调用传入内部队列。如果活跃调用小于最大并发数，将会被取出直接执行，反之则继续呆在队列中。当一个异步调用结束的时候，会从队列前取出调用执行。以此来保证异步调用的活跃量不高于限定值。
当队列的长度大于100或者大于最大并发数的2倍时，Bagpipe对象将会触发它的full事件，该事件传递队列长度值。

例如当你需要遍历文件目录的时候，异步可以确保充分利用IO。你可以轻松发起成千上万个文件的读取。但是，系统文件描述符是有限的。

我在书上看过Bagpipe可以设置额外参数控制行爲.例如:

refuse拒绝模式
timeout超时模式
但是我在文档API没找到!???所以不提了.

Nodejs异步编程方法选择

前言

nodejs原生方法