Node.js Streams: что за пределами основ — память, обратное давление и реальные сбои
Узнайте, как потоки Node.js взаимодействуют с Web Streams, оцените экономию памяти с помощью реальных тестов, а также ошибках в производственной среде, которые проявляются только при высокой нагрузке.
Источники: Node.js Streams in Production: The Parts Most Tutorials Skip
Многие введения в потоки Node.js хорошо освещают основы: читаемость, запись, двунаправленная связь, преобразование данных, функция pipe(), механизм обратного давления, пример работы с gzip, а иногда и сервер для запросов к определённым участкам видео в качестве финала. Если вы уже ознакомились с одним или двумя из таких материалов, вы понимаете, что такое поток, и почему полная загрузка 2-гигабайтного файла в память во время прямой демонстрации приводит к проблемам.
В этих введениях редко говорится о том, что происходит дальше. Сколько памяти на самом деле экономится при переходе на потоки, и каковы конкретные цифры вместо расплывчатого утверждения о «меньшем потреблении»? Как нативные потоки Node взаимодействуют с API Web Streams, используемым функцией fetch и браузерами? Как передавать результаты работы модели, работающей на собственном оборудовании, в виде потоков, вместо того чтобы платить за каждый токен через хостинговое API? И какие две-три ошибки в конечном итоге разрушают работу цепочки обработки, которая казалась надежной во время тестирования, при реальном объеме трафика?
В этой статье рассматриваются именно эти вопросы. Предполагается, что вы уже знаете, что такое поток с чтением. Если это не так, краткий обзор ниже поможет вам быстро освоить тему. Все, что будет рассказано далее, основано на этих основах.
Краткий обзор за 60 секунд
Readable -> produces data (fs.createReadStream, HTTP request body)
Writable -> consumes data (fs.createWriteStream, HTTP response)
Duplex -> both, independently (a TCP socket)
Transform -> consumes, changes, produces (zlib.createGzip)
Readable stream -> Transform stream -> Writable stream
Данные проходят через систему небольшими порциями, а не как один большой блок. Функция pipe() соединяет потоки между собой и обрабатывает ситуацию, когда пункт назначения не может поспевать за источником — это так называемое обратное давление. Функция pipeline() выполняет ту же задачу по соединению потоков, но также передает ошибки из любого потока в цепочке и гарантирует очистку при возникновении сбоев, поэтому именно её следует использовать в долгосрочных сервисах, а не в краткосрочных одноразовых скриптах.
import { pipeline } from 'node:stream/promises';
import { createReadStream, createWriteStream } from 'node:fs';
import { createGzip } from 'node:zlib';
await pipeline(
createReadStream('./input.txt'),
createGzip(),
createWriteStream('./input.txt.gz')
);
Если этот пример уже полностью понятен вам, можете пропустить его. Далее приведена информация, которую стоит знать перед тем, как впервые размещать код на основе потоков в производственной среде.
Число, которое никто не указывает в этих статьях
Большинство статей о потоковой обработке утверждают, что буферизированный подход «использует много памяти», в то время как потоковый подход «использует гораздо меньше», при этом ни в одном из утверждений не приводятся конкретные цифры. Чтобы исправить это, был создан файл простого текста объемом 501 МБ, и оба подхода были протестированы на одном компьютере по два раза каждый.
// buffered.js: read the whole file, then scan it
import { readFileSync } from 'node:fs';
const before = process.memoryUsage().rss;
const data = readFileSync('./big.txt', 'utf8');
const afterRead = process.memoryUsage().rss;
let count = 0;
for (let i = 0; i < data.length; i++) if (data[i] === 'q') count++;
console.log({
rssBeforeMB: (before / 1024 / 1024).toFixed(1),
rssAfterReadMB: (afterRead / 1024 / 1024).toFixed(1),
count
});
// streamed.js: read in 64KB chunks, scan as it arrives
import { createReadStream } from 'node:fs';
let peakRss = 0;
const timer = setInterval(() => {
peakRss = Math.max(peakRss, process.memoryUsage().rss);
}, 10);
let count = 0;
const stream = createReadStream('./big.txt', { encoding: 'utf8', highWaterMark: 64 * 1024 });
stream.on('data', (chunk) => {
for (let i = 0; i < chunk.length; i++) if (chunk[i] === 'q') count++;
});
stream.on('end', () => {
clearInterval(timer);
console.log({ peakRssMB: (peakRss / 1024 / 1024).toFixed(1), count });
});
Вот результаты тестирования на файле ASCII объемом 500 МБ с использованием Node.js 22, по два запуска для каждого подхода:
Approach Peak RSS Wall time Notes
--------------------------------------------------------------------
readFileSync (buffered) ~539 MB 1.24 - 1.76s Full file materialized as a string
createReadStream, 64KB chunks ~52 MB 1.56 - 1.67s Roughly 10x less peak memory
createReadStream, 1MB chunks ~50 MB 1.25 - 1.34s Bigger chunks, same ballpark memory
Следует четко указать несколько важных моментов, иначе результаты окажутся вводящими в заблуждение: речь идет о одной машине, одном файле и одном нагрузочном случае — простом сканировании символов. Время выполнения по часам оказалось примерно одинаковым у обоих подходов, причем версия с потоковой обработкой даже в некоторых местах слегка опередила ее, что было довольно неожиданно. Изначально предполагалось, что накладные расходы на обработку множества небольших фрагментов замедлят процесс сильнее, чем это произошло на самом деле. Что касается показателей памяти, то они в значительной степени соответствовали ожиданиям: при использовании буферизации пик потребления памяти приближался к полному размеру файла, тогда как при потоковой обработке пик оставался низким и практически не менялся независимо от размера файла. Именно этот контраст и является основанием для использования потоков, сведенным в одну таблицу — его легко проверить самостоятельно, запустив оба приведенных выше скрипта на файле
свой собственный.Практический вывод: для одного запроса, обрабатываемого достаточно мощным устройством, часто достаточно просто загрузить весь файл в память. Настоящая проблема никогда не заключалась в отдельном запросе в изоляции — речь всегда идет о следующем вопросе: что произойдёт, когда 100 таких запросов будут выполняться одновременно. Использование памяти при потоковом подходе практически не меняется с ростом параллельности, поскольку каждый отдельный запрос хранит лишь буфер определённого размера, а не весь объём данных сразу.
Node streams и Web Streams теперь относятся к одной семье
В течение многих лет потоки Node.js и API Streams от WHATWG (ReadableStream, WritableStream, TransformStream — классы, лежащие в основе функции fetch() как в браузерах, так и в Node) казались двумя отдельными экосистемами с минимальным перекрытием. Сейчас это уже не соответствует действительности. Модуль node:stream/web позволяет напрямую использовать классы WHATWG, а собственные классы Node Readable, Writable и Duplex оснащены методами .toWeb() и .fromWeb(); по состоянию на текущие версии Node.js их стабильность оценивается как уровень 2 („Стабильно“).
Это становится важным в тот момент, когда вы вызываете fetch() изнутри сервиса Node, поскольку объект response.body, который возвращается, является Web-ReadableStream, а не Node-Readable.
import { createWriteStream } from 'node:fs';
import { Writable } from 'node:stream';
import { pipeline } from 'node:stream/promises';
const response = await fetch('https://example.com/large-report.csv');
if (!response.ok || !response.body) {
throw new Error(`Download failed with status ${response.status}`);
}
// response.body is a Web ReadableStream, so convert the Node side instead
const fileStream = Writable.toWeb(createWriteStream('./report.csv'));
await response.body.pipeTo(fileStream);
console.log('Download saved without buffering the response in memory');
Преобразование происходит так же плавно в обратном направлении — вы можете превратить объект Node Readable в поток, который будет принят любым API, основанным на Web-stream:
import { createReadStream } from 'node:fs';
import { Readable } from 'node:stream';
const webStream = Readable.toWeb(createReadStream('./large-file.bin'));
// Now this works with fetch's request body, a browser upload API,
// or any library built against the WHATWG interface instead of Node's.
await fetch('https://uploads.example.com/put', {
method: 'PUT',
body: webStream,
duplex: 'half' // required by fetch when the body is a stream
});
Один момент, который сбивает людей в первый раз: функция fetch выбросит ошибку, если вы передадите ей тело запроса в формате потока без указания параметра duplex: 'half'. Это требование существует потому, что тела запросов в формате потока не поддерживаются одинаково везде, поэтому спецификация требует явного указания этого параметра. Это незначительная проблема, которая займет пятнадцать минут один раз и больше не будет встречаться.
Настоящая польза заключается в том, что ваш backend на Node, функции на периферии и код браузера теперь могут в значительной степени использовать одну и ту же логику обработки потоков, вместо того чтобы поддерживать версию, основанную на fetch, и отдельную версию, использующую функцию pipe() и модули fs с http от Node.
stream.compose(): конвейер, возвращающий один поток
stream.compose() технически существует с версии Node.js 16.9.0, но долгое время находился в экспериментальном режиме, прежде чем наконец стабилизироваться в одном из последних выпусков Node.js. Теперь, когда его API устоялся, он заслуживает более внимательного рассмотрения.
Идея заключается в следующем: вместо создания полного подкласса Transform с методами _transform и _flush вы пишете асинхронную генераторную функцию, и compose() объединяет всю эту цепочку в один поток типа Duplex, который можно использовать так же, как любой другой поток.
import { compose } from 'node:stream';
import { createReadStream, createWriteStream } from 'node:fs';
async function* toUpperCase(source) {
for await (const chunk of source) {
yield chunk.toString().toUpperCase();
}
}
async function* onlyNonEmptyLines(source) {
let leftover = '';
for await (const chunk of source) {
const lines = (leftover + chunk).split('\n');
leftover = lines.pop() ?? '';
for (const line of lines) {
if (line.trim()) yield line + '\n';
}
}
if (leftover.trim()) yield leftover;
}
const pipeline = compose(
createReadStream('./raw.txt', { encoding: 'utf8' }),
onlyNonEmptyLines,
toUpperCase
);
pipeline.pipe(createWriteStream('./clean.txt'));
Если добавить это к написанию двух отдельных классов Transform — включая логику ручного буферизации для разделения строк — получится такое же основное поведение, но с значительно меньшим количеством вспомогательных элементов, плюс код, который читается сверху вниз в том же порядке, в котором фактически поступают данные. Функция compose() также гибка в отношении своих концов: если передать в качестве первого аргумента обычный объект AsyncIterable, она превращается в удобный для чтения объект Duplex; если передать в качестве последнего аргумента объект AsyncFunction, она становится объектом для записи. Это означает, что целую систему обработки можно построить исключительно с использованием асинхронных генераторов, если вы предпочитаете такой подход.
Ничто из этого не является причиной для удаления уже работающих классов Transform. Однако для нового кода — особенно того, который сводится к операциям «парсинга, фильтрации и преобразования» — использование асинхронного генератора в сочетании с функцией compose() позволяет избежать необходимости создания нескольких небольших подклассов Transform, которые в итоге будут выполнять одинаковую функцию.
Стримирование токенов из локальной модели
Почти во всех разделах учебников, посвящённых «реальным потокам данных», приводятся три одинаковых примера: загрузка файлов, сжатие в формате gzip и запросы к видео по определённым диапазонам. Это действительно важные сценарии использования, но существует более современный подход, который так же хорошо соответствует модели стриминга и редко рассматривается: получение выходных данных по токенам из крупной языковой модели.
Поскольку пример, который действительно можно скопировать и вставить, не должен зависеть от чужого API-ключа, приведённый ниже пример взаимодействует с Ollama, запущенным локально, а не с облачным провайдером:
# one-time setup
ollama pull llama3.2
ollama serve # usually already running as a background service
По умолчанию конечная точка /api/generate в Ollama передаёт свой ответ в виде JSON, разделённого символами новой строки (application/x-ndjson): по одному объекту JSON на строку, каждый из которых содержит фрагмент response и флаг done, причём последняя строка содержит значение "done": true. Если убрать форматирование, это просто читаемый поток с простым парсингом текста — тот же подход к обработке данных по чанкам, используемый во всей статье, применим без изменений.
// server.js: proxies Ollama's token stream to the browser over SSE
import { createServer } from 'node:http';
const server = createServer(async (request, response) => {
if (request.url !== '/ask') {
response.statusCode = 404;
response.end('Not found');
return;
}
response.writeHead(200, {
'Content-Type': 'text/event-stream',
'Cache-Control': 'no-cache',
Connection: 'keep-alive'
});
const ollamaResponse = await fetch('http://localhost:11434/api/generate', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
model: 'llama3.2',
prompt: 'Explain backpressure in one short paragraph.',
stream: true
})
});
if (!ollamaResponse.body) {
response.end();
return;
}
// Web ReadableStream from fetch -> async iterate it directly (Node 18+)
let buffered = '';
for await (const chunk of ollamaResponse.body) {
buffered += Buffer.from(chunk).toString('utf8');
const lines = buffered.split('\n');
buffered = lines.pop() ?? '';
for (const line of lines) {
if (!line.trim()) continue;
const event = JSON.parse(line);
response.write(`data: ${JSON.stringify({ text: event.response })}\n\n`);
if (event.done) {
response.write('event: done\ndata: {}\n\n');
response.end();
return;
}
}
}
// client disconnected mid-stream? stop pulling tokens we can't deliver
request.on('close', () => {
if (!response.writableEnded) response.end();
});
});
server.listen(3000, () => console.log('Ask at http://localhost:3000/ask'));
Если вы хотите просмотреть необработанную строку NDJSON от Ollama без окружающей её структуры SSE, она выглядит так:
{"model":"llama3.2","created_at":"2026-07-30T10:15:24.121Z","response":"Back","done":false}
{"model":"llama3.2","created_at":"2026-07-30T10:15:24.132Z","response":"pressure","done":false}
{"model":"llama3.2","created_at":"2026-07-30T10:15:24.166Z","response":" happens...","done":true,"done_reason":"stop"}
Сразу два момента могут сбить людей с толку при первом знакомстве. Во-первых, в текущих версиях Node.js атрибут response.body у метода fetch уже сам по себе является асинхронно итерируемым, поэтому нет необходимости вызывать Readable.fromWeb() именно в этом случае — хотя этот вспомогательный метод всё ещё доступен, если какая-то библиотека ниже в цепочке ожидает классический объект Readable от Node. Во-вторых, строки в формате NDJSON могут распространяться на несколько границ блоков, что представляет собой ту же проблему, что и при парсинге CSV в потоковом режиме: нельзя гарантировать, что один блок данных от fetch совпадёт с одной строкой JSON. Именно по этой причине в приведённом выше примере частичные строки сохраняются в буфере, а не выполняется прямой вызов JSON.parse(chunk) для полученных данных.
Если вы предпочитаете использовать хостинговый API — OpenAI, Anthropic или аналогичные — общая схема практически не меняется: вы по-прежнему получаете HTTP-ответ в фрагментах и передаёте его клиенту в формате SSE. Локальная версия Ollama — это просто та, которую можно запустить сразу без необходимости регистрации или использования ключей, и она является разумным выбором для разработки клиентской части такой функции во время тестирования, прежде чем направить тот же код на хостинговую модель, когда потребуется качество вывода производственного уровня.
Потоковая передача файла непосредственно в объектное хранилище
Отправка файла в «S3» почти никогда не подразумевает наличия файла достаточно маленького, чтобы уместиться в памяти заранее. Пакет @aws-sdk/lib-storage существует именно потому, что простой класс PutObjectCommand требует знания длины содержимого до начала запроса. Его вспомогательный метод Upload принимает поток данных, самостоятельно определяет, подходит ли один запрос или многокомпонентная загрузка, и сообщает о ходе выполнения.
import { S3Client } from '@aws-sdk/client-s3';
import { Upload } from '@aws-sdk/lib-storage';
import { createReadStream } from 'node:fs';
const s3 = new S3Client({ region: 'us-east-1' });
async function uploadFile(localPath, bucket, key) {
const upload = new Upload({
client: s3,
params: {
Bucket: bucket,
Key: key,
Body: createReadStream(localPath)
}
});
upload.on('httpUploadProgress', (progress) => {
console.log(`${key}: ${progress.loaded} / ${progress.total ?? '?'} bytes`);
});
await upload.done();
console.log(`${key} uploaded`);
}
Чтобы экспериментировать без необходимости иметь реальный аккаунт AWS, MinIO предоставляет хранилище объектов, совместимое с S3, которое можно запустить локально одним командным приказом. Тот же код @aws-sdk/lib-storage, описанный выше, работает с ним без изменений, достаточно лишь указать клиенту адрес конечной точки MinIO вместо AWS:
docker run -p 9000:9000 -p 9001:9001 \
-e "MINIO_ROOT_USER=admin" \
-e "MINIO_ROOT_PASSWORD=admin12345" \
minio/minio server /data --console-address ":9001"
const s3 = new S3Client({
region: 'us-east-1',
endpoint: 'http://localhost:9000',
forcePathStyle: true, // required for MinIO's path-style URLs
credentials: { accessKeyId: 'admin', secretAccessKey: 'admin12345' }
});
Этот подход следует использовать по умолчанию для файлов размером более нескольких десятков мегабайт: избегайте загрузки всего файла в объект Buffer перед передачей его в SDK. Передавайте объект createReadStream(), и пусть модуль Upload займётся остальным — использование памяти останется неизменным независимо от того, является ли файл 10 МБ или 10 ГБ.
Обработка загружаемых файлов без буферизации тела запроса
Другой аспект загрузки файлов — это сервер, который должен их принимать, и именно здесь чаще всего возникает случайное буферизация — обычно потому, что промежуточный модуль для парсинга тела запроса не был разработан с учётом потоковой обработки и тихо считывает весь многопартный запрос в память до того, как ваш обработчик успеет к нему добраться.
busboy обрабатывает парсинг многокомпонентных форм поэтапно: как только начинают поступать байты для определенного поля файла, он сразу возвращает читаемый поток для этого поля, не дожидаясь полного прихода запроса:
import { createServer } from 'node:http';
import { createWriteStream } from 'node:fs';
import { pipeline } from 'node:stream/promises';
import busboy from 'busboy';
const server = createServer((request, response) => {
if (request.method !== 'POST' || request.url !== '/upload') {
response.statusCode = 404;
response.end();
return;
}
const bb = busboy({ headers: request.headers, limits: { fileSize: 200 * 1024 * 1024 } });
const uploads = [];
bb.on('file', (fieldName, fileStream, info) => {
const destinationPath = `./uploads/${Date.now()}-${info.filename}`;
uploads.push(
pipeline(fileStream, createWriteStream(destinationPath)).catch((error) => {
fileStream.resume(); // drain on failure so the request can still finish
throw error;
})
);
});
bb.on('close', async () => {
try {
await Promise.all(uploads);
response.writeHead(200, { 'Content-Type': 'application/json' });
response.end(JSON.stringify({ status: 'ok' }));
} catch (error) {
response.statusCode = 500;
response.end(JSON.stringify({ status: 'error', message: error.message }));
}
});
request.pipe(bb);
});
server.listen(3000);
В этом обработчике никогда не хранится весь тело запроса сразу. Байты перемещаются из сокета через парсер busboy прямо на диск. Именно это отличие позволяет сервису успешно справляться с несколькими крупными загрузками параллельно и не срываться при первой же попытке двух пользователей одновременно загрузить видео.
Тестирование пользовательского потока преобразования
Ручно создаваемые потоки Transform — идеальные кандидаты для тихого разрушения: никто не тратит усилия на написание тестов для них, все считают, что они «просто работают», а затем случай с _flush приводит к сбою, и никто этого не замечает до момента ввода в производство. Однако протестировать такой поток не так уж и сложно, как кажется. Вам не нужно создавать реальный файл или настоящий HTTP-сервер. Для этого достаточно использовать Readable.from() в сочетании с stream/promises.
// uppercase-transform.js
import { Transform } from 'node:stream';
export function createUppercaseTransform() {
return new Transform({
transform(chunk, encoding, callback) {
callback(null, chunk.toString().toUpperCase());
}
});
}
// uppercase-transform.test.js
import { describe, it, expect } from 'vitest'; // works the same with Jest
import { Readable } from 'node:stream';
import { pipeline } from 'node:stream/promises';
import { createUppercaseTransform } from './uppercase-transform.js';
async function collect(readable) {
const parts = [];
for await (const chunk of readable) parts.push(chunk.toString());
return parts.join('');
}
describe('uppercase transform', () => {
it('uppercases each chunk', async () => {
const source = Readable.from(['hello ', 'world']);
const transform = createUppercaseTransform();
source.pipe(transform);
expect(await collect(transform)).toBe('HELLO WORLD');
});
it('propagates errors from upstream instead of hanging', async () => {
const source = new Readable({
read() {
this.destroy(new Error('upstream failed'));
}
});
const transform = createUppercaseTransform();
await expect(pipeline(source, transform, async function drain(it) {
for await (const _ of it) { /* discard */ }
})).rejects.toThrow('upstream failed');
});
});
Второй тест в этом файле имеет большее значение, чем кажется на первый взгляд. Он подтверждает, что ошибка, возникшая выше по цепочке обработки, действительно передается дальше, вместо того чтобы заставить поток преобразования застрять на полпути с обещанием, которое так и не будет выполнено. Именно такая скрытая застревание является наиболее распространенной причиной того, почему канал обработки, который кажется работающим, превращается в застрявший запрос после запуска.
Легкий инструмент для мониторинга
Вам не нужна полноценная платформа для наблюдаемости, чтобы проверить работу конвейера. Часто достаточно небольшого потока PassThrough, который считает количество байтов и фрагментов данных по мере их прохождения; он может быть встроен в любой существующий конвейер без влияния на остальные компоненты:
import { PassThrough } from 'node:stream';
export function createMetricsTap(label) {
let bytes = 0;
let chunks = 0;
const startedAt = process.hrtime.bigint();
const tap = new PassThrough();
tap.on('data', (chunk) => {
bytes += chunk.length;
chunks += 1;
});
tap.on('end', () => {
const seconds = Number(process.hrtime.bigint() - startedAt) / 1e9;
console.log(`[${label}] ${chunks} chunks, ${(bytes / 1024 / 1024).toFixed(2)} MB, ` +
`${(bytes / 1024 / 1024 / seconds).toFixed(2)} MB/s`);
});
return tap;
}
await pipeline(
createReadStream('./large-file.bin'),
createMetricsTap('upload'),
createGzip(),
createWriteStream('./large-file.bin.gz')
);
Если вам нужно что-то большее, чем просто вывод в console.log, замените логику обработчиков 'data' и 'end' на объекты Counter или Histogram из библиотеки prom-client. Сам инструмент останется прежним: это просто еще один поток внутри конвейера, отчитывающий о проходящих данных.
Три ошибки, которые проявляются только при высокой нагрузке
Unhandled 'error' events bring down the whole process. A stream that emits an error with no 'error' listener attached doesn't just fail its own request. In Node.js, an 'error' event on an EventEmitter that nobody handles gets thrown, and if nothing catches that throw, the process exits. It only takes one malformed upload to take an entire service offline because a manually wired pipe() chain had error listeners on two out of three streams. pipeline() exists specifically to make this class of bug structurally difficult, so prefer it, or make sure every stream you build by hand gets its own .on('error', ...), with no exceptions.
Повторное использование экземпляра потока приводит к утечке слушателей. Предположим, вы оборачиваете создание потока в функцию, которая вызывается при каждом запросе, но какой-то элемент вне этой функции сохраняет к нему ссылку и привязывает свои собственные слушатели «на всякий случай». В конечном итоге ваших логов будет заполнено предупреждением MaxListenersExceededWarning, что является способом Node обозначить происходящую утечку памяти. Каждому запросу необходимы собственные, свежесозданные потоки, и ничто вне времени существования этого запроса не должно хранить к ним ссылку после его завершения.
Никто не уничтожает поток данных, когда клиент уходит. Именно эта ошибка приводит к финансовым потерям в производственной среде. Если пользователь закрывает вкладку браузера во время загрузки, а вы не настраиваете обработчик response.on('close', ...) для вызова метода .destroy() у исходного потока, сервер продолжает считывать данные с диска; в худшем случае остается открытым курсор базы данных или активной верхневерхнюю HTTP-соединение, всё это ради клиента, который уже ушел. При достаточной одновременной нагрузке именно так сервер постепенно исчерпывает свои описатели файлов или пул соединений без очевидных причин в логах. Решение всегда заключается в трех простых шагах: отслеживать закрытие целевого потока, уничтожать исходный поток в ответ на это событие и рассматривать такой подход как стандартную практику для всех маршрутов, основанных на потоках, а не как действие, которое нужно вспоминать после каждого инцидента.
Когда использовать тот или иной API
pipe()
Simple, two-or-three-stream chains. Handles backpressure.
Error handling is your responsibility per-stream.
pipeline()
Anything running in production. Handles backpressure,
forwards errors from every stream, cleans up on failure.
Has a promise-based version and supports AbortSignal.
stream.compose()
New pipelines built from async generators instead of
Transform subclasses. Returns one Duplex you can pipe
like any other stream. Stable as of recent Node.js releases.
node:stream/web (ReadableStream/WritableStream/TransformStream)
Anywhere you're mixing Node streams with fetch(), browser
code, or a library built against the WHATWG standard.
Convert with .toWeb() / .fromWeb() at the boundary.
Ни один из этих подходов не делает остальные ненужными. Каждый из них решает отдельную задачу в рамках общего потока данных, проходящего через систему. В настоящее время распространено использование одной пайплайн-структуры, которая начинается с Web ReadableStream, полученного с помощью fetch, преобразует эти данные, передаёт их через трансформатор, собранный с использованием compose(), и в конце передаёт результат в pipeline() для обработки записи. Раньше такое сочетание API казалось необычным, но сегодня это довольно распространённый паттерн.
Ничто здесь не освобождает от необходимости сначала понять концепцию обратного давления, параметр highWaterMark и четыре основные класса потоков. Эти базовые знания остаются обязательными прежде чем любые из вышеупомянутых методов станут полезными. Однако как только у вас есть эта основа, начинается настоящая инженерная работа: подтверждение улучшений в использовании памяти с помощью конкретных цифр, а не косвенных утверждений, память о том, что потоку, основанному на методе fetch, и Node-потоку для взаимодействия требуется всего одна конверсия с помощью метода .toWeb(), а также разработка плана на случай отключения клиента в процессе запроса как стандартной части реализации маршрута, а не детали, которую добавляют позже.
Если вы запустите приведённые ранее примеры тестирования на разном оборудовании и результаты существенно отличатся от тех, что показаны здесь, это будет полезным данным, которые стоит поделиться, а не игнорировать, поскольку показатели пропускной способности и использования памяти становятся гораздо надёжнее после проверки на нескольких устройствах.
Ресурсы
- Документация Node.js, посвящённая классическому API Stream: https://nodejs.org/api/stream.html
- Документация Node.js, посвящённая API Web Streams: https://nodejs.org/api/webstreams.html
- Справка Ollama по его концу для стриминга: https://docs.ollama.com/api/streaming
- Список пакета
@aws-sdk/lib-storageв npm: https://www.npmjs.com/package/@aws-sdk/lib-storage - Репозиторий проекта
busboy: https://github.com/mscdex/busboy