Bootcamp i backendutveckling med Node.js · Lektion

Backpressure, pipe() och verktyget pipeline()

Diagnostisera minnesuppsvällning och koppla strömmar säkert med pipeline() för att vidarebefordra fel och respektera backpressure.

Lektion 3 av 413 steg

Backpressure, pipe() och verktyget pipeline() är en gratis lektion i Bootcamp i backendutveckling med Node.js på CoddyKit. Detta är lektion 3 av 4. Ni kan läsa hela lektionen gratis nedan och sedan öva praktiskt i webbläsaren med en inbyggd kodredigerare och en AI-handledare som är tillgänglig dygnet runt. Den ingår i lärvägen för Bootcamp i backendutveckling med Node.js, och Era framsteg synkroniseras mellan webben och CoddyKit-appen. Kursen i Bootcamp i backendutveckling med Node.js innehåller totalt 4 lektioner.

Varför minnet sväller i strömmar

En Node.js-Readable-ström kan producera data snabbare än en Writable-ström kan konsumera den. Om ni aldrig säger åt producenten att sakta ner samlas obearbetade block i en intern buffert, och processens minne växer tills GC inte längre hinner med.

En långsam disk, ett långsamt nätverksuttag eller en långsam databasskrivning är den typiska konsumenten.
En snabb filläsning eller HTTP-uppladdning är den typiska producenten.

Mekanismen som får producenten att vänta på konsumenten kallas backpressure. Felaktig användning av strömmar innebär nästan alltid att backpressure har ignorerats.

Den naiva (trasiga) kopieringen

Här är det klassiska minnesfelet. Vi lyssnar efter data och anropar dst.write() för varje block, utan att ta hänsyn till returvärdet.

Om dst är långsammare än src köas de oskrivna blocken upp i bufferten för dst utan någon övre gräns. För en fil på flera gigabyte kan detta tömma RAM.

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

Vad write() faktiskt returnerar

writable.write(chunk) returnerar ett booleanvärde:

true — den interna bufferten ligger under highWaterMark; fortsätt skriva.
false — bufferten är full; ni bör sluta skriva och vänta på händelsen 'drain' innan ni skickar mer.

Att respektera detta returvärde är det manuella sättet att tillämpa backpressure. Producenten måste pausa tills konsumenten signalerar att bufferten har tömts.

Manuell backpressure med pause/resume

Manuellt innebär backpressure att ni, när write() returnerar false, anropar pause() på källan och, när destinationen skickar händelsen 'drain', anropar resume() på den.

Det fungerar, men är omständligt och lätt att göra fel — ni måste dessutom koppla in hantering av error och end för båda strömmarna.

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() gör detta åt er

readable.pipe(writable) kopplar automatiskt ihop samma pause/resume/drain-flöde och hanterar backpressure direkt.

Metoden returnerar destinationsströmmen, så ni kan kedja transformeringar:

src.pipe(gzip).pipe(dst)

För de flesta enkla kopieringar är pipe() mycket bättre än den manuella loopen ovan.

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

Den dolda bristen i pipe()

pipe() hanterar backpressure, men vidarebefordrar inte fel. Om gzip eller dst skickar 'error' förstörs källan inte automatiskt.

Uppströmsströmmen håller sitt filhandtag öppet — en resursläcka.
En ohanterad 'error'-händelse kastar ett fel och kan krascha processen.

För att använda pipe() säkert måste ni lägga till en error-hanterare för varje ström och manuellt förstöra de andra. Denna standardkod är precis det som pipeline() tar bort.

Här kommer stream.pipeline()

stream.pipeline() kopplar ihop en serie strömmar, vidarebefordrar backpressure, vidarebefordrar fel och förstör alla strömmar i kedjan när någon av dem misslyckas eller avslutas.

Den tar emot strömmarna i ordning, följda av en callback som körs en gång med ett fel (eller null vid framgång):

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

Den promise-baserade pipeline()

I modern kod använder ni promise-versionen från stream/promises. Den uppfylls vid framgång och avvisas vid fel, så en enda try/catch täcker hela kedjan och städningen.

Detta är det rekommenderade sättet att koppla strömmar i asynkrona backend-hanterare.

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

En körbar pipeline i minnet

Ni behöver inga filer för att se pipeline() fungera. Readable.from() omvandlar vilken itererbar datastruktur som helst till en ström, och en Transform kan konvertera varje block till versaler. Allt körs fristående.

Observera att fel från vilket steg som helst skulle avvisa den väntade pipeline().

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: justera bufferten

