post

Node.js-Streams: Backpressure verstehen und nutzen

Warum ein schneller Producer einen langsamen Consumer nicht überrennen darf – und wie pipe, pipeline, der Rückgabewert von write() und das drain-Event die Flusskontrolle in Node.js regeln.

Wenn ich in einer Schulung nach Streams frage, kommt fast immer dieselbe Antwort: "Das ist doch dieses Dateikram-Ding mit createReadStream." Stimmt – aber es greift zu kurz. Streams sind kein Datei-Feature, sie sind das grundlegende Modell, wie Node.js Daten stückweise durch eine Pipeline schiebt, ohne dass jemals alles gleichzeitig im Speicher liegt. Und der eigentlich spannende Teil – der, der in echten Systemen den Unterschied zwischen "läuft stabil" und "Prozess stirbt nachts um drei" macht – heißt Backpressure.

Genau da hakt es bei vielen. Die Stream-API sieht harmlos aus, man verkettet ein paar pipe-Aufrufe, und es funktioniert. Bis zu dem Tag, an dem die Quelle schneller liefert als das Ziel schlucken kann. Dann läuft entweder alles wie geschmiert, oder der Speicher wächst und wächst. Der Unterschied liegt darin, ob man Backpressure verstanden hat oder nur ignoriert.

Warum Streams überhaupt

Der klassische Weg, eine Datei zu verarbeiten, ist fs.readFile. Die Funktion liest die komplette Datei in einen Buffer und ruft dann den Callback auf. Bei einer 2-MB-Konfigurationsdatei völlig in Ordnung. Bei einem 4-GB-Logfile hat man ein Problem: Node.js versucht, vier Gigabyte in den Heap zu laden, und der Prozess kippt um, lange bevor überhaupt eine Zeile verarbeitet wurde.

Ein Stream macht es anders. Er liest die Datei in kleinen Häppchen – Chunks – und reicht jeden Chunk einzeln weiter. Zu jedem Zeitpunkt liegt nur ein kleiner Puffer im Speicher, unabhängig davon, ob die Datei zwei Megabyte oder zwei Terabyte groß ist. Der Speicherverbrauch bleibt konstant. Das ist der Kernnutzen, und er gilt nicht nur für Dateien: HTTP-Requests, TCP-Sockets, Kompression, Verschlüsselung – all das sind in Node.js Streams.

Die vier Typen

Bevor es um Flusskontrolle geht, lohnt sich der Überblick über die vier Stream-Typen. Alle vier erben von EventEmitter, kommunizieren also über Events – das ist der rote Faden, der sich durch die ganze asynchrone Seite von Node.js zieht, wie ich sie im Node.js-Training: Vom Event Loop zur Architekturentscheidung ausführlicher beschreibe.

  • Ein Readable ist eine Quelle, aus der man liest. fs.createReadStream() liefert einen Readable, ebenso der eingehende Body eines HTTP-Requests.
  • Ein Writable ist ein Ziel, in das man schreibt. fs.createWriteStream() oder die HTTP-Response sind Writables.
  • Ein Duplex ist lesbar und schreibbar zugleich, aber mit getrennten Kanälen. Ein net.Socket ist das Paradebeispiel: rein und raus über dieselbe Verbindung.
  • Ein Transform ist ein Duplex, bei dem der Output eine Funktion des Inputs ist. zlib.createGzip() nimmt rohe Bytes rein und gibt komprimierte raus. Genau hier setzt man eigene Verarbeitungslogik an.

Diese Typen kombiniert man zu einer Pipeline: eine Quelle, beliebig viele Transforms in der Mitte, ein Ziel am Ende.

flowchart LR
    A["Readable<br/>fs.createReadStream"] -->|chunks| B["Transform<br/>gzip / uppercase"]
    B -->|chunks| C["Writable<br/>fs.createWriteStream"]
    C -.->|drain-Signal| A
    style A fill:#2b6cb0,color:#fff
    style B fill:#2f855a,color:#fff
    style C fill:#c05621,color:#fff

