Streams в Node.js для больших файлов и контроль обратного давления вместо чтения файла целиком в память

Ночная задача выглядит просто: взять выгрузку в пять гигабайт, распаковать, посчитать строки и записать отчёт. На тестовом файле в 50 мегабайт скрипт из двадцати строк справляется за секунду. Ночью вместо отчёта приходит сообщение heap out of memory, контейнер перезапускают, история повторяется. Классический сюжет о том, как автор скрипта продумал всё, кроме скорости, с которой память уходит под буферы. Ниже разбор стримов Node.js: как устроено обратное давление, почему попытка прочитать большой файл целиком ломается предсказуемым образом и когда простой вариант всё же разумен.

Что реально происходит при чтении файла в пять гигабайт через fs.readFile

Метод fs.readFile честно делает то, что обещает: читает файл от первого байта до последнего и собирает всё в один Buffer. Для файла в пять гигабайт это пять гигабайт непрерывной памяти ещё до начала обработки. Следующий шаг обычно удваивает счёт: вызов toString строит поверх буфера строку, где каждый символ занимает два байта в кодировке UTF-16. Если дальше стоит JSON.parse, поверх строки вырастает ещё один слой из объектов и массивов. Один и тот же файл оказывается в памяти трижды, в трёх разных видах.

У этой конструкции есть и жёсткий потолок. Предельная длина строки в V8 измеряется примерно полумиллиардом символов, файл в пять гигабайт из однобайтовых символов его гарантированно превышает. Лимит старой памяти V8 по умолчанию измеряется единицами гигабайт и меняется от версии к версии, честную картину даёт сообщение heap out of memory в логах, а флаг --max-old-space-size управляет лимитом вручную. Буферы живут вне кучи V8, в так называемой внешней памяти, но полностью учитываются в RSS процесса, а значит и в лимите контейнера. ОЗУ не становится больше от того, что аллокации формально не в куче.

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

Отдельная ловушка ждёт авторов веб-обработчиков. Привычный await fs.promises.readFile внутри обработчика запроса выглядит невинно, пока кто-нибудь не загружает файл побольше. Сервер с несколькими одновременными обращениями умножает потолок на их число, и даже гигабайт ОЗУ заканчивается на третьем параллельном запросе. Встроенная утилита fs.copyFile в ловушке не участвует: она опирается на системные вызовы копирования и не тащит содержимое в память JavaScript вовсе.

Механика обратного давления от highWaterMark до события drain

Стримы решают ту же задачу иначе: файл идёт порциями. Размер порции задаётся параметром highWaterMark, это планка внутреннего буфера потока. Для файловых стримов по умолчанию действует планка в 64 килобайта, для стримов общего вида 16 килобайт, для объектного режима 16 объектов. Файл размером в 5 гибибайт при планке 64 килобайта распадается примерно на 82 тысячи порций, и каждая из них живёт в памяти считанные миллисекунды.

Обратное давление это протокол разговора между быстрым читателем и медленным писателем. Метод write возвращает false, когда внутренний буфер потока назначения достиг планки. Источник обязан в этот момент остановиться и ждать события drain, которое сообщает, что буфер освободился и можно продолжать. Если источник игнорирует сигнал, порции продолжают приезжать и складываться в очередь внутри принимающего потока, а очередь живёт в памяти процесса. Ровно это и происходит с наивным кодом: backpressure работает только для тех, кто его слушает.

У читаемого потока есть два режима. В режиме паузы порции выдаются строго по запросу через метод read, и темп задаёт потребитель. В режиме течения, который включается обработчиком события data или вызовом resume, порции идут потоком без запроса, и темп обязан удерживать тот, кто принимает. Функция pipe и асинхронный итератор переключают режимы сами, поэтому правильный код по виду почти не отличается от обычного цикла.

