限制Node.js中的异步调用

Question

我有一个Node.js应用程序,它在本地获取文件列表并将它们上传到服务器.此列表可能包含数千个文件.

for (var i = 0; i < files.length; i++) {
   upload_file(files[i]);
}

如果我用数千个文件执行此操作,upload_file将同时被调用数千次,并且很可能死亡(或者至少是挣扎).在同步世界中,我们创建一个线程池并将其限制为一定数量的线程.是否有一种简单的方法可以限制一次执行多少次异步调用？

Answer 1

像往常一样,我推荐Caolan McMahon的异步模块.

让你的upload_file函数作为第二个参数进行回调:

var async = require("async");

function upload_file(file, callback) {
    // Do funky stuff with file
    callback();
}

var queue = async.queue(upload_file, 10); // Run ten simultaneous uploads

queue.drain = function() {
    console.log("All files are uploaded");
};

// Queue your files for upload
queue.push(files);

queue.concurrency = 20; // Increase to twenty simultaneous uploads

Answer 2

上面的答案,重新:NPM上的异步是最好的答案,但是如果你想了解更多有关控制流的信息:

您应该研究控制流模式.在Mixu的Node Book的第7章中对控制流模式进行了精彩的讨论.也就是说,我将看一下7.2.3中的例子:有限并行 - 一个异步,并行,并发限制的循环.

我改编了他的例子:

function doUpload() {
    // perform file read & upload here...
}

var files   = [...];
var limit   = 10;       // concurrent read / upload limit
var running = 0;        // number of running async file operations

function uploader() {
    while(running < limit && files.length > 0) {
        var file = files.shift();
        doUpload(file, function() {
            running--;
            if(files.length > 0)
                uploader();
        });
        running++;
    }
}

uploader();

Answer 3

你应该尝试排队.我假设upload_file()完成时会触发回调.像这样的东西应该做的伎俩(未经测试):

function upload_files(files, maxSimultaneousUploads, callback) {
    var runningUploads = 0,
        startedUploads = 0,
        finishedUploads = 0;

    function next() {
        runningUploads--;
        finishedUploads++;

        if (finishedUploads == files.length) {
            callback();
        } else {
            // Make sure that we are running at the maximum capacity.
            queue();
        }
    }

    function queue() {
        // Run as many uploads as possible while not exceeding the given limit.
        while (startedUploads < files.length && runningUploads < maxSimultaneousUploads) {
            runningUploads++;
            upload_file(files[startedUploads++], next);
        }
    }

    // Start the upload!
    queue();
}