Varje ström har en highWaterMark (standardvärdet är 16 KB för byte-strömmar och 16 objekt för objektläge). Det är tröskeln där write() returnerar false och läsningen pausas.

En större highWaterMark ökar genomströmningen men använder mer minne per ström.
En mindre sådan aktiverar backpressure tidigare och begränsar minnesanvändningen mer strikt.

Det är en bufferttröskel, inte en hård gräns – men det är reglaget som styr hur aggressivt backpressure aktiveras.

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() i en HTTP-hanterare

Ett vanligt misstag i backend är att buffra en hel uppladdning eller nedladdning i minnet innan svaret skickas. Genom att strömma svarskroppen med pipeline() hålls minnesanvändningen jämn och allt stängs ned om klienten kopplar från.

Eftersom HTTP-svaret är en Writable gör backpressure från en långsam klient automatiskt att filläsningen stryps.

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

Snabb kontroll

Du strömmar en fil till en långsam klient genom en gzip-transformering. Vilket tillvägagångssätt hanterar backpressure säkert och städar upp alla strömmar om klienten kopplar från mitt under överföringen?

Besvara frågan i den interaktiva lektionen

Sammanfattning

Viktiga slutsatser för säker sammankoppling av strömmar:

Backpressure hindrar en snabb producent från att överbelasta en långsam konsument. Om det booleska returvärdet från write() ignoreras är det grundorsaken till att strömmars minnesanvändning växer okontrollerat.
pipe() hanterar backpressure men inte vidarebefordran av fel eller städning – en fälla som kan läcka filbeskrivare.
stream.pipeline() (med callback eller versionen i stream/promises) vidarebefordrar backpressure, vidarebefordrar fel och förstör alla strömmar i kedjan.
highWaterMark styr hur tidigt backpressure aktiveras, vilket innebär en avvägning mellan minne och genomströmning.
I HTTP-hanterare bör du strömma med pipeline() i stället för att buffra hela nyttolaster.

Gratis att börja

Lär dig JavaScript med en AI-lärare – gratis

Skriv och kör riktig kod i webbläsaren, få omedelbar hjälp av en AI-lärare dygnet runt och fortsätt där du slutade – på webben eller i appen.

Kurser: 22
Lektioner: 92

Börja lära dig gratis Visa hela kursen

Vanliga frågor

Är lektionen ”Backpressure, pipe() och verktyget pipeline()” gratis?

Ja – hela texten till ”Backpressure, pipe() och verktyget pipeline()” kan läsas gratis här på webben. Om Ni vill öva interaktivt med en inbyggd kodredigerare och en AI-handledare som är tillgänglig dygnet runt och låsa upp resten av kursen i Bootcamp i backendutveckling med Node.js, kan Ni uppgradera till CoddyKit PRO. Kursen i Bootcamp i backendutveckling med Node.js innehåller totalt 4 lektioner.

Vad lär jag mig i ”Backpressure, pipe() och verktyget pipeline()”?

Diagnostisera minnesuppsvällning och koppla strömmar säkert med pipeline() för att vidarebefordra fel och respektera backpressure. Ni övar på Bootcamp i backendutveckling med Node.js med praktisk kod som körs direkt i webbläsaren, medan en AI-handledare som är tillgänglig dygnet runt svarar på Era frågor under lektionen.

Behöver jag någon erfarenhet för att börja lära mig Bootcamp i backendutveckling med Node.js?

Du behöver inga förkunskaper. Utbildningen i Bootcamp i backendutveckling med Node.js på CoddyKit är upplagd för allt från nybörjare till avancerade elever, så att du kan börja här eller från början och gå fram i din egen takt. Detta är lektion 3 av 4.

Hur lång tid tar lektionen ”Backpressure, pipe() och verktyget pipeline()”?

De flesta CoddyKit-lektioner tar cirka 5–10 minuter. Varje lektion är kort och interaktiv, så att du gör stadiga framsteg och kan fortsätta precis där du slutade – på webben eller i appen.

Kan jag skriva och köra kod i den här Bootcamp i backendutveckling med Node.js-lektionen?

Ja. Varje Bootcamp i backendutveckling med Node.js-lektion innehåller en inbyggd kodredigerare, så att du kan skriva och köra riktig kod direkt i webbläsaren och få omedelbar AI-feedback – utan lokal installation.

Alla lektioner i den här kursen

Readable-, Writable-, Duplex- och Transform-strömmars interna funktion
Implementera anpassade Transform-strömmar med _transform och _flush
Backpressure, pipe() och verktyget pipeline()
Asynkrona iteratorer och for-await-of över strömmar

← Tillbaka till Bootcamp i backendutveckling med Node.js