Bootcamp backendontwikkeling met Node.js · Les

Backpressure, pipe() en het pipeline()-hulpprogramma

Diagnosticeer geheugengroei en verbind streams veilig met pipeline() om fouten door te geven en backpressure te respecteren.

Les 3 van 413 stappen

Backpressure, pipe() en het pipeline()-hulpprogramma is een gratis Bootcamp backendontwikkeling met Node.js-les op CoddyKit. Dit is les 3 van 4. Je kunt de volledige les hieronder gratis lezen en daarna in de browser praktisch oefenen met een ingebouwde code-editor en een AI-begeleider die 24/7 beschikbaar is. Deze les maakt deel uit van het leertraject Bootcamp backendontwikkeling met Node.js. Je voortgang wordt gesynchroniseerd op het web en in de CoddyKit-app. De cursus Bootcamp backendontwikkeling met Node.js bevat in totaal 4 lessen.

Waarom geheugen opzwelt in streams

Een Readable-stream van Node.js kan sneller gegevens produceren dan een Writable ze kan verbruiken. Als je de producent nooit vertelt dat deze moet vertragen, stapelen onverbruikte chunks zich op in een interne buffer en groeit het procesgeheugen totdat de GC het niet meer kan bijhouden.

Een trage schijf, trage netwerksocket of trage databasebewerking is doorgaans de verbruiker.
Een snelle bestandsbewerking of HTTP-upload is doorgaans de producent.

Het mechanisme dat ervoor zorgt dat de producent op de verbruiker wacht, heet backpressure. Streams verkeerd gebruiken betekent bijna altijd dat backpressure is genegeerd.

De naïeve, defecte kopie

Hier zie je de klassieke geheugenfout. We luisteren naar data en roepen voor elke chunk dst.write() aan, waarbij we de geretourneerde waarde negeren.

Als dst trager is dan src, worden de niet-weggeschreven chunks zonder bovengrens in de buffer van dst in de wachtrij geplaatst. Bij een bestand van meerdere gigabytes kan dit het RAM uitputten.

const fs = require('fs');

const src = fs.createReadStream('big.bin');
const dst = fs.createWriteStream('copy.bin');

// BUG: return value of write() is ignored, so backpressure is never honored
src.on('data', (chunk) => {
  dst.write(chunk);
});
src.on('end', () => dst.end());

Wat write() daadwerkelijk retourneert

writable.write(chunk) retourneert een boolean:

true — de interne buffer bevindt zich onder highWaterMark; blijf schrijven.
false — de buffer is vol; je moet stoppen met schrijven en op het 'drain'-event wachten voordat je meer verzendt.

Deze retourwaarde respecteren is de handmatige manier om backpressure toe te passen. De producent moet pauzeren totdat de verbruiker aangeeft dat de buffer is leeggemaakt.

Handmatige backpressure met pause/resume

Handmatig betekent backpressure het volgende: wanneer write() false retourneert, pauzeer je de bron met pause(); wanneer de bestemming 'drain' genereert, hervat je de bron met resume().

Dit werkt, maar is omslachtig en gemakkelijk verkeerd te doen — je moet ook de afhandeling van error en end voor beide streams instellen.

const fs = require('fs');

const src = fs.createReadStream('big.bin');
const dst = fs.createWriteStream('copy.bin');

src.on('data', (chunk) => {
  const ok = dst.write(chunk);
  if (!ok) {
    src.pause();              // stop reading until the buffer drains
    dst.once('drain', () => src.resume());
  }
});
src.on('end', () => dst.end());

pipe() doet dit voor je

readable.pipe(writable) regelt automatisch dezelfde dans van pauzeren, hervatten en leegmaken en respecteert backpressure direct.

De methode retourneert de bestemmingsstream, zodat je transforms kunt koppelen:

src.pipe(gzip).pipe(dst)

Voor de meeste eenvoudige kopieën is pipe() veel beter dan de handmatige lus hierboven.

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

const src = fs.createReadStream('big.bin');
const gzip = zlib.createGzip();
const dst = fs.createWriteStream('big.bin.gz');