Странный на первый взгляд параметр highWaterMark на деле просто бюджет задержки. Небольшая планка даёт быстрые сигналы и экономную память, крупная сокращает накладные расходы на каждый chunk и ускоряет дисковые копии. Поднятие планки до одного мегабайта уменьшает количество порций для того же файла с 82 тысяч до пяти тысяч, что заметно на сплошном копировании больших файлов.

Почему pipe и pipeline снимают с разработчика ручное управление потоком

Метод pipe собирает весь протокол из предыдущего раздела в одну строку. Он ставит читателя на паузу, когда write возвращает false, и возобновляет по событию drain, без единой строчки ручного кода. Для типовой цепочки вида чтение, распаковка, запись этого достаточно.

У pipe есть известная слабость: он не пересылает ошибки. Если распаковка падает на середине файла, писатель останавливается, читатель может продолжать тянуть байты, а недописанный результат остаётся лежать на диске как будто полноценный. Утилита pipeline, появившаяся в Node 10, решает вопрос одним махом: при первой же ошибке она уничтожает все потоки цепочки и отдаёт ошибку в колбэк или в отклонённый промис. Промис-версия живёт в модуле stream/promises:

const { pipeline } = require("node:stream/promises");
const fs = require("node:fs");
const zlib = require("node:zlib");

async function repack(src, dest) {
  await pipeline(
    fs.createReadStream(src),
    zlib.createGunzip(),
    zlib.createGzip({ level: 9 }),
    fs.createWriteStream(dest)
  );
}

Обратное давление в цепочке работает на каждом стыке: медленное звено сигнализирует вверх по цепочке, и вся линия замедляется до скорости самого медленного участка. Ошибки из любого звена доходят до вызывающего кода, временные файлы прибираются.

Для тонкой зачистки у каждого потока есть метод destroy: он останавливает поток и оповещает соседние звенья. pipeline вызывает его для всей цепочки автоматически, вручную он нужен в кастомных склейках.

Три типовые ошибки стриминга больших файлов и их видимые симптомы

Даже у знающих про стримы разработчиков находятся способы обнулить их пользу. Три самые частые ошибки выглядят так:

1. Собирать все порции в массив, чтобы склеить их в одну строку после конца чтения;
2. Подписаться на событие data и писать в другой поток, не проверяя результат метода write;
3. Запускать асинхронную обработку каждой порции параллельно без ограничения очереди.

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

Вторая ошибка требует пояснения. Обработчик события data переводит поток в режим течения, и порции начинают приезжать без запроса. Пока обработка быстрая, ничего страшного не происходит. Как только соседний поток замедляется, его буфер растёт, а вместе с ним растёт память. Классическое место такой ошибки, копирование в медленное хранилище или в сетевой сокет, где скорость плавает в разы в зависимости от удалённого конца.

Третья ошибка пришла вместе с async и await. Код запускает Promise.all для каждой порции, параллельность растёт, а обратное давление размыто: формально вызовы write проверяются, но очередь промисов между чтением и записью растёт без ограничений. Лечится ограничением параллельности или семафором на несколько одновременных задач.

Рабочий код с ручным контролем давления через событие drain

Ручной вариант нужен там, где pipeline не подходит: кастомный приёмник, вставки между звеньями, особая обработка ошибок. Каркас умещается в несколько строк:

const { once } = require("node:events");
const fs = require("node:fs");

async function copyWithPause(src, dest) {
  const input = fs.createReadStream(src);
  const output = fs.createWriteStream(dest);
  for await (const chunk of input) {
    if (!output.write(chunk)) {
      await once(output, "drain");
    }
  }
  output.end();
  await once(output, "finish");
}

Асинхронный итератор по читаемому потоку тянет порции по одной и естественно останавливается на await, так что пауза приходит не от таймера, а от самого протокола. Вызов end закрывает сторону записи, событие finish подтверждает, что всё записано. Любая асинхронная работа между получением порции и вызовом write встраивается в ту же схему: пока обработка порции не закончена, новые порции не запрашиваются.

Свой узел цепочки пишется как подкласс Transform, и он встраивается в ту же схему без ручного управления:

const { Transform } = require("node:stream");

class RowCounter extends Transform {
  constructor() {
    super({ readableObjectMode: true });
    this.rows = 0;
    this.tail = "";
  }
  _transform(chunk, enc, callback) {
    const text = this.tail + chunk.toString("utf8");
    const parts = text.split("\n");
    this.tail = parts.pop();
    this.rows += parts.length;
    callback(null, { rows: this.rows });
  }
  _flush(callback) {
    if (this.tail) this.rows += 1;
    callback(null, { rows: this.rows });
  }
}

Хвост this.tail здесь не украшение: граница порции может разрезать строку пополам, и наивный подсчёт по каждой порции отдельно даёт счёт, завышенный на количество таких разрезов. Именно эта мелочь отличает код, который гоняли на больших файлах, от кода, который писали по памяти.

Для построчной обработки поверх стрима есть и готовый модуль readline. Интерфейс создаётся поверх читаемого потока, строка приходит только когда встречен перевод строки, и память остаётся плоской независимо от длины файла:

const rl = readline.createInterface({ input: fs.createReadStream(src) });
let rows = 0;
rl.on("line", () => { rows += 1; });
await once(rl, "close");

Тот же принцип работает для разбора логов, импорта CSV и загрузки в базу: каждая строка обрабатывается и забывается.

Реальные цифры памяти и скорости при потоковой обработке больших файлов

Потоковая обработка ограничивает память конструктивно, а не удачей. Бюджет складывается из буферов звеньев цепочки и рабочей структуры: при планках 64 килобайта это единицы мегабайт, при мегабайтных планках десятки. Файл размером в гигабайты, десятки гигабайт или терабайт не меняет потолок ни на байт, меняется только время обработки. Контроль простой: process.memoryUsage показывает rss, heapUsed и external, и на честном стриминге все три величины стоят на месте от начала файла до конца. Диагностика не требует сложных инструментов: флаг --trace-gc печатает каждую сборку мусора с паузой и причиной, node --inspect открывает снапшоты кучи в инструментах разработчика, а периодический вывод памяти по интервалу сразу показывает картину. Кривая, ползущая вверх линейно от начала файла, это готовый диагноз: где-то в цепочке копятся порции или строки.

Накладные расходы на порцию реальны: событие, аллокация, вызовы. Отсюда практическое правило, крупные планки для сплошных дисковых копий, стандартные для сетевых и трансформирующих цепочек. Файловые операции Node.js ходят через пул поток libuv, по умолчанию из четырёх потоков, так что параллельные дисковые задачи делят между собой одну и ту же четвёрку. Переменная окружения UV_THREADPOOL_SIZE расширяет пул, когда дисковая подсистема способна переварить больше параллелизма.

Скорость обработки при переходе от чтения целиком к стримам обычно не страдает, а растёт: процессор работает вровень с диском, а не ждёт конца чтения. Выигрыш в памяти не процентный, а порядковый: вместо величины, пропорциональной файлу, память перестаёт зависеть от размера вовсе.

Когда чтение файла целиком остаётся разумным выбором

Стримы не отменяют простые случаи. Конфиги, манифесты, небольшие JSON документы, файлы в десятки мегабайт читаются через fs.readFile без малейшего риска и без лишнего кода. Случайный доступ по смещению стримами только осложняется: для двоичных форматов с заголовками честнее выглядит чтение через модуль fs с явными позициями. Локальные скрипты и тесты тоже не заслуживают асинхронных итераторов ради трёх строчек.

Разумная граница проходит по простому вопросу: может ли файл вырасти. Выгрузка, которая сегодня весит 40 мегабайт, через два года способна весить 40 гигабайт, а код останется тем же. Всё, что растёт, обрабатывается потоками с честным отношением к обратному давлению, всё, что ограничено по самой своей природе, читается целиком без угрызений. Пять минут на for await и once окупаются в первую же ночь, когда отчёт приходит вовремя, а не в виде сообщения о законченной памяти.