Strumienie w Node.js poza podstawami: pamięć, ciśnienie zwrotne i rzeczywiste błędy

Wiele wprowadzeń do strumieni w Node.js dobrze omawia podstawy: czytelność, zapisywalność, tryb dwukierunkowy, transformacje, funkcję pipe(), zasady backpressure, demonstrację kompresji gzip, a czasami nawet serwer obsługujący żądania o określony zakres wideo jako finał. Jeśli już przeczytałeś jedno lub dwa z takich materiałów, rozumiesz, czym jest strumień i dlaczego próba załadowania całego pliku o wielkości 2 GB do pamięci podczas demonstracji na żywo prowadzi do problemów.

To, czego rzadko dotyczą te wprowadzenia, to to, co następuje dalej. Ile pamięci faktycznie oszczędzasz, przechodząc na strumienie, przy konkretnych liczbach zamiast niejasnego stwierdzenia „mniej”? Jak strumienie natywne Node współpracują z API Web Streams używanym przez fetch i przeglądarki? Jak możesz przesyłać wyniki z modelu działającego na twoim sprzęcie zamiast płacić za każdy token przez hostowaną API? A które dwa lub trzy błędy ostatecznie niszczą proces, który wydawał się poprawny podczas testów, gdy napotyka rzeczywisty ruch?

To artykuł właśnie o tym. Zakłada on, że już wiesz, czym jest strumień do odczytu. Jeśli tak nie jest, krótki podsumowanie poniżej szybko cię na bieżąco wprowadzi. Wszystko, co następuje, opiera się na tych podstawach.

Podsumowanie w 60 sekund

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

Dane przepływają przez system w małych fragmentach, a nie jako jedna duża całość. Funkcja pipe() łączy strumienie ze sobą i radzi sobie z sytuacją, gdy destynacja nie jest w stanie nadążyć za źródłem – to właśnie tzw. backpressure. Funkcja pipeline() realizuje takie same połączenia, ale także przekazuje błędy z każdego strumienia w łańcuchu oraz gwarantuje sprzątanie po awarii, dlatego jest to funkcja, której należy używać w każdym długotrwałym serwisie, a nie w szybkim, jednorazowym skrypcie.

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')
);

Jeśli ten przykład jest już dla ciebie całkowicie zrozumiały, możesz przejść dalej. To, co następuje, to informacje istotne do poznania przed pierwszym umieszczeniem kodu opartego na strumieniach w środowisku produkcyjnym.

Liczba, której nikt nie podaje w tych artykułach

Większość tekstów na temat przepływu danych twierdzi, że podejście z buforowaniem „zużywa dużo pamięci”, podczas gdy podejście przepływowe „zużywa znacznie mniej”, nie podając przy tym żadnych konkretnych liczb. Aby to naprawić, utworzono plik tekstowy o wielkości 501MB i oba podejścia przetestowano na tym samym komputerze, kolejno po dwa razy każde.

// 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 });
});

Oto wyniki rzeczywistych testów tych podejść na pliku ASCII o wielkości 500MB przy użyciu Node.js 22, po dwa wykonywania dla każdego podejścia:

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

Należy jasno podkreślić kilka zastrzeżeń, ponieważ ich pominięcie mogłoby wprowadzić w błąd: chodzi tu o pojedynczą maszynę, pojedynczy plik oraz pojedyncze obciążenie — prosty skan znaków. Czas wykonywania w czasie rzeczywistym okazał się mniej więcej taki sam w obu podejściach, przy czym wersja strumieniowa była nawet miejscami nieco szybsza, co było nieco zaskakujące. Założeniem wyjściowym było, że obsługa wielu małych fragmentów spowolni proces bardziej, niż to się stało. Z kolei wartości pamięci doskonale odpowiadały oczekiwaniom: wersja z buforowaniem osiągnęła maksimum bliskie pełnej wielkości pliku, natomiast wersja strumieniowa osiągnęła niewielką wartość, która pozostawała w dużej mierze stała niezależnie od rozmiaru pliku. To właśnie ten kontrast stanowi całą rację używania strumieni, zebraną w jednej tabeli — a jej poprawność można łatwo zweryfikować samodzielnie, uruchamiając oba powyższe skrypty na danym pliku.

