Node.js Streams Performance Backpressure JavaScript

Node.js Streams: більше, ніж основи — пам’ять, зворотний тиск та справжні помилки

Дізнайтеся, як стріми Node.js взаємодіють із Web Streams, яку економію пам’яті можна досягти за допомогою реальних тестів, а також про помилки у продакшені, які проявляються лише під навантаженням.

Видавець: REACTAPP.TOPДодано на сайт 22 вер. 2026 р.18 хв читання4104 слів

Джерела: 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 });
});

Ось що насправді вийшло після тестування цих підходів на файлі розміром 500 МБ у форматі ASCII за допомогою 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 (“Stable”).

Це стає актуальним у момент, коли ви викликаєте 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, наведений вище, працює з ним без жодних змін, окрім вказівки клієнту на кінцеву точку 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

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

← Усі статті Надіслати бриф