Nodejs-异步并发控制

异步并发控制

在 node 中可以利用异步发起并行调用。但是如果并发量过大，就会导致下层服务器吃不消。

bagpipe 解决方案

解决方案

通过一个队列来控制并发量
如果当前活跃的异步调用小于限定值，从队列中取出执行
如果活跃调用达到限定值，调用暂时放到队列中
每个异步调用结束的时候，从队列中取出新的异步调用执行

var Bagpipe = require("bagpipe");
var bagpipe = new Bagpipe(10);
for (var i = 0; i < 100; i++) {bagpipe.push(async, function () {// 异步回调执行});
}
bagpipe.on("full", function (lenght) {console.warn("底层系统不能及时完成，队列拥堵");
});

核心实现

Bagpipe.prototype.push = function (method) {var args = [].slice.call(arguments, 1);var callback = args[args.length - 1];if (typeof callback !== "function") {args.push(function () {});}if (this.options.disabled || this.limit < 1) {method.apply(null, args);return this;}if (this.queue.length < this.queueLength || !this.options.refuse) {this.queue.push({method: method,args: args,});} else {var err = new Error("too much async call in queue");err.name = "TooMuchAsyncCallError";callback(err);}if (this.queue.length > 1) {this.emit("full", this.queue.lenght);}this.next();return this;
};

next()方法主要是用来判断活跃调用的数量，如果正常，使用内部方法 run 来执行真正的调用。

Bagpipe.prototype.next = function () {var that = this;if (that.active < that.limit && this.queue.length) {var req = this.queue.shift();this.run(req.method, req.args);}
};

Bagpine.prototype.run = function (method, args) {var that = this;that.active++;var callback = args[args.length - 1];var timer = null;var called = false;args[args.length - 1] = function (err) {if (timer) {clearTimeout(timer);timer = null;}if (!called) {this._next();callback.apply(null, arguments);} else {if (err) {that.emit("outdated", err);}}};var timeout = that.options.timeout;if (timeout) {timer = setTimeout(function () {called = true;that._next();// pass the exceptionvar err = new Error(timeout + "ms timeout");err.name = "BagpipeTimeoutError";err.data = {name: method.name,method: method.toString(),args: args.slice(0, -1),};callback(err);}, timeout);}method.apply(null, args);
};

拒绝模式
对于大量的异步调用，会分场景进行区分。设计到并发控制，会造成部分等待，如果调用由实时方面的需求，需要快速返回。这种情境下需要快速失败，让调用方竟早返回。
超时控制
超时控制是为异步调用设置一个时间阈值，如果异步调用没有在规定的时间内完成，先执行用户传入的回调函数。

async解决方案

paralleLimit()和parallel类似，多了一个限制并发数量的参数，使得任务只能同时并发一定数量，不是无限制并发。

async.parallelLimit([function(callback) {fs.readFile('file1.txt','utf-8',callback)},function(callback) {fs.readFile('file2.txt','utf-8',callback)}
], 1, function(err, results) {// todo
})

paralleLimit()方法不能动态的增加并行任务。async提供了queue()方法来满足需求。

var q = async.queue(function(file,callback) {fs.readFile(file, 'utf-8', callback);
},2)
q.drain = function () {};
fs.readdirSync('.').forEach(function (file) {q.push(file, function (err, data) {})
})