Die durchgezogenen Pfeile sind der Datenfluss vorwärts. Der gestrichelte Pfeil zurück ist das Entscheidende – er ist Backpressure. Dazu gleich mehr.

Der naive Ansatz und warum er kippt

Nehmen wir an, wir wollen eine große Datei in eine andere kopieren und dabei jeden Chunk verarbeiten. Der intuitive Code sieht so aus:

const fs = require('node:fs');

const source = fs.createReadStream('huge-input.log');
const target = fs.createWriteStream('huge-output.log');

source.on('data', (chunk) => {
  target.write(chunk);
});

Das läuft. Meistens. Und dann kommt der Fall, in dem huge-input.log von einer schnellen SSD kommt und huge-output.log auf ein langsames Netzlaufwerk geschrieben wird. Die Quelle feuert data-Events, so schnell sie kann, und target.write() nimmt jeden Chunk entgegen. Kann das Ziel nicht schnell genug wegschreiben, puffert Node.js die Chunks intern – und dieser Puffer kennt keine Obergrenze. Der Speicher füllt sich, RSS klettert, der Garbage Collector arbeitet gegen eine immer größere Menge, und irgendwann beendet das Betriebssystem den Prozess.

Das ist der Fallstrick in einem Satz: Wer den Rückgabewert von write() ignoriert und einfach weiterschreibt, produziert unbegrenztes Puffern.

Was write() eigentlich sagt

writable.write(chunk) gibt einen Boolean zurück, und dieser Boolean ist die halbe Miete:

  • true bedeutet: Der interne Puffer liegt noch unter highWaterMark, du darfst weiterschreiben.
  • false bedeutet: Der Puffer hat highWaterMark erreicht oder überschritten. Hör auf zu schreiben, sonst puffert Node.js ins Blaue.

highWaterMark ist die Puffergrenze. Für Byte-Streams liegt der Default bei 16384 Bytes, also 16 KiB. Im Object-Mode – dazu später – sind es 16 Objekte. Wichtig zu verstehen: Diese Grenze ist ein Schwellwert, keine harte Mauer. Node.js hört nicht auf zu puffern, wenn man highWaterMark überschreitet. Es sagt einem nur höflich per false-Rückgabe Bescheid, dass es genug ist. Wer die Warnung überhört, puffert trotzdem weiter – auf eigenes Risiko.

Wenn write() einmal false geliefert hat, sollte man pausieren und auf das 'drain'-Event warten. Das Writable emittiert 'drain', sobald sein Puffer wieder geleert ist – das Signal, dass man weiterschreiben darf. Von Hand sieht der korrekte Zyklus so aus:

source.on('data', (chunk) => {
  const ok = target.write(chunk);
  if (!ok) {
    source.pause();
    target.once('drain', () => source.resume());
  }
});

Wir schreiben, prüfen den Rückgabewert, pausieren die Quelle bei false und nehmen sie erst nach 'drain' wieder auf. Genau das ist Flusskontrolle: Ein schneller Producer wird ausgebremst, damit er einen langsamen Consumer nicht überrennt.

Den Zyklus kann man auch als Zustandsmaschine zeichnen:

stateDiagram-v2
    [*] --> Schreiben
    Schreiben --> Schreiben: write() liefert true
    Schreiben --> Pausiert: write() liefert false<br/>source.pause()
    Pausiert --> Schreiben: 'drain'-Event<br/>source.resume()

pipe erledigt das automatisch

Die gute Nachricht: Diesen ganzen pause/resume-Tanz muss man in der Praxis selten von Hand schreiben. pipe() macht genau das im Hintergrund.

const fs = require('node:fs');

fs.createReadStream('huge-input.log')
  .pipe(fs.createWriteStream('huge-output.log'));

