← назад к разделу

Потоковая обработка нужна не для красоты API. Она не дает быстрому источнику заполнить память, пока медленный потребитель читает диск, сеть или другой сервис.

Зачем нужен backpressure

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

Backpressure - это сигнал «пока не приноси еще». Writable сообщает false из write, источник ждет drain, после чего продолжает.

const input = createReadStream('/tmp/orders.ndjson');
for await (const chunk of input) {
  if (!response.write(chunk)) {
    await once(response, 'drain');
  }
}
response.end();

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

pipeline делает это за вас

pipeline связывает потоки, учитывает backpressure и уничтожает всю цепочку при ошибке.

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

await pipeline(
  createReadStream('/tmp/orders.ndjson'),
  createWriteStream('/tmp/orders-copy.ndjson'),
);

После успешного await pipeline потребитель завершен. При ошибке входа или выхода исключение доходит до вызывающего кода, а потоки не остаются висеть.

highWaterMark - размер буфера, не предохранитель

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

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

Коротко

  • Backpressure останавливает источник, пока потребитель не освободил место.
  • false из write требует ждать drain.
  • pipeline учитывает backpressure, ошибки и закрытие цепочки.
  • highWaterMark управляет буфером, а не устраняет очередь.
  • Потоки экономят память только при правильном ожидании завершения.

Что почитать дальше