Потоки в Node.js: Stream, pipeline и почему не readFile

Потоки — та часть Node, которую чаще всего обходят стороной: readFile короче и понятнее. Пока файлы маленькие, это и правда так. Проблема появляется, когда «маленький файл» однажды оказывается на 2 ГБ, а процесс падает с out of memory.

Разберём, что именно даёт поток, на замерах.

Что такое поток простыми словами

Поток — это способ обработать данные по частям, не загружая их в память целиком.

Аналогия: чтобы перелить воду из бочки в бочку, не нужен резервуар размером с бочку — достаточно шланга. readFile — это резервуар: сначала налить всё, потом отдать. Поток — шланг: пока одна порция уходит, следующая уже течёт.

Замер: readFile против потока

Возьмём файл на 183 МБ и прочитаем его двумя способами, измеряя реальную память процесса (RSS) с принудительной сборкой мусора перед замером.

import { readFile } from 'node:fs/promises';
import { createReadStream } from 'node:fs';

// вариант 1: целиком в память
const whole = await readFile(big, 'utf8');

// вариант 2: потоком, по кускам
let bytes = 0, chunks = 0, maxChunk = 0;
for await (const chunk of createReadStream(big)) {
  bytes += chunk.length;
  chunks += 1;
  if (chunk.length > maxChunk) maxChunk = chunk.length;
}

размер файла: 183 МБ
readFile: RSS 83 -> 252 МБ  (+169 МБ)
  строка в памяти целиком: 192000000 символов
поток   : RSS 252 -> 288 МБ  (+35 МБ)
  прочитано байт: 192000000 | кусков: 2930 | самый большой кусок: 65536 байт

Читаем результат честно:

• readFile вырос почти на размер файла — иначе и быть не может, всё содержимое лежит в одной строке.
• У потока живых данных никогда не больше 64 КБ: это видно по maxChunk = 65536. Прирост RSS в 35 МБ — это не данные, а память, которую аллокатор забрал у ОС и не вернул сразу.
• Ключевая разница не в 169 против 35, а в масштабировании: у потока потребление не зависит от размера файла. На файле в 10 ГБ кусок останется 64 КБ, а readFile просто упадёт.

Четыре типа потоков

Смысл в том, что они соединяются в конвейер: источник → преобразование → приёмник. Все четыре — наследники EventEmitter, поэтому у них есть on('data'), on('error') и on('end'); pipeline просто избавляет от необходимости подписываться руками.

pipeline вместо pipe

Забудьте pipe() — берите pipeline(). Причина в обработке ошибок: если в цепочке a.pipe(b).pipe(c) упадёт b, потоки a и c останутся открытыми, а ошибка потеряется. Это классическая утечка дескрипторов.

pipeline закрывает все потоки цепочки при любом сбое и отдаёт ошибку. А версия из node:stream/promises ещё и работает с await:

import { pipeline } from 'node:stream/promises';
import { createReadStream, createWriteStream } from 'node:fs';
import { createGzip } from 'node:zlib';

await pipeline(
  createReadStream('big.txt'),
  createGzip(),
  createWriteStream('big.txt.gz')
);

исходный : 192000000 байт
сжатый   : 558826 байт
сжатие   : 343.6x

Оговорка про 343x: наш тестовый файл — одна и та же строка, повторённая тысячи раз, а gzip живёт за счёт повторов. На реальных логах ждите 5–10x, на JSON — примерно столько же. Цифра здесь показывает не «какой хороший gzip», а то, что весь конвейер отработал, ни разу не подняв в память все 183 МБ.

Ошибка в конвейере превращается в обычное исключение:

try {
  await pipeline(
    Readable.from(['a']),
    new Writable({ write(c, e, cb) { cb(new Error('писать некуда')); } })
  );
} catch (err) {
  console.log('pipeline бросил:', err.message);
}

pipeline бросил: писать некуда

Свои Readable и Writable

Готовые потоки — не единственный вариант, свои создаются просто. Readable.from() делает поток из любого итерируемого:

import { Readable, Writable } from 'node:stream';

const src = Readable.from(['раз', 'два', 'три']);

const collected = [];
const dst = new Writable({
  write(chunk, encoding, callback) {
    collected.push(chunk.toString());
    callback();               // сигнал «готов к следующему куску»
  }
});

await pipeline(src, dst);
console.log('собрано:', collected.join(','));

собрано: раз,два,три

Вызов callback() здесь обязателен: именно им поток сообщает, что готов принимать дальше. Забыли — конвейер встанет навсегда и никакой ошибки не будет.

Backpressure: зачем всё это нужно

Главное, что даёт поток помимо памяти, — обратное давление. Если приёмник медленнее источника (диск быстрее сети), поток сам притормаживает чтение, пока приёмник разбирает очередь.

Именно это ломается, когда пишут «вручную»:

// ❌ игнорирует backpressure: читаем как можем, пишем куда придётся
readable.on('data', (chunk) => writable.write(chunk));

// ✅ pipeline сам согласует скорости
await pipeline(readable, writable);

В первом варианте при медленном приёмнике данные копятся во внутреннем буфере — и память растёт ровно так же, как при readFile, только неожиданно.

highWaterMark: размер куска настраивается

64 КБ — это highWaterMark по умолчанию, а не константа. Его можно менять:

createReadStream(path, { highWaterMark: 256 * 1024 });  // куски по 256 КБ

Размен простой: больше кусок — меньше системных вызовов, но выше расход памяти. На быстром SSD и крупных файлах увеличение даёт выигрыш; на мелких файлах — только лишний расход. Трогать значение стоит, когда вы упёрлись в производительность и замерили, а не «на всякий случай»: дефолт выбран разумно.

Отдельно живёт объектный режим. Обычные потоки гоняют байты, но если поставить objectMode: true, через поток пойдут любые значения — объекты, строки, что угодно:

const src = Readable.from([{ id: 1 }, { id: 2 }]);   // objectMode включается сам

Здесь highWaterMark считается уже не в байтах, а в штуках объектов (по умолчанию 16). Это рабочий способ построить конвейер обработки записей: читать CSV построчно, преобразовывать и писать в базу, не держа в памяти весь файл.

Что было на этой странице в 2018 году

Первая версия вышла 8 июля 2018 года — тогда потоки описывали через pipe(), потому что stream/promises ещё не существовало: он появился только в Node 15, осенью 2020-го. То есть совет «берите pipeline» физически не мог быть дан.

Что изменилось за эти годы:

Сам механизм при этом не поменялся ни на байт: те же четыре типа, тот же backpressure, тот же highWaterMark в 64 КБ. Поменялся только способ их соединять — и именно он снял класс ошибок с утечкой дескрипторов.

Когда поток не нужен

Честно: в большинстве задач. Конфиг, JSON на пару сотен килобайт, шаблон письма — читайте readFile, код будет короче и понятнее, а разницы не будет никакой.

Поток оправдан, когда:

• размер данных заранее неизвестен или потенциально велик (загрузки пользователей, логи, выгрузки);
• данные можно обрабатывать по мере поступления, не дожидаясь конца (разбор CSV построчно);
• вы соединяете источник и приёмник напрямую — например, отдаёте файл в HTTP-ответ;
• вывод внешней программы нужно читать на лету, не копя в памяти, — так работает spawn из child_process.

Ещё один ориентир — жёсткий предел: readFile с кодировкой физически не прочитает файл больше ~512 МБ. У потока такого потолка нет вообще.

Остальные разборы платформы собраны в справочнике по Node.js.