// pipe() handles backpressure across all three streams
src.pipe(gzip).pipe(dst);

De verborgen tekortkoming van pipe()

pipe() handelt backpressure af, maar geeft fouten niet door. Als gzip of dst 'error' genereert, wordt de bron niet automatisch vernietigd.

De upstream-stream houdt zijn bestandsdescriptor open — een resourcelek.
Een niet-afgehandeld 'error'-event veroorzaakt een uitzondering en kan het proces laten crashen.

Om pipe() veilig te gebruiken, moet je aan elke stream een error-handler koppelen en de andere streams handmatig vernietigen. Precies die boilerplate neemt pipeline() weg.

Maak kennis met stream.pipeline()

stream.pipeline() koppelt een reeks streams, geeft backpressure door, stuurt fouten door en vernietigt elke stream in de keten wanneer een ervan mislukt of klaar is.

Je geeft de streams in volgorde door, gevolgd door een callback die één keer wordt uitgevoerd met een fout (of null bij succes):

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

pipeline(
  fs.createReadStream('big.bin'),
  zlib.createGzip(),
  fs.createWriteStream('big.bin.gz'),
  (err) => {
    if (err) {
      console.error('Pipeline failed:', err.message);
    } else {
      console.log('Pipeline succeeded');
    }
  }
);

De promisegebaseerde pipeline()

Gebruik in moderne code de promisevariant uit stream/promises. Deze wordt bij succes opgelost en bij een fout afgewezen, zodat één try/catch de hele keten en de opruiming afhandelt.

Dit is de aanbevolen manier om streams aan te sluiten in asynchrone backend-handlers.

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

async function compress() {
  try {
    await pipeline(
      fs.createReadStream('big.bin'),
      zlib.createGzip(),
      fs.createWriteStream('big.bin.gz')
    );
    console.log('done');
  } catch (err) {
    console.error('failed:', err.message);
  }
}

compress();

Een uitvoerbare pipeline in het geheugen

Je hebt geen bestanden nodig om pipeline() te zien werken. Readable.from() maakt van elke iterable een stream en een Transform kan elke chunk omzetten naar hoofdletters. Het geheel draait zelfstandig.

Let erop dat fouten uit elke fase ervoor zouden zorgen dat de gewachte pipeline() wordt afgewezen.

const { Readable, Transform } = require('stream');
const { pipeline } = require('stream/promises');

const source = Readable.from(['hello ', 'stream ', 'world']);

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

const chunks = [];
const sink = new Transform({
  transform(chunk, _enc, cb) {
    chunks.push(chunk.toString());
    cb();
  }
});

(async () => {
  await pipeline(source, upper, sink);
  console.log(chunks.join(''));
})();

highWaterMark: de buffer afstemmen

Elke stream heeft een highWaterMark (standaard 16 KB voor byte-streams, 16 objecten voor objectmodus). Dit is de drempel waarbij write() false retourneert en het lezen wordt gepauzeerd.

Een grotere highWaterMark verhoogt de doorvoer, maar gebruikt meer geheugen per stream.
Een kleinere waarde activeert tegendruk eerder, waardoor het geheugengebruik strakker wordt begrensd.

Het is een bufferdrempel, geen harde limiet — maar wel de instelling die bepaalt hoe snel tegendruk wordt geactiveerd.

const fs = require('fs');

// Pause reads after only 64 KB is buffered downstream
const src = fs.createReadStream('big.bin', { highWaterMark: 64 * 1024 });
const dst = fs.createWriteStream('copy.bin', { highWaterMark: 64 * 1024 });

src.pipe(dst);

pipeline() in een HTTP-handler

Een veelgemaakte backend-fout is een volledige upload of download in het geheugen bufferen voordat je antwoord geeft. Door de antwoordtekst met pipeline() te streamen blijft het geheugengebruik constant en wordt alles opgeruimd als de client de verbinding verbreekt.

Omdat het HTTP-antwoord een Writable is, remt tegendruk van een trage client het lezen van het bestand automatisch af.

const http = require('http');
const fs = require('fs');
const { pipeline } = require('stream');

