Потоки и буферы в Node.js

Поток (stream) обрабатывает данные по кускам, а не целиком. Файл на 5 ГБ можно прочитать через readFile, но он весь окажется в памяти. Поток читает его кусками по 64 КБ: память занята чуть-чуть, а первые байты обрабатываются сразу.

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

• Readable: источник данных. fs.createReadStream, тело входящего HTTP-запроса req, process.stdin.
• Writable: приёмник. fs.createWriteStream, HTTP-ответ res, process.stdout.
• Duplex: и то и другое, независимо. TCP-сокет.
• Transform: Duplex, где выход вычисляется из входа. zlib.createGzip(), шифрование.

Все потоки наследуют EventEmitter: data, end, error, finish, close.

pipeline вместо pipe

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

await pipeline(
  fs.createReadStream('access.log'),
  zlib.createGzip(),
  fs.createWriteStream('access.log.gz'),
);

a.pipe(b) тоже соединяет потоки и управляет скоростью, но не передаёт ошибки дальше и не закрывает остальные потоки при сбое: остаются утечки дескрипторов. pipeline закрывает всю цепочку при ошибке и сообщает о ней одним промисом. В новом коде используют его.

Backpressure

Если источник быстрее приёмника (диск читает быстрее, чем клиент скачивает), данные копятся в памяти. Защита называется backpressure:

• writable.write(chunk) возвращает false, когда внутренний буфер превысил highWaterMark;
• писатель должен остановиться и ждать события drain.

for (const line of lines) {
  if (!out.write(line)) {
    await once(out, 'drain'); // once из node:events
  }
}
out.end();

pipe и pipeline делают это сами. Ошибка новичка: писать в цикле, игнорируя false, и получить рост памяти до падения процесса.

highWaterMark в Node 22 по умолчанию 64 КБ для байтовых потоков и 16 объектов в объектном режиме (objectMode: true).

Чтение потока

Удобнее всего асинхронной итерацией:

for await (const chunk of fs.createReadStream('big.csv')) {
  process(chunk);
}

Readable бывает в двух режимах: paused (данные забирают вызовом read()) и flowing (данные сами идут через data). Подписка на data, pipe или resume() включают flowing.

Свой Transform

const { Transform } = require('node:stream');
const upper = new Transform({
  transform(chunk, encoding, callback) {
    callback(null, chunk.toString().toUpperCase());
  },
});

Buffer

Buffer это кусок байтов вне кучи V8, подкласс Uint8Array. Потоки по умолчанию отдают именно буферы.

• Buffer.from('привет') занимает 12 байт: кириллица в UTF-8 по 2 байта на символ. Длина строки 6, длина буфера 12.
• Buffer.alloc(n) выделяет память и заполняет нулями. Buffer.allocUnsafe(n) быстрее, но в памяти может лежать мусор из старых данных: показывать такой буфер наружу нельзя, пока не заполнишь.
• buf.subarray() не копирует, а смотрит на ту же память.
• Склеивая куски текста из потока, не делай str += chunk для многобайтовых кодировок: символ может разорваться между чанками. Собирай буферы и делай Buffer.concat(chunks).toString(), или вызови setEncoding('utf8').

Частые ошибки

• readFile для огромных файлов вместо потока.
• pipe без обработки ошибок.
• Игнорировать false от write.
• Думать, что string.length это размер в байтах.

Что спросят на собесе

• Какие бывают потоки и примеры каждого.
• Что такое backpressure и как Node с ним справляется.
• Чем pipeline лучше pipe.
• Как отдать большой файл по HTTP, не загрузив его в память.
• Чем Buffer.alloc отличается от Buffer.allocUnsafe.