JavaScript Streams API

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

API построен вокруг трёх основных типов. ReadableStream — это источник, из которого вы извлекаете данные. WritableStream — это приёмник, в который вы записываете данные. TransformStream располагается посередине: принимает фрагменты с одного конца и выдаёт изменённые фрагменты с другого. Как только вы поймёте эти три типа, вы сможете объединять их в эффективные конвейеры.

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

Самый распространённый способ получить поток — Fetch API. Объект Response предоставляет своё тело как ReadableStream через response.body, так что вы можете получать загрузку фрагмент за фрагментом, не ожидая полного ответа с помощью response.text().

Для ручного чтения вызовите getReader(), чтобы заблокировать ридер на потоке, затем зациклитесь на reader.read(). Каждый вызов разрешается объектом с полями done и value:

const response = await fetch('/large-file.txt');
const reader = response.body.getReader();

while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  // value is a Uint8Array chunk of bytes
  console.log('Received', value.length, 'bytes');
}

Каждое value — это Uint8Array, фрагмент сырых байтов, а не строка (см. Типизированные массивы). Когда done равно true, поток завершён, а value — undefined. Для преобразования байтов в текст обычно используется TextDecoder, который может склеивать фрагменты, даже если многобайтовый символ разделён между двумя чтениями:

Этот же цикл лежит в основе создания индикаторов прогресса загрузки: суммируйте длину каждого фрагмента и сравнивайте её с заголовком Content-Length.

Асинхронная итерация

В современных окружениях ReadableStream поддерживает асинхронную итерацию, поэтому ручной цикл с ридером можно заменить на for await...of (см. асинхронные итераторы и генераторы):

const response = await fetch('/large-file.txt');

for await (const chunk of response.body) {
  // chunk is a Uint8Array
  console.log('Received', chunk.length, 'bytes');
}

Это более чистый способ, поскольку цикл сам обрабатывает done и автоматически освобождает ридер. Однако здесь есть нюанс с поддержкой: Node.js справляется с этим хорошо, но прямая асинхронная итерация по response.body всё ещё неравномерно реализована в разных браузерах.

Создание ReadableStream

Вы можете создать собственный источник, передав объект underlying source в конструктор ReadableStream. Он может определять три необязательных метода:

• start(controller) выполняется один раз при создании потока — удобно для инициализации или отправки начальных данных.
• pull(controller) вызывается каждый раз, когда потребитель хочет получить больше данных и во внутренней очереди есть место.
• cancel(reason) выполняется, если потребитель прекращает чтение раньше времени, позволяя выполнить очистку.

Вы добавляете данные с помощью controller.enqueue(chunk) и сигнализируете о завершении с помощью controller.close():

Поток может содержать любые значения JavaScript, не только байты — здесь он выдаёт обычные числа. Когда источник медленный или не имеет конца (WebSocket, таймер, данные сенсора), поместите логику в pull(), чтобы фрагменты создавались только по запросу потребителя.

TransformStream

TransformStream изменяет фрагменты по мере их прохождения. Вы передаёте ему функцию transform(chunk, controller), которая получает каждый входящий фрагмент и вызывает controller.enqueue() с преобразованным результатом:

const upperCaser = new TransformStream({
  transform(chunk, controller) {
    controller.enqueue(chunk.toUpperCase());
  }
});

Transform stream предоставляет сторону writable (куда поступают фрагменты) и сторону readable (откуда они выходят), что именно и делает конвейерную обработку возможной.

Платформа поставляет несколько готовых трансформаций, поэтому вам редко придётся писать логику на уровне байтов вручную:

• TextDecoderStream / TextEncoderStream — преобразуют между фрагментами байтов и фрагментами текста.
• CompressionStream / DecompressionStream — применяют gzip или deflate на лету.

Объединение потоков в конвейер

Вместо того чтобы вручную подключать ридеры и врайтеры, вы можете соединять потоки напрямую. Для этого есть два метода:

• readable.pipeTo(writable) отправляет каждый фрагмент из читаемого потока в записываемый и разрешает промис по завершении.
• readable.pipeThrough(transformStream) пропускает данные через трансформацию и возвращает новый читаемый поток — идеально для цепочки.

Сочетание pipeThrough с TextDecoderStream даёт вам текстовые фрагменты прямо из сетевого ответа без ручного управления декодером:

const response = await fetch('/large-file.txt');
const textStream = response.body.pipeThrough(new TextDecoderStream());

for await (const textChunk of textStream) {
  console.log(textChunk); // already a string
}

Вы можете выстраивать столько этапов, сколько захотите — например, response.body.pipeThrough(new DecompressionStream('gzip')).pipeThrough(new TextDecoderStream()), чтобы распаковать и декодировать данные в одном декларативном конвейере.

Обратное давление

Ключевое преимущество потоков перед буферизацией всего подряд — это обратное давление (backpressure). Когда потребитель работает медленно, поток автоматически сигнализирует источнику приостановить производство и возобновляет его, как только очередь опустеет. С pipeTo и pipeThrough это происходит автоматически — быстрая загрузка не обгонит медленную запись на диск и не переполнит память.

Случаи применения

Потоки особенно эффективны, когда данные объёмны, поступают медленно или непрерывно:

• Постепенный рендеринг — отображайте начало большого ответа, пока остальное ещё загружается, вместо того чтобы смотреть на пустой экран.
• Загрузка и выгрузка с прогрессом — измеряйте байты по мере их передачи для отображения индикатора прогресса.
• Обработка больших файлов — обрабатывайте файл фрагмент за фрагментом, чтобы потребление памяти оставалось постоянным даже для файлов, превышающих размер RAM.
• Конвейеры сжатия — пропускайте данные через CompressionStream или DecompressionStream для gzip-сжатия на лету.

Поддержка браузерами и средами выполнения

ReadableStream, WritableStream и TransformStream поддерживаются во всех современных браузерах и в Node.js (где они также доступны через node:stream/web). Внимания заслуживают более поздние дополнения: асинхронная итерация по response.body и CompressionStream появились позже, поэтому проверяйте поддержку или предоставляйте запасной вариант с getReader(), если вам нужен широкий охват. Потоки тесно связаны с объектами Blob — blob.stream() возвращает ReadableStream, позволяя интегрировать файлоподобные объекты в потоковый конвейер.

Проверьте свои знания