swojego.

Praktyczne implikacje: w przypadku pojedynczej prośby obsługiwanej przez wystarczająco mocny komputer, samo odczytanie całego pliku do pamięci często wystarcza. Prawdziwym problemem nigdy nie była pojedyncza prośba w izolacji — chodzi o pytanie, które zawsze pojawia się później: co się stanie, gdy 100 takich próśb będzie działać jednocześnie? Zużycie pamięci w podejściu strumieniowym prawie się nie zmienia wraz ze wzrostem równoczesności, ponieważ każda pojedyncza prośba przechowuje jedynie bufor o określonej wielkości, a nie całą zawartość naraz.

Node Streams i Web Streams należą teraz do tej samej rodziny

Przez lata strumienie w Node.js oraz API Streams WHATWG (ReadableStream, WritableStream, TransformStream — klasy stanowiące podstawę funkcji fetch() zarówno w przeglądarce, jak i w Node) wydawały się dwoma odrębnymi ekosystemami z niewielkim nakładaniem się. To już nie jest prawdą. Moduł node:stream/web umożliwia bezpośrednie korzystanie z klas WHATWG, a własne klasy Node Readable, Writable oraz Duplex posiadają metody .toWeb() i .fromWeb(); od najnowszych wersji Node.js są one oznaczone jako stabilne (Stability 2 – „Stable”).

Sytuacja ta staje się istotna w momencie, gdy wywołujemy fetch() z wnętrza usługi Node, ponieważ response.body, który otrzymujemy, to strumień typu Web ReadableStream, a nie 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');

Konwersja działa równie płynnie w odwrotnym kierunku — można przekształcić obiekt Node Readable w strumień, który przyjmie każda API oparta na 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
});

Jeden szczegół, który myli ludzi przy pierwszym użyciu: funkcja fetch rzuci błąd, jeśli przekażesz jej ciało żądania w formie strumienia bez jednoczesnego podania parametru duplex: 'half'. Takie rozwiązanie istnieje, ponieważ ciała żądań w formie strumienia nie są wszędzie równomiernie obsługiwane, więc specyfikacja wymaga wyraźnego wskazania tej opcji. To drobna kwestia, która zajmuje piętnaście minut raz i już nigdy więcej.

Prawdziwą korzyścią jest to, że backend w Node, funkcje edge oraz kod w przeglądarce mogą teraz w dużej mierze korzystać z tej samej logiki strumieniowania, zamiast zmuszać do utrzymywania wersji opartej na fetch oraz oddzielnej wersji bazującej na funkcji pipe() i modułach fs oraz http w Node.

stream.compose(): pipeline, który zwraca jeden strumień

stream.compose() istnieje technicznie od wersji Node.js 16.9.0, ale przez długi czas pozostawał w stanie eksperymentalnym, zanim ostatecznie stał się stabilny w najnowszej wersji Node.js. Teraz, gdy jego API jest ustalone, zasługuje na bliższe przyjrzenie się.

Idea polega na tym, że zamiast tworzyć pełną podklasę Transform z metodami _transform i _flush, piszemy funkcję generatoryczną asynchroniczną, a compose() łączy cały ten łańcuch w jeden strumień Duplex, który można przekierowywać tak samo jak każdy inny strumień.

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'));

Jeśli oprócz napisania dwóch oddzielnych klas Transform — w tym logiki buforowania ręcznego dla części odpowiedzialnej za dzielenie wierszy — ustawisz to, uzyskasz takie samo podstawowe zachowanie przy znacznie mniej kodzie pomocniczym, a ponadto kod, który jest czytany od góry do dołu w dokładnej kolejności, w jakiej faktycznie przepływają dane. Funkcja compose() jest również elastyczna pod względem swoich końcówek: jeśli jako pierwszy argument podasz zwykły obiekt AsyncIterable, stanie się on czytelnym obiektem Duplex; jeśli jako ostatni argument podasz AsyncFunction, stanie się pisalnym zbiornikiem danych. Oznacza to, że cały pipeline może zostać stworzony wyłącznie z generatorów asynchronicznych, jeśli taki styl jest Ci bliższy.

