具有最大并发性的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(); }); });
问题在于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)); };
没有别的需要改变。