Стримы в Node.js: рабочие приёмы для потоковой обработки

Задача: обработать CSV на 4 ГБ на сервере с 512 МБ памяти под процесс. Наивный fs.readFileSync падает с JavaScript heap out of memory ещё до первой строки. Стримы решают это тем, что тянут данные кусками (chunk) и держат в памяти только буфер, а не весь файл. Но именно на стримах чаще всего теряют ошибки, забывают про backpressure и получают процессы, которые едят память несмотря на «потоковую» обработку.

Всё ниже — для Node.js 18+ и 20+ (LTS). Ключевые вещи вроде stream.pipeline и промис-обёрток стабильны с Node 15, а stream/promises — с Node 15+.

Почему pipe() — не то, что вам нужно

Классический приём выглядит красиво:

const fs = require('fs');
const zlib = require('zlib');

fs.createReadStream('input.txt')
  .pipe(zlib.createGzip())
  .pipe(fs.createWriteStream('output.txt.gz'));

Проблема одна, но серьёзная: pipe() не пробрасывает ошибки по цепочке и не закрывает промежуточные стримы при сбое. Если createGzip упадёт, читающий и пишущий стримы останутся висеть — это утечка дескрипторов файлов. Обрабатывать error вручную на каждом звене можно, но легко забыть одно, и тогда ошибка уронит процесс через unhandledError.

Правило простое: pipe() — для интерактивных случаев вроде проксирования HTTP-ответа, где вы контролируете обе стороны. Для обработки данных используйте pipeline, который сам чистит за собой всю цепочку при любой ошибке.

pipeline вместо ручной обвязки

Тот же код через stream/promises — с автоматической очисткой и нормальным try/catch:

const fs = require('fs');
const zlib = require('zlib');
const { pipeline } = require('stream/promises');

async function gzipFile(src, dest) {
  await pipeline(
    fs.createReadStream(src),
    zlib.createGzip(),
    fs.createWriteStream(dest)
  );
}

gzipFile('input.txt', 'output.txt.gz')
  .then(() => console.log('готово'))
  .catch((err) => console.error('сбой:', err.message));

Если любое звено выдаст ошибку, pipeline уничтожит все стримы в цепочке и отклонит промис. Дескрипторы не текут, ошибка приходит в один catch.

Backpressure: что это и почему память всё равно течёт

Backpressure — механизм, при котором быстрый источник притормаживает, пока медленный приёмник не разгрузит буфер.¹ Когда вы пишете в стрим через write(), метод возвращает false, если внутренний буфер переполнен. Если вы игнорируете этот сигнал и продолжаете писать, данные копятся в памяти — и «потоковая» обработка превращается в тот же OutOfMemory, только медленнее.

pipe() и pipeline соблюдают backpressure автоматически. Ручной цикл — нет. Вот где ломается типичный самописный код:

// НЕПРАВИЛЬНО: игнорируем возврат write()
readable.on('data', (chunk) => {
  const processed = transform(chunk);
  writable.write(processed); // при false буфер растёт бесконтрольно
});

Правильный ручной вариант учитывает false и ждёт события drain:

readable.on('data', (chunk) => {
  const ok = writable.write(transform(chunk));
  if (!ok) {
    readable.pause();
    writable.once('drain', () => readable.resume());
  }
});

Но писать это руками почти никогда не нужно. Async-итераторы (Node 10+, стабильны с 12) делают то же самое читаемо:

const fs = require('fs');

async function countLines(path) {
  let count = 0;
  const stream = fs.createReadStream(path, { encoding: 'utf8' });
  for await (const chunk of stream) {
    // for await сам приостанавливает чтение, пока тело цикла занято
    count += chunk.split('\n').length - 1;
  }
  return count;
}

Здесь backpressure работает бесплатно: цикл не запросит следующий chunk, пока не выполнит тело. Никаких pause/resume.

Transform-стримы для построчной обработки

CSV и логи удобно обрабатывать построчно, а не по случайным границам chunk (chunk может обрываться на середине строки). Для разбиения на строки есть встроенный readline, а для произвольной трансформации — Transform.

readline для построчного чтения

const fs = require('fs');
const readline = require('readline');

async function processCsv(path) {
  const rl = readline.createInterface({
    input: fs.createReadStream(path),
    crlfDelay: Infinity, // корректно склеивает \r\n
  });
  let rows = 0;
  for await (const line of rl) {
    const cols = line.split(',');
    rows++;
  }
  return rows;
}

readline сам собирает chunk в строки и поддерживает async-итерацию — память под контролем даже на 4 ГБ.

Свой Transform в объектном режиме

Когда нужно пропустить поток объектов через несколько этапов, используйте objectMode: true. В этом режиме через стрим ходят не буферы, а любые JS-значения:

const { Transform } = require('stream');

const toUpperName = new Transform({
  objectMode: true,
  transform(record, encoding, callback) {
    // record — объект, а не Buffer
    record.name = String(record.name).toUpperCase();
    callback(null, record); // первый аргумент — ошибка или null
  },
});

Важная деталь: если внутри transform вы ловите ошибку, передайте её первым аргументом в callback(err), а не бросайте через throw. Брошенное из асинхронного контекста исключение pipeline не поймает.

Сравнение подходов

Если не получилось

• Error [ERR_STREAM_PREMATURE_CLOSE] — цепочка закрылась раньше, чем стрим дочитал данные. Частая причина: приёмник (например HTTP-ответ) закрылся, а вы продолжаете писать. Используйте pipeline, он корректно завершит остальные звенья.
• MaxListenersExceededWarning — вешаете обработчики в цикле на один стрим. Проверьте, не добавляете ли once('drain') без снятия внутри data.
• Память растёт, хотя используете стримы — где-то игнорируется backpressure. Ищите write() в цикле без проверки возврата или накопление chunk в массив «на потом».
• callback called multiple times в Transform — вызвали callback дважды в одном transform. Убедитесь, что после callback(null, data) нет второго вызова в ветке с ошибкой.
• Ошибка из Transform не приходит в catch — вы бросили throw вместо callback(err). Асинхронные исключения из тела transform не перехватываются pipeline.

Что делать дальше

Соберите минимальную цепочку createReadStream → Transform → createWriteStream через pipeline и прогоните на файле заведомо больше вашего лимита памяти — так вы проверите, что backpressure реально работает, а не только в теории. В проде смотрите на process.memoryUsage().heapUsed под нагрузкой: если он растёт линейно с объёмом данных, где-то потерян backpressure. Инвалидация и параллельная обработка нескольких стримов — отдельная тема; здесь мы разбирали только корректную одиночную цепочку. Для отмены долгих операций изучите передачу AbortSignal в pipeline (поддержка с Node 15+).