Żadne z tych powodów nie uzasadnia usuwania już działających klas Transform. Jednak w przypadku nowego kodu — szczególnie takiego, który sprowadza się do „przetwarzania tego, filtrowania tamtego, kształtowania tego” — użycie generatora asynchronicznego w połączeniu z compose() pomaga uniknąć pisania kilku małych podklas Transform, które ostatecznie mają identyczny wygląd.

Strumieniowanie tokenów z lokalnego modelu

Prawie każdy rozdział o „prawdziwych strumieniach” w tutorialach opiera się na tych samych trzech przykładach: pobieraniu plików, kompresji gzip oraz żądaniach o określonym zakresie wideo. Są to uzasadnione przypadki użycia, ale istnieje nowszy wzorzec, który równie dobrze pasuje do modelu strumieniowego i rzadko jest omawiany: pobieranie wyników token po tokenie z dużego modelu językowego.

Ponieważ przykład rzeczywiście nadający się do skopiowania i wklejenia nie powinien polegać na kluczu API kogokolwiek, ten poniżej komunikuje się z Ollamą uruchomionym lokalnie, a nie z usługodawcą hostowanym online:

# one-time setup
ollama pull llama3.2
ollama serve   # usually already running as a background service

Domyślnie punkt końcowy /api/generate w Ollamie przesyła swoją odpowiedź jako JSON oddzielone nowymi liniami (application/x-ndjson): jeden obiekt JSON na linię, z których każdy zawiera fragment response oraz flagę done, przy czym ostatnia linia informuje o "done": true. Po usunięciu formatowania jest to po prostu czytelny strumień wykonujący proste parsowanie tekstu — ten sam mechanizm oparty na fragmentach, używany we wszystkich innych miejscach w tym artykule, może być zastosowany bez żadnych modyfikacji.

// 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'));

Jeśli chcesz przejrzeć surową linię NDJSON od Ollamy bez otaczającej ją ramy SSE, wygląda ona w ten sposób:

{"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"}

Dwa aspekty mogą wprowadzić w błąd osoby po raz pierwszy. Po pierwsze, we współczesnych wersjach Node.js response.body z funkcji fetch jest już sam w sobie możliwy do iteracji asynchronicznej, więc nie ma potrzeby wywoływania Readable.fromWeb() tylko w tym przypadku – chociaż ta funkcja pomocnicza nadal jest dostępna, jeśli jakaś późniejsza biblioteka oczekuje klasycznego obiektu Readable z Node. Po drugie, linie w formacie NDJSON mogą obejmować kilka granic fragmentów, co stanowi dokładnie ten sam problem, jaki pojawia się podczas parsowania plików CSV w formie strumieniowej: nie ma gwarancji, że jeden fragment uzyskany za pomocą fetch będzie odpowiadał jednej linii JSON. Właśnie dlatego powyższy przykład przechowuje w buforze niekompletne linie, zamiast bezpośrednio wywoływać JSON.parse(chunk) na otrzymanym fragmencie.

Jeśli wolisz korzystać z API hostowanego — OpenAI, Anthropic lub podobnych — ogólna struktura prawie się nie zmienia: nadal otrzymujesz odpowiedź HTTP w formie fragmentów i przekazujesz ją do klienta w formacie SSE. Lokalna wersja Ollamy to po prostu ta, którą możesz uruchomić natychmiast bez konieczności posiadania konta czy klucza, i stanowi rozsądną opcję do budowy strony klienta takiej funkcji w trakcie rozwoju, zanim skierujesz ten sam kod na model hostowany, gdy będzie potrzebna jakość wyników na poziomie produkcyjnym.

Przepływ danych pliku bezpośrednio do przechowywania obiektów

