具有最大并发性的asynchronous并发队列

我遇到了一个自定义asynchronous队列，一次调用10个asynchronous函数的错误。

我开始有50个工作的队列，一旦前10个工作完成，队列移动到随后的10个，直到完成全部。

我遇到的错误是，一旦它完成了50，它将重新开始前5个工作，每次2或3或1个工作。 在队列结束时，还需要less于10个工作。

请创build这两个文件，并用摩卡testing，并亲自看到输出。

注意 ：将摩卡的超时设置为0可以使testing长时间运行。

Queue.js

function Queue(func, max) { this.jobs = []; this.func = func; this.max = max ? max : 10; } Queue.prototype.push = function(data) { var self = this; return new Promise(function(resolve, reject){ self.jobs.push({data: data, resolve: resolve, reject: reject}); if(!self.progress) { self.progress = true; self.run(); } }); }; Queue.prototype.run = function() { var self = this; var tasks = []; console.log("--------------------"); for(var i=0; i<this.jobs.length && i < this.max; i++) { tasks.push(this.jobs.shift()); console.log("queuing", tasks[tasks.length-1].data); } console.log("Total jobs queued", tasks.length); Promise.all( tasks.map(function(task){ return self.func(task.data) .then(task.resolve, task.reject); } )).then(this.next.bind(this)); }; Queue.prototype.next = function(){ if(this.jobs.length) { this.run(); } else { this.progress = false; } }; module.exports = Queue; 

QueueTest.js

 function async(data) { return new Promise(function(resolve, reject){ setTimeout(function(){ console.log("resolving", data); resolve(data); }, Math.random() * 5000); }); } it("should test queue", function(done){ var queue = new Queue(async); Promise.all( [0, 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, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50].map(queue.push.bind(queue)) ).then(function(){ done(); }); }); 

Answer 1

问题在于Queue.prototype.run中的for循环。

我不能立即明白为什么它应该行事不端，但修复是用self.jobs.splice()replacefor循环来创buildtasks数组。

 Queue.prototype.run = function() { console.log("--------------------"); var self = this; var tasks = self.jobs.splice(0, self.max); // <<<<<<<< this is the fix console.log("tasks:", tasks.map(obj => obj.data)); Promise.all( tasks.map(function(task){ return self.func(task.data) .then(task.resolve, task.reject); } )).then(this.next.bind(this)); }; 

没有别的需要改变。