Node.js Streams jenseits der Grundlagen: Speicher, Rückdruck und tatsächliche Ausfälle
Erfahren Sie, wie Node.js Streams mit Web Streams interagieren, die durch tatsächliche Benchmarks gemessenen Speichereinsparungen sowie die Produktionsfehler, die nur unter Last auftreten.
Quellen: Node.js Streams in Production: The Parts Most Tutorials Skip
Viele Einführungen zu Node.js Streams behandeln die Grundlagen gründlich: Lesbarkeit, Schreibfähigkeit, Duplex-Modus, Transformation, pipe(), Rückdrucksteuerung, eine gzip-Beispielanwendung und manchmal auch einen Video-Range-Request-Server als krönenden Abschluss. Wenn Sie bereits eine oder zwei dieser Einführungen durchgearbeitet haben, wissen Sie, was ein Stream ist – und warum es problematisch ist, während einer Live-Demo eine 2 GB große Datei vollständig in den Speicher zu laden.
Was in solchen Einführungen selten behandelt wird, ist das, was danach kommt. Wie viel Speicherplatz spart man tatsächlich durch den Umstieg auf Streams – mit konkreten Zahlen statt einer vagen Angabe von „weniger“? Wie interagieren Nodes native Streams mit der Web Streams API, die von fetch und Browsern verwendet wird? Wie kann man Ausgaben eines auf dem eigenen Hardware laufenden Modells streamen, anstatt pro Token über eine gehostete API zu zahlen? Und welche zwei oder drei Fehler führen letztendlich dazu, dass ein in Tests gut funktionierender Pipeline unter echtem Traffic zusammenbricht?
Dieser Artikel behandelt genau das. Er setzt voraus, dass Sie bereits wissen, was ein readable Stream ist. Falls nicht, wird die kurze Zusammenfassung unten Ihnen schnell auf den neuesten Stand bringen. Alles, was folgt, baut auf diesen Grundlagen auf.
Die 60-Sekunden-Zusammenfassung
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
Daten fließen im System in kleinen Teilen statt als ein großer Datensatz. pipe() verbindet Ströme miteinander und kümmert sich um die Situation, in der das Ziel nicht mit der Quelle Schritt halten kann – das ist Rückdruck. pipeline() erreicht dieselbe Verbindung, überträgt aber auch Fehler aus jedem Stream in der Kette und stellt sicher, dass Ressourcen freigegeben werden, wenn etwas fehlschlägt. Deshalb ist es die Funktion, die man in jedem langlaufenden Service benötigt und nicht in einem schnellen Einmal-Skript.
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')
);
Falls Ihnen dieses Beispiel bereits vollkommen klar ist, können Sie gerne weiterblättern. Was folgt, sind die Informationen, die man kennen sollte, bevor man zum ersten Mal streambasierten Code in eine Produktionsumgebung einsetzt.
Die Zahl, die in diesen Artikeln nie erwähnt wird
Die meisten Artikel zu Streams behaupten, der gepufferte Ansatz „verwende viel Speicher“, während der Streaming-Ansatz „weitaus weniger“ verbrauche, ohne jemals eine Zahl zu diesen Aussagen anzugeben. Um das zu beheben, wurde eine 501MB große Textdatei erstellt, und beide Ansätze wurden auf demselben Rechner nacheinander zweimal ausgeführt.
// 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 });
});
Hier ist das Ergebnis der Ausführung auf einer 500MB großen ASCII-Datei mit Node.js 22, jeweils zwei Durchläufe pro Ansatz:
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
Mehrere wichtige Hinweise, die klar erwähnt werden sollten, da ihr Auslassen zu irreführenden Ergebnissen führen würde: Es handelt sich dabei um einen einzigen Rechner, eine einzige Datei und eine einzige Arbeitslast – eine einfache Zeichensequenz-Überprüfung. Die reine Laufzeit war bei beiden Ansätzen ungefähr gleich, wobei die streambasierte Version an einigen Stellen sogar leicht vorn lag, was etwas überraschend war. Die ursprüngliche Annahme war, dass die Verarbeitung vieler kleiner Datenblöcke die Geschwindigkeit stärker verlangsamen würde, als tatsächlich der Fall war. Die Speicherverbrauchsangaben entsprachen hingegen weitgehend den Erwartungen: Der bufferbasierte Ansatz erreichte einen Höchstwert nahe am vollen Dateigröße, während der streambasierte Ansatz einen geringen Wert erreichte, der unabhängig von der Dateigröße größtenteils konstant blieb. Dieser Kontrast fasst im Grunde den gesamten Nutzen von Streams zusammen – und er lässt sich leicht unabhängig überprüfen, indem man die beiden obigen Skripte auf einer Datei ausführt.
Ihr eigenes.Die praktische Folge: Für eine einzelne Anfrage, die von einem ausreichend leistungsstarken Rechner bearbeitet wird, reicht es oft bereits aus, die gesamte Datei in den Speicher zu lesen. Das eigentliche Problem lag nie in einer einzelnen Anfrage für sich genommen – es ist vielmehr die anschließende Frage, die immer stellt: Was passiert, wenn 100 solcher Anfragen gleichzeitig laufen? Der Speicherverbrauch beim Streaming-Ansatz ändert sich kaum mit zunehmender Parallelität, denn jede einzelne Anfrage enthält lediglich einen buffergroßen Teil des Datenstroms und nicht die gesamte Ladung auf einmal.
Node Streams und Web Streams gehören jetzt zur gleichen Familie
Jahrelang wirkten die Node.js Streams sowie die WHATWG Streams API (ReadableStream, WritableStream, TransformStream – die Klassen hinter fetch() sowohl im Browser als auch in Node) wie zwei getrennte Ökosysteme mit nur geringem Überschneidungsbereich. Das ist nicht mehr zutreffend. Die node:stream/web-Modul stellt die WHATWG-Klassen direkt zur Verfügung, und Nodes eigene Readable, Writable sowie Duplex-Klassen verfügen alle über die Methoden .toWeb() und .fromWeb(), die in den aktuellen Node.js-Versionen als Stabilitätsstufe 2 ("Stable") gekennzeichnet sind.
Dies wird relevant, sobald man fetch() von innerhalb einer Node-Dienstleistung aufruft, da response.body, das zurückgegeben wird, ein Web-ReadableStream und kein Node-Readable ist.
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');
Die Umwandlung funktioniert genauso reibungslos in die andere Richtung – Sie können einen Node Readable-Objekt in einen Stream umwandeln, den jede API, die auf Web-Streams basiert, akzeptieren wird:
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
});
Ein Detail, das viele Anfänger zunächst verwirrt: fetch wirft einen Fehler, wenn man ihm ein streambares Anfrage-Body ohne gleichzeitige Angabe von duplex: 'half' übergibt. Dies liegt daran, dass streambare Anfrage-Body nicht überall einheitlich unterstützt werden, weshalb die Spezifikation eine explizite Aktivierung erfordert. Es handelt sich um eine Kleinigkeit, die einmal fünfzehn Minuten Zeit kostet und danach nie wieder Probleme verursacht.
Der eigentliche Vorteil ist, dass Ihr Node-Backend, Ihre Edge-Funktionen sowie Ihr Browsercode nun größtenteils dieselbe Streaming-Logik teilen können, anstatt dass Sie eine fetch-basierte Version sowie eine separate, auf pipe() basierende Version, die auf Node’s fs- und http-Modulen aufbaut, pflegen müssen.
stream.compose(): ein Pipeline, das einen Stream zurückgibt
stream.compose() existiert technisch gesehen bereits seit Node.js 16.9.0, war jedoch über einen langen Zeitraum im experimentellen Status, bevor es in einer neueren Node.js-Version endlich stabil wurde. Da seine API nun festgelegt ist, verdient es eine genaue Betrachtung.
Die Idee: Anstatt eine vollständige Transform-Unterklasse mit den Methoden _transform und _flush zu schreiben, erstellt man eine asynchrone Generatorfunktion, wobei compose() die gesamte Kette dieser Funktionen in einen einzigen Duplex-Stream zusammenfasst, den man genauso wie jeden anderen Stream weiterleiten kann.
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'));
Wenn man neben der Erstellung zweier separater Transform-Klassen – einschließlich einer manuellen Pufferlogik für den Teil, der Zeilen trennt – auch dies berücksichtigt, erhält man dasselbe zugrundeliegende Verhalten mit deutlich weniger Struktur, außerdem Code, der von oben nach unten in derselben Reihenfolge gelesen wird, in der die Daten tatsächlich fließen. compose() ist zudem in Bezug auf seine Endpunkte flexibel: Geben Sie ihm einen einfachen AsyncIterable als erstes Argument, wird daraus ein leicht verständlicher Duplex; geben Sie ihm eine AsyncFunction als letztes Argument, wird daraus ein schreibbarer Empfänger. Das bedeutet, dass eine ganze Pipeline ausschließlich aus asynchronen Generatoren aufgebaut werden kann, falls man diesen Stil bevorzugt.
Nichts davon ist ein Grund, bereits funktionierende Transform-Klassen zu entfernen. Für neuen Code – insbesondere für alles, was im Grunde darin besteht, „dies zu parsen, jenes zu filtern und das wiederum umzuformen“ – führt die Verwendung eines asynchronen Generators in Kombination mit compose() in der Regel dazu, dass man nicht mehrere kleine Transform-Unterklassen schreiben muss, die letztendlich alle identisch aussehen.
Token streaming aus einem lokalen Modell
Fast jeder Abschnitt zu „Reallife-Streams“ in Tutorials verwendet immer wieder dieselben drei Beispiele: Dateiherunterlade, gzip-Kompression und Anfragen nach Videoabschnitten. Das sind durchaus legitime Anwendungsfälle, doch es gibt ein neueres Muster, das genauso gut zum Streaming-Modell passt und nur selten behandelt wird: Das Abrufen von Ausgaben in Form von Token für Token aus einem großen Sprachmodell.
Da ein wirklich kopier- und einfügbares Beispiel nicht von jemandes API-Schlüssel abhängen sollte, kommuniziert das untenstehende Beispiel mit Ollama, das lokal läuft, anstatt mit einem gehosteten Anbieter:
# one-time setup
ollama pull llama3.2
ollama serve # usually already running as a background service
Standardmäßig streamt Ollamas /api/generate-Endpunkt seine Antwort als mit Zeilenumbrüchen getrenntes JSON (application/x-ndjson): ein JSON-Objekt pro Zeile, wobei jedes Objekt einen response-Fragment sowie ein done-Flag enthält, wobei die letzte Zeile "done": true angibt. Wenn man die Formatierung entfernt, handelt es sich dabei um einen lesbaren Stream, der eine einfache Textanalyse durchführt – die gleiche auf Blöcken basierende Logik, wie sie sonst in diesem Artikel verwendet wird, kann unverändert übernommen werden.
// 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'));
Falls Sie eine rohe NDJSON-Zeile von Ollama ohne das darum gewickelte SSE-Framing untersuchen möchten, sieht sie so aus:
{"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"}
Zwei Aspekte führen bei Erstnutzung zu Verwirrung. Erstens sind in aktuellen Node.js-Versionen fetch's response.body bereits von selbst asynchron iterierbar, weshalb es nicht notwendig ist, in diesem Fall Readable.fromWeb() aufzurufen – obwohl dieser Hilfsfunktion weiterhin verfügbar ist, falls eine nachgelagerte Bibliothek ein klassisches Node Readable-Objekt erwartet. Zweitens können NDJSON-Zeilen mehrere Chunk-Grenzen überspannen, was genau das gleiche Problem darstellt wie beim Parsen von CSV im Stream-Modus: Es ist nicht garantiert, dass eine einzelne fetch-Chunk mit einer einzigen JSON-Zeile übereinstimmt. Genau deshalb puffert das obige Beispiel teilweise Zeilen, anstatt direkt JSON.parse(chunk) auf das Eingehende anzuwenden.
Falls Sie lieber auf eine gehostete API zurückgreifen möchten – OpenAI, Anthropic oder ähnliche – ändert sich die Gesamtstruktur kaum: Sie verarbeiten weiterhin eine in Blöcken strukturierte HTTP-Antwort und leiten sie als SSE an den Client weiter. Die lokale Ollama-Version ist einfach diejenige, die Sie sofort ohne Account oder Schlüssel ausführen können, und sie stellt eine sinnvolle Wahl dar, um während der Entwicklung die Client-Seite einer solchen Funktion zu entwickeln, bevor derselbe Code auf ein gehostetes Modell verweist, sobald eine produktionstaugliche Ausgabequalität erforderlich ist.
Direktes Streamen einer Datei in das Objektspeicher-System
Das Senden einer Datei „an S3“ beinhaltet fast nie eine Datei, die bereits im Voraus klein genug ist, um bequem in den Speicher zu passen. Das Paket @aws-sdk/lib-storage existiert genau deshalb, weil die einfache PutObjectCommand-Klasse den Inhaltslänge bereits vor Beginn der Anfrage kennen muss. Ihr Hilfsfunktion Upload verwendet stattdessen einen Stream, entscheidet selbst, ob eine einzige Anfrage oder ein mehrteiliger Upload angemessen ist, und gibt unterwegs Fortschrittsmeldungen aus.
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`);
}
Um ohne echtes AWS-Konto zu experimentieren, bietet MinIO einen S3-kompatiblen Objektspeicher, den man mit einem einzigen Befehl lokal starten kann. Derselbe @aws-sdk/lib-storage-Code wie oben funktioniert auch mit MinIO – es sind lediglich Anpassungen notwendig, um den Client auf MinIOS Endpunkt statt auf AWS zu richten:
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' }
});
Dies ist der Ansatz, den man bei Dateien größer als ein paar Dutzend Megabyte bevorzugen sollte: Vermeiden Sie es, die gesamte Datei in einen Buffer umzuwandeln, bevor Sie sie an das SDK weitergeben. Geben Sie stattdessen einen createReadStream()-Objekt ein und überlassen Sie es Upload, den Rest zu handhaben – der Speicherverbrauch bleibt unverändert, egal ob die Datei 10 MB oder 10 GB groß ist.
Auswerten eines Uploads ohne Buffern des Anfragekörpers
Die andere Seite der Dateiuploads ist der Server, der sie empfangen muss, und genau hier tritt versehentliches Buffern am häufigsten auf – in der Regel, weil ein Middleware-Modul zur Analyse des Anfragekörpers nicht für das Streamen konzipiert wurde und heimlich den gesamten multipart-Anfrageinhalt in den Speicher las, bevor Ihr Handler ihn bearbeiten konnte.
busboy bearbeitet die Parsung von mehrteiligen Formularen schrittweise: Sobald Bytes für ein bestimmtes Dateifeld ankommen, liefert es Ihnen einen lesbaren Stream für dieses Feld zurück, anstatt zu warten, bis die gesamte Anfrage eingetroffen ist:
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);
In diesem Handler wird niemals der gesamte Anfragekörper gleichzeitig gespeichert. Die Bytes gelangen vom Socket über Busboys Parser direkt auf die Festplatte. Genau diese Struktur unterscheidet einen Dienst, der problemlos mit mehreren großen Uploads gleichzeitig umgehen kann, von einem Dienst, der bereits beim ersten gleichzeitigen Upload zweier Benutzer zusammenbricht.
Testen eines benutzerdefinierten Transformationsstreams
Handgerollte Transform-Streams sind ideale Kandidaten für stillen Verschleiß: Niemand macht sich die Mühe, Tests dafür zu schreiben, alle gehen davon aus, dass sie „einfach funktionieren“, und dann wird ein _flush-Edge Case fehlerhaft bereitgestellt, ohne dass jemand es bemerkt, bis in der Produktion. Dennoch ist das Testen eines solchen Streams weniger aufwendig, als es klingt. Man muss weder eine echte Datei noch einen echten HTTP-Server erstellen. Readable.from() in Kombination mit stream/promises reicht aus.
// 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');
});
});
Der zweite Test in dieser Datei ist wichtiger, als es auf den ersten Blick scheint. Er bestätigt, dass ein Fehler, der weiter oben ausgelöst wird, tatsächlich weitergeleitet wird, anstatt den Transform-Stream mitten in seiner Arbeit mit einer nie abgeschlossenen Promise stecken zu lassen. Dieses stille Stocken ist die häufigste Ursache dafür, dass ein scheinbar funktionierender Pipeline nach dem Go-Live zu einer blockierten Anfrage wird.
Eine leichte Metriken-Überwachung
Man benötigt keine vollständige Observabilitätsplattform, um nur zu überprüfen, ob ein Pipeline ordnungsgemäß funktioniert. Oft reicht bereits ein kleiner PassThrough-Stream aus, der Bytes und Blöcke während ihres Durchflusses zählt – dieser kann problemlos in jede bestehende Pipeline eingefügt werden, ohne etwas anderes zu stören:
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')
);
Falls man mehr als nur Ausgaben per console.log möchte, sollte die Logik innerhalb der Handler für 'data' und 'end' durch einen prom-client-Counter oder Histogram ersetzt werden. Die Struktur der Überwachung bleibt unverändert: Es handelt sich dabei einfach um einen weiteren Stream mitten in der Pipeline, der über das Durchfließen von Daten berichtet.
Drei Fehler, die nur unter Last auftauchen
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.
Das Wiederverwenden einer Stream-Instanz führt zum Leck von Zuhörern. Angenommen, Sie umschließen die Erstellung eines Streams in einer Funktion, die bei jeder Anfrage aufgerufen wird, aber etwas außerhalb dieser Funktion behält weiterhin eine Referenz darauf bei und fügt „für alle Fälle“ seine eigenen Zuhörer hinzu. Irgendwann werden Sie in Ihren Protokollen eine MaxListenersExceededWarning sehen, was die Art und Weise von Node ist, auf ein laufendes Speicherverlustproblem hinzuweisen. Jede Anfrage benötigt ihre eigenen, frisch erstellten Streams, und nichts außerhalb der Lebensdauer dieser Anfrage sollte nach Abschluss eine Referenz darauf behalten.
Niemand zerstört den Stream, wenn der Client sich abmeldet. Das ist der Fehler, der in der Produktion tatsächlich Kosten verursacht. Wenn ein Benutzer seine Browser-Tab während des Downloads schließt und man nicht auf response.on('close', ...) lauscht, um .destroy() am Quellstream auszuführen, zieht der Server weiterhin Daten vom Speicher ab – im schlimmsten Fall bleibt ein Datenbankcursor offen oder eine upstream-HTTP-Verbindung aktiv, und das alles für einen Client, der bereits weg ist. Bei ausreichendem gleichzeitigen Traffic führt das genau dazu, dass ein Server heimlich seine Dateidescriptoren oder Verbindungs-Pool aufbraucht, ohne dass in den Logs ein offensichtlicher Grund sichtbar wird. Die Lösung besteht immer aus denselben drei Schritten: auf das Schließen des Ziels lauschen, daraufhin den Quellstream zerstören und dieses Vorgehen als Standardpraxis für jede streambasierte Route betrachten – anstatt es erst nach einem Vorfall im Gedächtnis zu behalten.
Wann welche API verwendet werden sollte
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.
Keiner dieser Ansätze macht die anderen überflüssig. Jeder von ihnen löst ein anderes Problem im selben Gesamtfluss von Daten durch Ihr System. Heutzutage ist es üblich, dass ein einziger Pipeline mit einem Web ReadableStream beginnt, der über fetch geliefert wird, diese Daten umwandelt, sie durch eine mit compose() erstellte Transformation leitet und schließlich die Ausgabe an pipeline() weitergibt, um den Schreibvorgang zu handhaben. Früher wirkte das Mischen solcher APIs ungewöhnlich; heute ist es ein ziemlich gängiges Muster.
Fazit
Nichts hier entbindet von der Notwendigkeit, zunächst Backpressure, highWaterMark sowie die vier grundlegenden Stream-Klassen zu verstehen. Dieses Grundwissen bleibt eine unverzichtbare Voraussetzung, bevor die oben genannten Techniken nützlich werden können. Was sich ändert, sobald man dieses Wissen hat, ist der Punkt, an dem die eigentliche ingenieurtechnische Arbeit beginnt: Die Überprüfung von Speicherverbesserungen anhand konkreter Messwerte statt vager Behauptungen, das Bedenken, dass ein auf fetch basierender Stream und ein Node-Stream nur eine .toWeb()-Konvertierung benötigen, um miteinander zu funktionieren, sowie die Einbeziehung eines Plans für den Fall, dass der Client mitten in einer Anfrage abbricht – als fester Bestandteil der Routenprogrammierung und nicht als Detail, das später nachträglich behoben wird.
Falls Sie die zuvor erwähnten Benchmark-Beispiele auf unterschiedlicher Hardware ausführen und die Ergebnisse erheblich von denen abweichen, die hier gezeigt werden, handelt es sich um einen nützlichen Datenpunkt, den es wert ist zu teilen – anstatt ihn zu ignorieren –, denn Werte bezüglich Durchsatz und Speichernutzung sind viel zuverlässiger, wenn sie auf mehr als einem Gerät überprüft wurden.
Ressourcen
- Node.js-Dokumentation zur klassischen Stream API: https://nodejs.org/api/stream.html
- Node.js-Dokumentation zur Web Streams API: https://nodejs.org/api/webstreams.html
- Ollamas Referenz zu seinem Streaming-Endpunkt: https://docs.ollama.com/api/streaming
- Die Liste des Pakets
@aws-sdk/lib-storageauf npm: https://www.npmjs.com/package/@aws-sdk/lib-storage - Das Projektverzeichnis von
busboy: https://github.com/mscdex/busboy