Node.js стрімы за межамі базавых прыемаў: памяць, зворотны тырчын і рэальныя неудачы

Большая частка введэнняў у стрымы Node.js добра раскладваюць базавыя аспекты: чытанне, запіс, двунаправленая праця, трансформацыя, pipe(), эфект зворачнага тиску, дэманстрацыя архівавання у формате gzip, а інодзе — сервер для запытак пра паведамленні ў відпаведным дыяпазоне як фінал. Якщо вы вже працавалі з аднам з такіх матэрыялаў, вы розумееце, што такое стрым і чаму прабава запісаць 2 ГБ файл увесь у памяць пад час жывай дэманстрацыі є прычыной проблем.

У тых вступленнях рэдкая час згадваецца тое, што выйшае далей. Някія памяць фактычна заўсёды захоўваецца праз пераход на стрімы, з конкрэтнымі цифрамі, а не простаю фразай «менш»? Як натыўныя стрімы Node взаімаюцца з Web Streams API, які выкарыстоўваецца пры fetch і браузерах? Як можна передаваць выходны даны з модэлю, які працуе на вашай сабе апаратурэ, замест таго, каб плаціць за кожны токен через хоставаны API? І якія два-тры прычыны ў канцэ нарабляюць проблемы ў ланцугу, які пад час тэставання выглядаў нормальна, калі стыкаецца з рэальным трафікам?

У гэтым артыкуле розглядаецца самэ гэта. У яму прыпускаецца, што вы вялікі час ведаеце, шта такое readable stream. Якщо гэтага няма, короткае падсумованне нижэй дапаможае быстра адразу зрозумець сутнасць. Усё, што будзе після, будуе на адной базе з гэтыми фундаментальнымі панясцямі.

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(); у ныях статус стабільнасці 2 ("Stable") па данным версіям Node.js.

Гэта становіць значэнне ў той момент, калі вы вызываеце 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'. Гэта неабходна, таму што тэлы запита у формате стріму не ўсюды падтрымваюцца аднакова, таму стандарт выказывае трэбаць явнага выбору. Це незначная проблема, якая займае падзеў панцыхоў паўгодзіны і больш ніколи не стварае прычын для проблем.

Асалоджэнне заключаецца у тым, што ваш 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, які быў паказаны вышэй, працюе з яй без жадных змян, за выключэнням перакіравання кліента да канцэнтра МінІО заместо 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', на prom-client Counter або Histogram. Сам засіб залишаецца тым жа: цэльва ўсё той самы поток, які знаходзится сярод процесу і фіксуе тое, што праходзіць чераз яго.

Тры помылкі, якія пазіраюць толькі пад навантажэнням

Некарыстоўваныя з’язды 'error' прыводзяць да заканчэння цэлага процесу. Стрэм, який выдае памылку без прыўязанага слухальніка 'error', не толькі парабоюе свой сабэтны запит. У Node.js з’язд 'error' у об’екте EventEmitter, які ніхто не карыстоўваецца, выкалываеся, і якщо нічога не ловіць гэты выкал, процес завершваецца. Дастатнья ўсьмо толькі адна некоректная загрузка, каб усё служба застала без роботы, адтак як ручна сформаваная лянцюг pipe() мела слухальнікі памылак на два з трох стрэмов. Функцыя pipeline() існуе самэ прычым, каб такія багі былі структурна складныя для выканання, таму лепш яе выкарыстоўваць, альбо пераканацца, што кожны стрэм, які вы ствараеце ручна, мае свой сабэтны .on('error', ...), без жадных асаблівэй.

Павтарэнне выкарыстоўвання экзампляра стрыму прыводзіць да вытэкання слухачаў. Падазроўваючы, што стварэнне стрыму адбываецца ўнутры функцыі, якая вызываецца з кожным запитам, але якась частка паza гэтай функцыяю заставляе рэферэнс да яго і прыкладзае своіх слухачаў „на всякы случай“. У канцэ на вашых лагах з’явіцца паведамленне MaxListenersExceededWarning, якое ў Node ёсць спосабам адзначэння працюючага вытэкання памяці. Кожны запит патрэбуе своіяя, што ствараныя ўпершыню стрымы, і нічто паza трымкам часу гэтага запиту не должна заставляць рэферэнс да яных пасля таго, как запит завершыўся.

Ніхто не знішчае стрым, калі кліент пакінуў сайт. Гэта тая памылка, якая насправдзе коштуе грошаў у працэсе выработкі. Якщо корыстнік закрывае вікно браузера пад час завантажэння, а вы не стежыце за сігналам 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 такім спосабам раней здавалася незвычайным; сёньня гэта досыць звычайны патэрн.

Заключныя мыслі

Нічыта тут не пазбярэляе патрабаванне спачатку разумець, што такое backpressure, highWaterMark і чатыры фундаментальныя класы стрымоў. Гэтыя базовыя знанні застаюцца абавязковымі прыёмамі, перш чым які-небудзь з вышэўказаных методаў стане корыстным. Калі ж у вас ёсць гэты фундамент, пачынаецца справжняі інжынерны работа: падтверджэнне павышэння эфектывасці выкарыстоўвання памяці на аднойчы змераных цифрах, а не толькі расплывчатыя тверджэння; памяць, што стрым, заснованы на fetch, і стрым Node патрэбуюць толькі адной пераработкі .toWeb(), каб можна было ўзаўмна працаваць; а таксама стварэнне плану на случай, калі кліент ад’ўяжаецца падчас запыту, як стандартнай часткі напісання маршруту, а не як деталі, яку пасля таго трэба латаваць.

Якщо вы запустіце прыклады тэставання, упомянутыя раней, на разным апаратнам забезпечэнні і рэзультаты значна адвярнуцца ад тых, што паказаны тут, то гэта будзе корисны даны, якія варта паделіцца, а не ўсё, што трэба ігнараваць, адтаколькі цифры ў часе обробкі дадзеных і выкарыстоўванні памяці стаюць надзейнымі, калі іх перакантролюваць на больш чым адной машыне.

Рэсурсы

• Дакументацыя Node.js, якая раскрывае класычны Stream API: https://nodejs.org/api/stream.html
• Дакументацыя Node.js, якая раскрывае Web Streams API: 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

Сайт проекта MinIO: https://min.io/