pipe() verdrahtet Quelle und Ziel, prüft intern den Rückgabewert von write(), pausiert den Readable bei Backpressure und resumt ihn nach 'drain'. Backpressure ist also gelöst, ohne dass man eine Zeile Flusskontrolle selbst schreibt. pipe() gibt außerdem das Ziel zurück, sodass man verketten kann – praktisch, wenn ein Transform dazwischenhängt:

const fs = require('node:fs');
const zlib = require('node:zlib');

fs.createReadStream('huge-input.log')
  .pipe(zlib.createGzip())
  .pipe(fs.createWriteStream('huge-input.log.gz'));

Und trotzdem rate ich in Schulungen davon ab, produktiv nur pipe() zu verketten. Der Grund ist die Fehlerbehandlung.

Warum pipeline die bessere Wahl ist

pipe() hat eine tückische Schwäche: Es propagiert Fehler nicht. Wirft eine Stage mitten in der Kette einen Fehler – die Netzverbindung bricht weg, die Platte ist voll – dann bekommt das ein pipe()-Verbund nicht sauber mit. Die anderen Streams werden nicht zerstört, Dateideskriptoren bleiben offen, der Stream hängt. In einem langlaufenden Serverprozess sammeln sich solche Leaks an, bis irgendwann die Ressourcen ausgehen.

stream.pipeline() löst das. Es verkettet die Stages genauso, aber es propagiert Fehler über alle Stufen hinweg und ruft bei Fehler oder regulärem Ende destroy() auf jedem beteiligten Stream auf. Nichts leakt. Es gibt die Callback-Variante:

const fs = require('node:fs');
const zlib = require('node:zlib');
const { pipeline } = require('node:stream');

pipeline(
  fs.createReadStream('huge-input.log'),
  zlib.createGzip(),
  fs.createWriteStream('huge-input.log.gz'),
  (err) => {
    if (err) {
      console.error('pipeline failed', err);
      return;
    }
    console.log('pipeline succeeded');
  }
);

Seit Node.js 15 – in den 2022 aktuellen Release-Linien 16 und 18 also fest verankert – gibt es zusätzlich eine Promises-Variante aus node:stream/promises. Die liest sich in async-Funktionen deutlich sauberer, weil man Fehler mit try/catch fängt:

const fs = require('node:fs');
const zlib = require('node:zlib');
const { pipeline } = require('node:stream/promises');

async function compress() {
  try {
    await pipeline(
      fs.createReadStream('huge-input.log'),
      zlib.createGzip(),
      fs.createWriteStream('huge-input.log.gz')
    );
    console.log('done');
  } catch (err) {
    console.error('pipeline failed', err);
  }
}

compress();

Die Promises-Variante nimmt auch Optionen, etwa ein signal für ein AbortSignal, mit dem man die Pipeline von außen abbrechen kann. Der Punkt, den ich Teilnehmern mitgebe: Backpressure regelt pipeline() genauso automatisch wie pipe() – aber es räumt auch auf. Für Produktionscode ist pipeline() die Standardwahl, pipe() höchstens für kurze, isolierte Skripte.

Einen eigenen Transform schreiben

Der Punkt, an dem Streams richtig nützlich werden, ist der eigene Transform in der Mitte. Ein Transform bekommt Chunks rein, verarbeitet sie und schiebt das Ergebnis raus. Das Skelett besteht aus einer transform-Funktion, die drei Dinge bekommt: den Chunk, das Encoding und einen Callback, den man am Ende aufrufen muss.

const { Transform } = require('node:stream');

const upperCase = new Transform({
  transform(chunk, encoding, callback) {
    const transformed = chunk.toString().toUpperCase();
    this.push(transformed);
    callback();
  }
});

this.push() reicht das Ergebnis an die nächste Stage weiter, callback() signalisiert, dass dieser Chunk fertig verarbeitet ist. Ruft man callback() mit einem Fehler auf – callback(err) –, propagiert pipeline() diesen Fehler sauber durch die ganze Kette. Eingebaut in eine Pipeline sieht das so aus:

const fs = require('node:fs');
const { pipeline } = require('node:stream/promises');

await pipeline(
  fs.createReadStream('input.txt'),
  upperCase,
  fs.createWriteStream('output.txt')
);

Und weil der upperCase-Transform wie jeder Stream an der Flusskontrolle teilnimmt, greift Backpressure auch hier: Ist das Ziel langsam, bremst die Pipeline über den Transform hinweg bis zur Quelle zurück. Man muss dafür nichts extra tun.

Async-Iteration als moderne Alternative

Seit Node.js 10 implementiert jeder Readable Symbol.asyncIterator, lässt sich also direkt mit for await ... of konsumieren. Das liest sich in async-Funktionen oft klarer als on('data') und behält die Flusskontrolle bei, ohne dass man pause und resume anfassen muss.

const fs = require('node:fs');

async function countLines(path) {
  const stream = fs.createReadStream(path, { encoding: 'utf8' });
  let lines = 0;
  for await (const chunk of stream) {
    lines += chunk.split('\n').length - 1;
  }
  return lines;
}

Der Backpressure entsteht hier implizit: Solange der Schleifenkörper mit einem Chunk beschäftigt ist – und ein await darin ihn zusätzlich aufhält –, fordert die Schleife keinen neuen Chunk an, und der Readable liefert von selbst erst dann den nächsten. Man bekommt also dieselbe Bremse wie bei pipe(), nur in gewöhnlichem async/await-Code. Wichtig bleibt die Regel, die gleich noch kommt: Wer so iteriert, sollte den Stream nicht parallel auch noch über on('data') konsumieren.

Object-Mode und ein paar Regeln

Standardmäßig arbeiten Streams mit Buffers und Strings. Setzt man objectMode: true, dürfen die Chunks beliebige JavaScript-Objekte sein – nützlich, wenn man eine Pipeline aus geparsten Datensätzen baut statt aus rohen Bytes. Zu beachten ist dabei, dass highWaterMark im Object-Mode als Anzahl Objekte zählt, nicht als Bytes; der Default sind 16 Objekte. Und ein bestehender Stream lässt sich nicht nachträglich sicher auf Object-Mode umschalten – das entscheidet man bei der Konstruktion.

Eine Regel, an der ich in Schulungen immer wieder hänge: Konsummethoden nicht mischen. Pro Readable wählt man genau eine Art zu lesen – entweder on('data'), oder on('readable'), oder pipe(), oder async-Iteration. Wer zwei davon kombiniert, produziert Race Conditions, verlorene Chunks und schwer reproduzierbare Bugs. Ein Readable, eine Konsummethode.

Was hängenbleiben soll

Streams sind das Werkzeug, um beliebig große Datenmengen mit konstantem Speicher zu verarbeiten – der Puffer bleibt klein, egal wie groß die Quelle ist. Backpressure ist der Mechanismus, der das trägt: Der Rückgabewert von write() und das 'drain'-Event bilden die Flusskontrolle, die einen schnellen Producer davon abhält, einen langsamen Consumer zu überrennen. highWaterMark ist dabei der Schwellwert, an dem write() von true auf false kippt – eine Warnung, kein Zwang.

Von Hand muss man diesen Zyklus selten schreiben, weil pipe() und vor allem pipeline() ihn automatisch regeln. Der entscheidende Unterschied liegt in der Fehlerbehandlung: pipeline() propagiert Fehler und räumt alle Streams auf, pipe() tut das nicht. Wer das einmal verinnerlicht hat, schreibt Node.js-Verarbeitungscode, der auch bei vier Gigabyte und einem langsamen Ziel nicht umkippt – sondern einfach ein bisschen langsamer wird, so wie es sein soll.

Weiterführende Quellen

Kommentare

Kommentar schreiben