Node.js Streams

Warum Streams wichtig sind

Das Laden einer großen Datei in den Arbeitsspeicher funktioniert so lange, bis die Datei größer ist als der verfügbare Speicher. Streams lösen dieses Problem, indem sie Daten in kleinen Stücken (Chunks) verarbeiten: Man liest ein Stück, verarbeitet es und macht dann weiter. So bleibt der Speicherverbrauch weitgehend konstant, unabhängig von der Größe der Eingabedaten.

Streams sind in Node. allgegenwärtig. HTTP-Request- und Response-Bodies, das Lesen und Schreiben von Dateien, Kompression, Verschlüsselung und viele Parser basieren auf Streams. Wenn man sie versteht, kann man Server und Tools bauen, die unbegrenzte Datenmengen verarbeiten, ohne abzustürzen.

Die vier Stream-Typen

Jeder Stream gehört zu einem von vier Typen:

• Readable — eine Quelle, aus der gelesen wird. Dateien, HTTP request bodies, Sockets und process.stdin.
• Writable — ein Ziel, in das geschrieben wird. Dateien, HTTP responses, Sockets und process.stdout.
• Duplex — sowohl readable als auch writable, wie zum Beispiel ein TCP socket.
• Transform — ein Duplex-Stream, der Daten während der Übertragung modifiziert, wie etwa zlib.createGzip() oder ein CSV-Parser.

Readable- und Writable-Streams werden in Pipelines verbunden, wobei Transforms in der Mitte sitzen.

Lesen und Schreiben

Der einfachste Weg, einen readable stream zu konsumieren, ist die asynchrone Iteration.

// read.js
import { createReadStream } from "node:fs";

const stream = createReadStream("access.log", { encoding: "utf8" });

for await (const chunk of stream) {
  process.stdout.write(chunk);
}

Rufen Sie bei writable streams write() auf und signalisieren Sie das Ende mit end().

// write.js
import { createWriteStream } from "node:fs";

const out = createWriteStream("out.txt");

out.write("first line\n");
out.write("second line\n");
out.end();

Der Stream puffert Schreibvorgänge intern und führt sie effizient aus, sodass Sie die Chunks nicht selbst verwalten müssen.

Piping und Pipelines

Eine Pipeline verbindet einen Readable-Stream, null oder mehr Transform-Streams und einen Writable-Stream.

// compress.js
import { createReadStream, createWriteStream } from "node:fs";
import { pipeline } from "node:stream/promises";
import { createGzip } from "node:zlib";

await pipeline(
  createReadStream("access.log"),
  createGzip(),
  createWriteStream("access.log.gz"),
);

Bevorzuge immer pipeline gegenüber manuellem .pipe()-Chaining. pipeline leitet Fehler aus jeder Stufe weiter, zerstört die Streams im Fehlerfall und gibt ein Promise zurück, was die Fehlerbehandlung erheblich vereinfacht. Manuelles .pipe() propagiert keine Fehler, sodass ein Fehler im Upstream die Pipeline hängen lassen und das Ziel offen halten kann.

Backpressure

Streams verfügen über interne Buffer. Wenn Sie Daten schneller schreiben, als das Ziel sie verarbeiten kann, füllt sich der Buffer und write() gibt false zurück.

// backpressure.js
function writeAll(stream, chunks) {
  return new Promise((resolve, reject) => {
    let i = 0;
    const next = () => {
      while (i < chunks.length) {
        const ok = stream.write(chunks[i++]);
        if (!ok) {
          stream.once("drain", next);
          return;
        }
      }
      stream.end(resolve);
    };
    stream.on("error", reject);
    next();
  });
}

Das Warten auf drain, bevor weitere Daten geschrieben werden, ist die Art und Weise, wie ein Producer einen langsamen Consumer berücksichtigt. pipeline und pipe handhaben dies automatisch, was ein weiterer Grund für deren Verwendung ist.

Transform Streams

Ein Transform Stream wendet eine Funktion auf jeden Chunk an. Du kannst deine eigenen erstellen.

// upper.js
import { Transform } from "node:stream";

const upper = new Transform({
  transform(chunk, encoding, callback) {
    callback(null, chunk.toString().toUpperCase());
  },
});

process.stdin.pipe(upper).pipe(process.stdout);

Node liefert nützliche Transforms in node:zlib (gzip, deflate), node:crypto (Ciphers und Hashes) sowie in vielen Bibliotheken, wie zum Beispiel CSV- und JSON-Parsern, mit.

Object mode

Standardmäßig übertragen Streams Bytes. Der Object mode ermöglicht es ihnen, stattdessen JavaScript-Objekte zu übertragen, was ideal für strukturierte Pipelines ist.

// object-mode.js
import { Readable, Transform } from "node:stream";

Readable.from([{ id: 1 }, { id: 2 }])
  .pipe(
    new Transform({
      objectMode: true,
      transform(record, _encoding, callback) {
        callback(null, { ...record, seen: true });
      },
    }),
  )
  .on("data", console.log);

Im Object mode zählt highWaterMark Objekte anstelle von Bytes. Auf diese Weise werden Datenbank-Row-Streams, Log-Prozessoren und ETL-Pipelines implementiert.

Gängige Patterns

• Datei kopieren: pipeline(createReadStream(src), createWriteStream(dest)).
• Komprimieren oder verschlüsseln: createGzip() oder einen Cipher-Transform einfügen.
• HTTP-Response streamen: Eine Datei oder ein Abfrageergebnis direkt an res pipen.
• Zeilengetrennte Daten parsen: Einen Transform verwenden, der bei Zeilenumbrüchen splittet.
• Upload-Fortschritt: Bytes in einem Transform während des Durchlaufs zählen.

Best Practices

• Bevorzuge pipeline gegenüber .pipe() für die Fehlerbehandlung und das Cleanup.
• Streame große Dateien und Antworten, anstatt sie zu puffern.
• Beachte Backpressure; ignoriere niemals, wenn write() false zurückgibt.
• Optimiere highWaterMark erst nach entsprechenden Messungen.
• Nutze den Object Mode für strukturierte Daten.
• Behandle error bei jedem Stream, den du erstellst.
• Zerstöre Streams im Fehlerfall, damit File Descriptors wieder freigegeben werden.

Häufige Fehler

• Verwendung von readFile für große oder unbegrenzte Datenmengen.
• Verkettung von .pipe(), wodurch Fehler verloren gehen.
• Ignorieren von Backpressure und das Puffern unbegrenzter Daten.
• Vergessen, dass ein Transform-Callback genau einmal aufgerufen werden muss.
• Mischen von Encodings, was zu korruptem Output führt.
• Streams nach einem Fehler geöffnet lassen.

Wie geht es weiter?

Streams sind die Art und Weise, wie Node.js Daten in großem Umfang verarbeitet. Setzen Sie dieses Wissen im File System Guide in die Praxis um und verstehen Sie die zugrunde liegende Planung im Event Loop Guide. Schreiben Sie anschließend einen readFile-Aufruf als Stream um und beobachten Sie, wie die Speicherauslastung sinkt.