Исторически в среде Node.js развивалась собственная модель потоков данных (node:stream), основанная на системном классе EventEmitter, коллбэках и бинарных объектах Buffer. Параллельно консорциум WHATWG стандартизировал Web Streams API для браузеров и edge-рантаймов. Веб-стандарт строится на парадигме Promise-контрактов, типизированных массивах Uint8Array, явных блокировках чтения (Reader Locks) и функциональных конвейерах pipeThrough.
В современном fullstack-стеке (Node.js 18-24+, Next.js 15+, React 19, Cloudflare Workers, Vercel Serverless) Web Streams API стал универсальным фундаментом для передачи данных. Нативные методы глобального fetch(), спецификация Request и Response, потоковый рендеринг React Server Components (RSC Flight Protocol) и обработчики Next.js Route Handlers используют стандарт WHATWG Streams по умолчанию.
Понимание механизмов работы Web Streams и способов их бесшовной интеграции с традиционными потоками Node.js (node:stream) является критическим навыком для разработки отказоустойчивых высоконагруженных бэкендов и современных SSR-приложений.
Спецификация Web Streams API определяет три ключевые сущности:
ReadableStream: источник данных. Управляется через внутренний контроллер:
ReadableStreamDefaultController: контроллер общего назначения, передающий чанки произвольных типов.ReadableByteStreamController (BYOB — Bring Your Own Buffer): оптимизированный контроллер для низкоуровневого бинарного I/O. Позволяет потребителю передавать собственный предварительно аллоцированный буфер, исключая лишние копирования памяти в ядре V8.WritableStream: абстракция целевого приемника данных, управляемая контроллером WritableStreamDefaultController.TransformStream: пара из связанного WritableStream (вход stream.writable) и ReadableStream (выход stream.readable), предназначенная для модификации или перекодирования чанков на лету.В отличие от потоков Node.js, где на события 'data' можно навесить неограниченное число слушателей, Web Streams реализуют строгую модель безопасности данных через блокировку:
const reader = readableStream.getReader();
console.log(readableStream.locked); // trueКак только вызван метод getReader(), поток переходит в заблокированное состояние (locked === true). Никакой другой компонент приложения не может подключиться к потоку или вызвать pipeTo(), пока текущий ридер не вызовет reader.releaseLock() или не завершит чтение. Если требуется передать один поток двум независимым потребителям, поток необходимо явно разветвить методом readableStream.tee().
Управление обратным давлением в Web Streams API определяется стратегиями очередей (Queuing Strategies):
CountQueuingStrategy({ highWaterMark: N }): измеряет наполненность буфера в количестве дискретных элементов.ByteLengthQueuingStrategy({ highWaterMark: N }): вычисляет суммарный объем данных в байтах на основе свойства chunk.byteLength.Когда суммарный вес элементов во внутренней очереди достигает highWaterMark, свойство controller.desiredSize становится неположительным (меньше или равно 0). Это сигнализирует источнику о необходимости приостановить генерацию чанков.
Для устранения фрагментации рантайма в модуль node:stream были внедрены двусторонние адаптеры с нулевыми накладными расходами на копирование:
Readable.toWeb(nodeReadable): трансформирует поток Node.js в стандартный ReadableStream.Readable.fromWeb(webReadableStream): преобразует WHATWG ReadableStream в привычный Readable из node:stream.Writable.toWeb(nodeWritable) и Writable.fromWeb(webWritableStream): аналогичные мосты для потоков записи.Конвейер Web Streams представляет собой замкнутую цепь со строгим контролем блокировок и сквозной передачей сигналов готовности.
Универсальный стандарт потоковой передачи данных на основе промисов, стандартизированный для браузеров, Edge и серверных сред Node.js. Состояние эксклюзивного владения потоком чтения (ReadableStream), исключающее состояние гонки (Race Conditions) между несколькими потребителями. Операция детерминированного расщепления одного ReadableStream на два параллельных независимых потока с сохранением исходных чанков. Метод сквозной передачи данных через цепочку TransformStream (аналог конвейерного оператора в Unix).[ ReadableStream ] (locked: true)
|
| (pipeThrough)
v
+-----------------------+
| TransformStream |
| [writable] -> [readable]
+-----------------------+
|
| (pipeThrough)
v
+------------------------------------+
| CompressionStream('gzip') |
| [writable: raw] -> [readable: gzip]|
+------------------------------------+
|
| (pipeTo)
v
[ WritableStream / HTTP Response ]Клиент отправляет HTTP-запрос к Route Handler с заголовком Accept-Encoding: gzip.
Создание ReadableStream с контролем наполнения буфера через desiredSize.
Прокачка данных через CompressionStream('gzip') методом pipeThrough.
Передача сжатых байтов в сокет с заголовком Transfer-Encoding: chunked.
Вызов controller.close() по окончании данных и автоматическое закрытие HTTP-сессии.
Почему при вызове stream.getReader() свойство stream.locked переходит в состояние true и что это означает?
new Response(readableStream).<Suspense>.CompressionStream('gzip') без использования внешних библиотек и одинаково работающая в Node.js и Edge Workers.Реализуем генератор потока чисел, учитывающий обратное давление через desiredSize.
export function createIncrementalStream(maxItems: number): ReadableStream<Uint8Array> {
let current = 0;
const encoder = new TextEncoder();
return new ReadableStream<Uint8Array>({
// Инициализация внутреннего состояния
start(controller) {
console.log('Поток инициализирован');
},
// Метод pull вызывается автоматически, когда внутренний буфер готов принять данные
async pull(controller) {
// Пока потребитель готов принимать данные (desiredSize > 0)
while ((controller.desiredSize ?? 0) > 0 && current < maxItems) {
current++;
const chunk = encoder.encode(`data: {"id": ${current}, "timestamp": ${Date.now()}}\n\n`);
controller.enqueue(chunk);
// Имитация асинхронной паузы для демонстрации backpressure
if (current % 100 === 0) {
await new Promise((resolve) => setTimeout(resolve, 10));
}
}
if (current >= maxItems) {
controller.close();
}
},
// Вызывается при отмене потока потребителем (например, обрыв сетевого соединения)
cancel(reason) {
console.warn('Поток отменен клиентом:', reason);
},
}, new ByteLengthQueuingStrategy({ highWaterMark: 64 * 1024 }));
}Современные библиотеки (драйверы баз данных, AWS SDK) часто возвращают потоки node:stream. Для их использования в Route Handlers Next.js требуется мост Readable.toWeb().
import { Readable, Writable } from 'node:stream';
import { createReadStream } from 'node:fs';
// 1. Конвертация потока чтения Node.js в WHATWG ReadableStream
const nodeReadable = createReadStream('large-dataset.csv');
const webReadableStream: ReadableStream<Uint8Array> = Readable.toWeb(nodeReadable);
// 2. Использование Web Stream в стандартном цикле чтения с блокировкой ридера
async function consumeWebStream(stream: ReadableStream<Uint8Array>): Promise<number> {
const reader = stream.getReader();
let totalBytes = 0;
try {
while (true) {
const { value, done } = await reader.read();
if (done) break;
totalBytes += value.byteLength;
}
} finally {
// Критически важно освободить блокировку при завершении или ошибке
reader.releaseLock();
}
return totalBytes;
}
// 3. Обратная конвертация: WHATWG ReadableStream -> Node.js Readable
const convertedBackToNode: Readable = Readable.fromWeb(webReadableStream);Платформенный класс CompressionStream позволяет строить высокопроизводительные конвейеры сжатия без сторонних npm-пакетов.
import { NextRequest, NextResponse } from 'next/server';
import { Readable } from 'node:stream';
import { createReadStream } from 'node:fs';
import { join } from 'node:path';
export async function GET(req: NextRequest): Promise<Response> {
const filePath = join(process.cwd(), 'data', 'audit-records.json');
// 1. Получаем нативный Node.js поток чтения файла
const fileReadStream = createReadStream(filePath);
// 2. Преобразуем в WHATWG ReadableStream
const sourceWebStream = Readable.toWeb(fileReadStream);
// 3. Пропускаем через нативный конвейер сжатия GZIP
const compressedStream = sourceWebStream.pipeThrough(
new CompressionStream('gzip')
);
// 4. Возвращаем потоковый HTTP-ответ клиенту
return new Response(compressedStream, {
status: 200,
headers: {
'Content-Type': 'application/json',
'Content-Encoding': 'gzip',
'Transfer-Encoding': 'chunked',
'Cache-Control': 'no-cache, no-store',
},
});
}Чтение файла через fs.createReadStream с размером чанка 64 КБ.
Оборачивание в WHATWG ReadableStream без копирования нативных буферов.
Наложение сжатия CompressionStream('gzip') с динамическим сжатием чанков.
Формирование ответа с заголовком Transfer-Encoding: chunked.
Браузер декомпрессирует чанки на лету по мере их поступления по протоколу HTTP/2.
Полнофункциональный изоморфный хелпер для безопасного конструирования Web Streams с поддержкой Server-Sent Events, компрессии и автоматической очистки при обрыве клиентского соединения.
export interface StreamPipelineOptions {
compression?: 'gzip' | 'deflate' | 'none';
signal?: AbortSignal;
highWaterMarkBytes?: number;
onChunkSent?: (bytes: number) => void;
}
export class WebStreamPipelineHelper {
private static readonly DEFAULT_HIGH_WATER_MARK = 64 * 1024; // 64 КБ
/**
* Оборачивает генератор данных в оптимизированный ReadableStream.
*/
public static fromAsyncGenerator<T>(
generator: AsyncGenerator<T, void, unknown>,
serializer: (item: T) => Uint8Array,
options: StreamPipelineOptions = {}
): ReadableStream<Uint8Array> {
const highWaterMark = options.highWaterMarkBytes ?? this.DEFAULT_HIGH_WATER_MARK;
const baseStream = new ReadableStream<Uint8Array>(
{
async pull(controller) {
try {
const { value, done } = await generator.next();
if (done) {
controller.close();
return;
}
const binaryChunk = serializer(value);
controller.enqueue(binaryChunk);
if (options.onChunkSent) {
options.onChunkSent(binaryChunk.byteLength);
}
} catch (err) {
controller.error(err);
}
},
cancel(reason) {
console.warn('Pipeline cancelled by consumer:', reason);
},
},
new ByteLengthQueuingStrategy({ highWaterMark })
);
return this.applyCompression(baseStream, options.compression);
}
/**
* Накладывает потоковую компрессию при необходимости.
*/
public static applyCompression(
stream: ReadableStream<Uint8Array>,
compression: 'gzip' | 'deflate' | 'none' = 'none'
): ReadableStream<Uint8Array> {
if (compression === 'none') {
return stream;
}
const compressor = new CompressionStream(compression);
return stream.pipeThrough(compressor);
}
/**
* Формирует заголовки HTTP-ответа для потоковой отдачи данных.
*/
public static createStreamHeaders(
contentType: string = 'text/event-stream',
compression: 'gzip' | 'deflate' | 'none' = 'none'
): Headers {
const headers = new Headers({
'Content-Type': contentType,
'Cache-Control': 'no-cache, no-transform',
'Connection': 'keep-alive',
'X-Content-Type-Options': 'nosniff',
});
if (compression !== 'none') {
headers.set('Content-Encoding', compression);
}
return headers;
}
}Каким образом реализуется механизм Backpressure в WHATWG ReadableStream по сравнению с классическими потоками Node.js?
Разработчик вызывает stream.getReader() в одном месте приложения, а затем пытается передать тот же stream в new Response(stream) или вызвать stream.pipeThrough(). Приложение выбрасывает фатальную ошибку TypeError: Cannot get a reader for a locked stream.
Решение: Перед повторным использованием освобождайте блокировку через reader.releaseLock(), либо, если поток требуется направить в два адресата, предварительно разветвляйте его через const [branchA, branchB] = stream.tee().
В кастомном ReadableStream разработчик обрабатывает условие завершения данных, но забывает вызвать controller.close(). В результате сетевой клиент (браузер или curl) бесконечно ожидает завершающих байтов, соединение висит до наступления системного тайм-аута, а сервер удерживает выделенные ресурсы.
Решение: Всегда вызывайте controller.close() при исчерпании источника, а также обрабатывайте ошибки через controller.error(err).
Попытка передать fs.createReadStream() напрямую в конструктор глобального new Response(nodeStream) в современных версиях Next.js или Node.js. Конструктор Response ожидает экземпляр ReadableStream (WHATWG) или AsyncIterable. В зависимости от окружения это приводит к немедленной ошибке типизации или передаче пустого ответа без чанков.
Решение: Всегда оборачивайте классические потоки Node.js через Readable.toWeb(nodeStream).
Клиент закрывает вкладку браузера или прерывает загрузку, но сервер продолжает генерировать чанки и выгружать данные из базы данных. Сервер тратит процессорное время и сетевой трафик на передачу данных в разорванный сокет.
Решение: Подписывайтесь на сигнал отмены req.signal в Route Handler и передавайте его в генератор данных для своевременной остановки вычислений.