Wysyłanie pliku „do S3” prawie nigdy nie polega na użyciu pliku na tyle małego, by swobodnie zmieścić się w pamięci przed realizacją operacji. Pakiet @aws-sdk/lib-storage istnieje właśnie dlatego, że zwykły PutObjectCommand wymaga znania długości treści przed rozpoczęciem żądania. Jego pomocnik Upload przyjmuje zamiast tego strumień, sam decyduje, czy odpowiednie jest jedno żądanie, czy transfer wieloczęściowy, oraz informuje o postępach w trakcie procesu.

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`);
}

Aby eksperymentować bez konieczności posiadania rzeczywistego konta AWS, MinIO oferuje magazyn obiektów kompatybilny z S3, który można uruchomić lokalnie za pomocą jednego polecenia. Dokładnie ten sam kod @aws-sdk/lib-storage z powyższego działa z MinIO bez żadnych zmian, poza skierowaniem klienta na endpoint MinIO zamiast na 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' }
});

To jest podejście, które warto stosować dla plików większych niż kilkadziesiąt megabajtów: unikaj przetwarzania całego pliku do formatu Buffer przed przekazaniem go do SDK. Użyj funkcji createReadStream() i pozwól narzędziu Upload zająć się resztą — zużycie pamięci pozostaje stałe, niezależnie od tego, czy plik ma rozmiar 10 MB, czy 10 GB.

Analiza pliku przesyłanego do serwera bez buforowania ciała żądania

Z drugiej strony przesyłania plików jest serwer, który musi je otrzymać, i to właśnie tutaj najczęściej występuje przypadkowe buforowanie — zwykle dlatego, że middleware do analizy ciała żądania nie został zaprojektowany z myślą o przepływie danych i w tajemnicy odczytał całe żądanie typu multipart do pamięci, zanim Twój obsłużający skrypt miał szansę je przetworzyć.

busboy przetwarza formularze wieloczęściowe stopniowo: gdy tylko zaczynają napływać bajty dla danego pola pliku, zwraca dostępny do odczytu strumień dla tego pola, zamiast czekać, aż cała prośba dotrze w całości:

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);

W tym mechanizmie przetwarzania nigdy nie przechowuje się całości ciała żądania naraz. Bajty przemieszczają się z gniazdka sieciowego, przez parsera busboy’a, i bezpośrednio na dysk. To właśnie ta różnica odróżnia usługę, która dobrze radzi sobie z kilkoma dużymi plikami przesyłanymi równolegle, od takiej, która zawodzi już przy pierwszym przypadku, gdy dwóch użytkowników jednocześnie przesyła wideo.

Testowanie niestandardowego strumienia transformacji

Ręcznie tworzone strumienie Transform są głównymi kandydatami do cichego awariowania: nikt nie zadaje sobie trudu, by pisać dla nich testy, wszyscy zakładają, że „po prostu działają”, a potem przypadek krawędziowy _flush zostaje wdrożony w uszkodzonej formie, a nikt tego nie zauważa aż do chwili uruchomienia w produkcji. Testowanie takiego strumienia jest jednak mniej uciążliwe, niż się wydaje. Nie musisz tworzyć rzeczywistego pliku ani prawdziwego serwera HTTP. Readable.from() w połączeniu z stream/promises pomaga to osiągnąć.

// 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');
  });
});

Drugi test w tym pliku ma większe znaczenie, niż się na pierwszy rzut oka wydaje. Potwierdza on, że błąd powstały w górnej części łańcucha rzeczywiście się rozprzestrzenia, zamiast sprawić, że strumień transformacji utknie w połowie pracy z obietnicą, która nigdy się nie spełnia. To ciche zatrzymanie jest najczęstszym sposobem, w jaki łańcuch operacji, który wydaje się działać, zamienia się w zablokowaną prośbę po uruchomieniu.

Lekki narzędzie do pomiarów

Nie potrzebujesz pełnej platformy obserwowalności tylko po to, by sprawdzić, jak funkcjonuje pipeline. Często wystarczy mały strumień PassThrough, który liczy bajty i fragmenty danych w miarę ich przepływu – taki strumień można łatwo włączyć do istniejącego pipeline bez zakłócania działania pozostałych elementów:

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')
);

Jeśli chcesz czegoś więcej niż wyjście typu console.log, zastąp logikę znajdującą się w obsługiwanach zdarzeń 'data' i 'end' obiektem Counter lub Histogram z biblioteki prom-client. Sam mechanizm pozostaje niezmieniony – to po prostu kolejny strumień umieszczony w pipeline, który raportuje o tym, co przez niego przechodzi.

Trzy błędy, które pojawiają się tylko pod obciążeniem

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.

Ponowne użycie instancji strumienia powoduje wyciek słuchaczy. Załóżmy, że tworzenie strumienia umieszczamy w funkcji wywoływanej przy każdej prośbie, ale coś poza tą funkcją przechowuje do niej referencję i dodaje własne słuchacze „na wszelki wypadek”. W końcu zobaczysz w swoich logach komunikat MaxListenersExceededWarning, który jest sposobem Node na zaznaczenie trwającego wycieku pamięci. Każda prośba wymaga swoich świeżo utworzonych strumieni, a nic poza czasem trwania tej prośby nie powinno przechowywać do nich referencji po jej zakończeniu.

Nikt nie niszczy strumienia, gdy klient odchodzi. To właśnie ten błąd powoduje straty finansowe w produkcji. Jeśli użytkownik zamknie kartę przeglądarki w trakcie pobierania, a nie monitorujesz zdarzenia response.on('close', ...), aby wywołać metodę .destroy() na źródłowym strumieniu, serwer nadal pobiera dane z dysku, a w najgorszym przypadku pozostawia otwarty cursor bazy danych lub utrzymuje aktywną połączenie HTTP, wszystko to dla klienta, który już odszedł. Przy wystarczająco dużej liczbie jednoczesnych żądań w ten sposób serwer może po cichu wyczerpać swoje deskrypcje plików lub zasoby połączeń, bez wyraźnej przyczyny widocznej w logach. Rozwiązanie zawsze polega na tych samych trzech krokach: monitorowaniu zamknięcia celu, niszczeniu źródła w odpowiedzi oraz traktowaniu tego schematu jako standardowej praktyki na każdej trasie opartej na strumieniach, a nie jako czegoś, co pamięta się dopiero po incydencie.

Kiedy użyć której 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.

Żaden z tych podejść nie czyni pozostałych zbędnymi. Każde z nich rozwiązuje inny aspekt tego samego ogólnego przepływu danych w systemie. Obecnie powszechnym jest to, że pojedynczy pipeline zaczyna się od Web ReadableStream uzyskanego za pomocą fetch, konwertuje te dane, przekazuje je przez transformację zbudowaną za pomocą compose(), a na końcu przekazuje wynik do pipeline() w celu zapisania. Mieszanie API w ten sposób kiedyś wydawało się nietypowe; dziś jest to dość powszechny wzorzec.

Podsumowanie

Nic tutaj nie eliminuje konieczności najpierw zrozumienia koncepcji backpressure, highWaterMark oraz czterech podstawowych klas strumieni. Ta wiedza bazowa pozostaje obowiązkowym warunkiem przed tym, jak jakakolwiek z powyższych technik może być przydatna. To, co się zmienia po zdobyciu tej wiedzy, to moment, w którym rozpoczyna się prawdziwa praca inżynierska: potwierdzanie ulepszeń pamięci na podstawie dokładnych danych, a nie ogólnikowych stwierdzeń, pamiętanie o tym, że strumień oparty na fetch oraz strumień Node wymagają tylko jednej konwersji .toWeb(), aby mogły ze sobą współpracować, oraz opracowywanie planu na wypadek rozłączenia się klienta w trakcie żądania jako standardowej części pisania ścieżki, a nie szczegółu, który dodaje się później.

Jeśli uruchomisz wcześniej wspomniane przykłady testów na różnym sprzęcie, a wyniki znacznie odbiegną od tych pokazanych tutaj, jest to przydatny punkt danych wart udostępnienia, a nie coś, co należy zignorować, ponieważ liczby dotyczące przepustowości i zużycia pamięci są znacznie bardziej wiarygodne po sprawdzeniu na więcej niż jednym urządzeniu.

Zasoby

• Dokumentacja Node.js dotycząca klasycznej API Stream: https://nodejs.org/api/stream.html
• Dokumentacja Node.js dotycząca API Web Streams: https://nodejs.org/api/webstreams.html
• Dokumentacja Ollamy dotycząca jej endpointu do streamowania: https://docs.ollama.com/api/streaming
• Spis pakietu @aws-sdk/lib-storage na npm: https://www.npmjs.com/package/@aws-sdk/lib-storage
• Repozytorium projektu busboy: https://github.com/mscdex/busboy

Strona projektu MinIO: https://min.io/