Ограничение одновременных запросов HTTP-запросов в Node.js

Question

Я пишу сценарий узла для массового скачивания файлов. В примере это изображения из файла, в котором каждая строка имеет имя файла и URL-адрес. Я бы хотел, чтобы этот скрипт был масштабируемым до миллионов URL-адресов для загрузки.

Узел 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 

Answer 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')); 
    } 
});