http.createServer((req, res) => {
  pipeline(
    fs.createReadStream('big.bin'),
    res,
    (err) => {
      if (err) {
        console.error('stream error:', err.message);
        res.destroy();
      }
    }
  );
}).listen(3000);

Korte controle

Je streamt een bestand naar een trage client via een gzip-transformatie. Welke aanpak respecteert tegendruk veilig en ruimt elke stream op als de client tijdens de overdracht de verbinding verbreekt?

Beantwoord dit in de interactieve les

Samenvatting

Belangrijkste punten voor het veilig verbinden van streams:

Tegendruk voorkomt dat een snelle producent een trage consument overbelast; het negeren van de booleaanse retourwaarde van write() is de hoofdoorzaak van overmatig streamgeheugengebruik.
pipe() handelt tegendruk af, maar niet het doorgeven van fouten of het opruimen — een valkuil die tot gelekte bestandsdescriptors leidt.
stream.pipeline() (de callback-versie of de versie uit stream/promises) geeft tegendruk door, stuurt fouten door en vernietigt alle streams in de keten.
highWaterMark bepaalt hoe snel tegendruk wordt geactiveerd en vormt een afweging tussen geheugen en doorvoer.
Gebruik in HTTP-handlers pipeline() om te streamen in plaats van volledige ladingen te bufferen.

Gratis beginnen

Leer JavaScript met een AI-tutor — gratis

Schrijf echte code en voer die uit in je browser, krijg direct hulp van een AI-tutor die 24/7 beschikbaar is en ga verder waar je gebleven bent op het web of in de app.

Cursussen: 22
Lessen: 92

Gratis beginnen met leren Bekijk de volledige cursus

Veelgestelde vragen

Is de les “Backpressure, pipe() en het pipeline()-hulpprogramma” gratis?

Ja — de volledige tekst van “Backpressure, pipe() en het pipeline()-hulpprogramma” kun je hier gratis op het web lezen. Als je interactief wilt oefenen met een ingebouwde code-editor en een AI-begeleider die 24/7 beschikbaar is, en de rest van de cursus Bootcamp backendontwikkeling met Node.js wilt ontgrendelen, kun je upgraden naar CoddyKit PRO. De cursus Bootcamp backendontwikkeling met Node.js bevat in totaal 4 lessen.

Wat leer ik in “Backpressure, pipe() en het pipeline()-hulpprogramma”?

Diagnosticeer geheugengroei en verbind streams veilig met pipeline() om fouten door te geven en backpressure te respecteren. Je oefent met Bootcamp backendontwikkeling met Node.js door code rechtstreeks in de browser uit te voeren. Een AI-begeleider die 24/7 beschikbaar is beantwoordt je vragen terwijl je de les doorwerkt.

Heb ik ervaring nodig om met Bootcamp backendontwikkeling met Node.js te beginnen?

Ervaring vooraf is niet nodig. Bootcamp backendontwikkeling met Node.js op CoddyKit is opgebouwd voor beginners tot gevorderden, zodat je hier of bij het begin kunt starten en in je eigen tempo kunt leren. Dit is les 3 van 4.

Hoe lang duurt de les “Backpressure, pipe() en het pipeline()-hulpprogramma”?

De meeste lessen van CoddyKit duren ongeveer 5–10 minuten. Elke les is kort en interactief, zodat je gestaag vooruitgaat en op het web en in de app precies verdergaat waar je was gebleven.

Kan ik code schrijven en uitvoeren in deze les over Bootcamp backendontwikkeling met Node.js?

Ja. Elke les over Bootcamp backendontwikkeling met Node.js bevat een ingebouwde code-editor, zodat je rechtstreeks in je browser echte code kunt schrijven en uitvoeren en direct feedback van AI krijgt — lokale installatie is niet nodig.

Alle lessen in deze cursus

Internals van readable-, writable-, duplex- en transformstreams
Aangepaste transformstreams implementeren met _transform en _flush
Backpressure, pipe() en het pipeline()-hulpprogramma
Async-iterators en for-await-of over streams

← Terug naar Bootcamp backendontwikkeling met Node.js