在 Node.js 中限制同时传输的 HTTP 请求答案

【问题标题】：Limiting simultaneous piped HTTP requests in Node.js在 Node.js 中限制同时传输的 HTTP 请求
【发布时间】：2015-01-27 21:34:23
【问题描述】：

我正在编写一个节点脚本来批量下载文件。在示例中，这些是来自文件的图像，其中每一行都有一个文件名和一个 URL。我希望这个脚本可以扩展到数百万个 URL 来下载。

Node JS 流似乎是一个很好的方法，因为我们可以通过管道输入 URL 列表，http 请求 URL，并将响应写入文件。

此脚本使我的计算机崩溃，并丢失了空白 jpg 文件。 pipe() 方法似乎没有处理背压；似乎该脚本正在立即并行请求所有 URL。如何限制同时 HTTP 请求的数量，以便在扩展以下载大量 URL 时它可以一致地运行？谢谢。

'use strict';
var fs      = require('fs'),
    request = require('request'),
    through = require('through'),
    split   = require('split'),
    urlList = 'https://gist.githubusercontent.com/phelma/e1558aeb181c0cfe47b8/raw/cc5e667277308fda408f6af1404bc2d322b5186c/images.txt';
    // 10000 images

var splitByTab = through(function(buf) {
    var item = buf.toString().split('\t');
    this.queue(item);
});

var downloadStream = through(function(item) {
    // item is array [ filename , URL ]
    if (item[1]) {
        console.log('Requesting ' + item[1]);
        request
            .get(item[1])
            .on('error', function(err) {
                console.log('\nError: ' + err.message + '\n' + item[1]);
            })
            .pipe(fs.createWriteStream(__dirname + '/out/' + item[0] + '.jpg'));
    }
});

request
    .get(urlList) // Request the
    .pipe(split()) // Split file into rows
    .pipe(splitByTab) // Split each row into a array items
    .pipe(downloadStream); // Download each item

【问题讨论】：

标签： javascript node.js node.js-stream

【解决方案1】：

through 中的流控制是使用 pause/resume 完成的。

var requestLimit = 10, activeRequests = 0;
var downloadStream = through(function(item) {
    // item is array [ filename , URL ]
    if (item[1]) {
        if (activeRequests++ > requestLimit)
            this.pause();
        console.log('Requesting ' + item[1]);
        request
            .get(item[1])
            .on('error', function(err) {
                console.log('\nError: ' + err.message + '\n' + item[1]);
                if (--activeRequests <= requestLimit)
                    this.resume();                    
            })
            .on('response', function(e) { 
                if (--activeRequests <= requestLimit)
                    this.resume();
            })
            .pipe(fs.createWriteStream(__dirname + '/out/' + item[0] + '.jpg'));
    }
});

【讨论】：