У меня есть файл CSV, который я читаю как поток, и использую преобразования для преобразования в JSON, а затем асинхронно сохраняю каждую строку в БД.

Проблема в том, что чтение из файла происходит быстро, что приводит к очень большому количеству одновременных асинхронных операций с БД, что приводит к остановке приложения.

Я бы хотел ограничить приложение таким образом, чтобы в любой момент времени выполнялось не более N незавершенных операций с БД.

Это основное ядро ​​моей функции _transform:

parser._transform = function(data, encoding, done) {
    //push data rows
    var tick = this._parseRow(data);

    //Store tick
    db.set(tick.date, tick, function(err, result) {
      console.log(result);
      if(err) throw err;
    });

    this.push(tick);
    done();
};

Я рассмотрел несколько вариантов, но они казались лучшими кандидатами:

Используйте async api forEachLimitПроблема, которую я вижу здесь, состоит в том, что в моем преобразовании потока я работаю только с одним объектом (строка из файла) при выполнении операций.Чтение всего файла невозможно из-за размераИспользуйте асинхронное параллельное параллельное решение с ограничением, как описано здесь, в разделе 7.2.3:http://book.mixu.net/node/ch7.htmlПроблема для меня в том, что делать в случае достижения «предела».Вращение или использование setTimeout, кажется, использует все запланированное время и предотвращает мои обратные вызовы БД, которые должны уменьшить инициируемый счетчик «выполнения».

Это были мои первые попытки «решения с параллельным доступом»:

var limit = 100;
var running = 0;

parser._transform = function(data, encoding, done) {
  //push data rows
  var tick = this._parseRow(data);

  this.push(tick);
  //Store tick to db
  if (running < limit) {
    console.log("limit not reached, scheduling set");
    running++;
    cb.set(tick.date, tick, function(err, result) {
      running--;
      console.log("running is:" + running);
      console.log(result);
      if(err) throw err;
    });
  } else {
    console.log("max limit reached, sleeping");
    setTimeout(this._transform(data, encoding, done),1000);
  }
  done();
};

Я только запустил node.js на этой неделе, поэтому мне не ясно, какова правильная модель для решения этой проблемы.

Замечания: Мне известно о нескольких вещах: это должно быть, по крайней мере, экспоненциальный откат, если используется последняя модель, и должна быть система «максимальных откатов», чтобы не разрушать стек вызовов. Попытался сделать это здесь пока просто.