Bootcamp i backendudvikling med Node.js · Lektion

Backpressure, pipe() og pipeline()-værktøjet

Diagnosticér voksende hukommelsesforbrug, og forbind streams sikkert med pipeline(), så fejl videregives og backpressure respekteres.

Lektion 3 af 413 trin

Backpressure, pipe() og pipeline()-værktøjet er en gratis Bootcamp i backendudvikling med Node.js-lektion på CoddyKit. Dette er lektion 3 af 4. Du kan læse hele lektionen gratis nedenfor — og derefter øve dig praktisk i browseren med en indbygget kodeeditor og en AI-vejleder, der er tilgængelig døgnet rundt. Den er en del af læringsforløbet i Bootcamp i backendudvikling med Node.js, og dine fremskridt synkroniseres på tværs af nettet og CoddyKit-appen. Bootcamp i backendudvikling med Node.js-kurset indeholder 4 lektioner i alt.

Hvorfor hukommelsen vokser i strømme

En Node.js-Readable-strøm kan producere data hurtigere, end en Writable kan forbruge dem. Hvis du aldrig beder producenten sætte farten ned, hober uforbrugte blokke sig op i en intern buffer, og proceshukommelsen vokser, indtil GC ikke længere kan følge med.

En langsom disk, en langsom netværkssocket eller en langsom databaseskrivning er den typiske forbruger.
En hurtig fillæsning eller en HTTP-upload er den typiske producent.

Den mekanisme, der får producenten til at vente på forbrugeren, kaldes modtryk. Forkert brug af strømme betyder næsten altid, at modtryk blev ignoreret.

Den naive (ødelagte) kopiering

Her er den klassiske hukommelsesfejl. Vi lytter efter data og kalder dst.write() for hver blok, men ignorerer returværdien.

Hvis dst er langsommere end src, samler de ikke-skrevne blokke sig i dsts buffer uden nogen øvre grænse. For en fil på flere gigabytes kan det opbruge 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());

Hvad write() faktisk returnerer

writable.write(chunk) returnerer en boolsk værdi:

true — den interne buffer er under highWaterMark; fortsæt med at skrive.
false — bufferen er fuld; du skal stoppe med at skrive og vente på hændelsen 'drain', før du sender mere.

Ved at respektere denne returværdi anvender du modtryk manuelt. Producenten skal sætte farten ned, indtil forbrugeren signalerer, at bufferen er tømt.

Manuelt modtryk med pause/resume

Manuelt betyder modtryk følgende: Når write() returnerer false, kalder du pause() på kilden; når destinationen udsender 'drain', kalder du resume() på den.

Det fungerer, men er omstændeligt og nemt at gøre forkert — du skal også forbinde håndtering af error og end for begge strømme.

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 dette for dig

readable.pipe(writable) forbinder automatisk den samme pause-/genoptagelses-/tømningsmekanisme og respekterer modtryk fra starten.

Den returnerer destinationsstrømmen, så du kan kæde transformeringer sammen:

src.pipe(gzip).pipe(dst)

Til de fleste enkle kopieringer er pipe() langt bedre end den manuelle løkke ovenfor.

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 skjulte fejl i pipe()

pipe() håndterer modtryk, men videresender ikke fejl. Hvis gzip eller dst udsender 'error', ødelægges kilden ikke automatisk.

Den foregående strøm holder sin fildeskriptor åben — en ressourcelækage.
En uhåndteret 'error'-hændelse kastes og kan få processen til at gå ned.

Hvis du vil bruge pipe() sikkert, skal du tilføje en error-håndtering til hver strøm og manuelt ødelægge de andre. Det gentagne standardarbejde er netop det, som pipeline() fjerner.

Introduktion til stream.pipeline()

stream.pipeline() forbinder en række strømme, videresender modtryk, videresender fejl og ødelægger alle strømme i kæden, når en af dem fejler eller afsluttes.

Den tager strømmene i rækkefølge efterfulgt af en callback, der kaldes én gang med en fejl (eller null ved 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');
    }
  }
);

Den promise-baserede pipeline()

I moderne kode skal du bruge promise-versionen fra stream/promises. Den opfyldes ved succes og afvises ved fejl, så én enkelt try/catch dækker hele kæden og oprydningen.

Det er den anbefalede måde at forbinde strømme i asynkrone backend-håndteringer.

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 hukommelsen

Du behøver ikke filer for at se pipeline() fungere. Readable.from() omdanner enhver iterérbar værdi til en strøm, og en Transform kan ændre hver blok til store bogstaver. Det hele kører selvstændigt.

Bemærk, hvordan fejl fra ethvert trin ville afvise den afventede 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: Tilpasning af bufferen

Hver strøm har en highWaterMark (standardværdien er 16 KB for bytestrømme og 16 objekter i objekttilstand). Det er den tærskel, hvor write() returnerer false, og læsninger sættes på pause.

En større highWaterMark øger gennemløbet, men bruger mere hukommelse pr. strøm.
En mindre værdi aktiverer tilbagetryk tidligere og begrænser hukommelsesforbruget mere.

Det er en tærskel for bufferlagring, ikke en hård grænse — men det er den indstilling, der styrer, hvor aggressivt tilbagetryk aktiveres.

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-handler

En almindelig fejl i backend-kode er at gemme en hel upload eller download i hukommelsen, før der svares. Når svarlegemet sendes som en strøm med pipeline(), forbliver hukommelsesforbruget stabilt, og alt lukkes ned, hvis klienten afbryder forbindelsen.

Fordi HTTP-svaret er en Writable, begrænser tilbagetryk fra en langsom klient automatisk læsningen af filen.

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

Hurtigt tjek

Du sender en fil som strøm til en langsom klient gennem en gzip-transformation. Hvilken tilgang respekterer tilbagetryk korrekt og rydder op efter alle strømme, hvis klienten afbryder forbindelsen midt i overførslen?

Besvar den i den interaktive lektion

Opsummering

De vigtigste pointer om sikker sammenkobling af strømme:

Tilbagetryk forhindrer en hurtig producent i at overvælde en langsom forbruger; hvis returværdien fra write() ignoreres, er det den grundlæggende årsag til overdreven hukommelsesbrug i strømme.
pipe() håndterer tilbagetryk, men ikke videresendelse af fejl eller oprydning — en fælde, der kan lække filbeskrivelser.
stream.pipeline() (med callback eller versionen stream/promises) videresender tilbagetryk, videresender fejl og ødelægger alle strømme i kæden.
highWaterMark styrer, hvor tidligt tilbagetryk aktiveres, som en afvejning mellem hukommelse og gennemløb.
I HTTP-handlere skal du bruge pipeline() til at sende data som strøm i stedet for at gemme hele nyttelasten i hukommelsen.

Gratis at komme i gang

Lær JavaScript med en AI-underviser — gratis

Skriv og kør rigtig kode i din browser, få øjeblikkelig hjælp fra en AI-underviser døgnet rundt, og fortsæt, hvor du slap, på web eller i appen.

Kurser: 22
Lektioner: 92

Start gratis Se hele kurset

Ofte stillede spørgsmål

Er lektionen “Backpressure, pipe() og pipeline()-værktøjet” gratis?

Ja — hele teksten til “Backpressure, pipe() og pipeline()-værktøjet” kan læses gratis her på nettet. Hvis du vil øve dig interaktivt med en indbygget kodeeditor og en AI-vejleder døgnet rundt og få adgang til resten af Bootcamp i backendudvikling med Node.js-kurset, skal du opgradere til CoddyKit PRO. Bootcamp i backendudvikling med Node.js-kurset indeholder 4 lektioner i alt.

Hvad lærer jeg i “Backpressure, pipe() og pipeline()-værktøjet”?

Diagnosticér voksende hukommelsesforbrug, og forbind streams sikkert med pipeline(), så fejl videregives og backpressure respekteres. Du øver dig i Bootcamp i backendudvikling med Node.js med praktisk kode, som du kører direkte i browseren, og en AI-vejleder døgnet rundt besvarer dine spørgsmål, mens du arbejder dig gennem lektionen.

Skal jeg have erfaring for at begynde på Bootcamp i backendudvikling med Node.js?

Der kræves ingen tidligere erfaring. Bootcamp i backendudvikling med Node.js på CoddyKit er tilrettelagt for både begyndere og øvede, så du kan starte her eller fra begyndelsen og lære i dit eget tempo. Dette er lektion 3 af 4.

Hvor lang tid tager lektionen “Backpressure, pipe() og pipeline()-værktøjet”?

De fleste CoddyKit-lektioner tager cirka 5–10 minutter. Hver lektion er kort og interaktiv, så du gør løbende fremskridt og kan fortsætte, hvor du slap – på både web og app.

Kan jeg skrive og køre kode i denne Bootcamp i backendudvikling med Node.js-lektion?

Ja. Alle Bootcamp i backendudvikling med Node.js-lektioner har en indbygget kodeeditor, så du kan skrive og køre rigtig kode direkte i din browser og få øjeblikkelig feedback fra AI – uden lokal opsætning.

Alle lektioner i dette kursus

Interne detaljer i Readable-, Writable-, Duplex- og Transform-streams
Implementering af brugerdefinerede Transform-streams med _transform og _flush
Backpressure, pipe() og pipeline()-værktøjet
Async-iteratorer og for-await-of over streams

← Tilbage til Bootcamp i backendudvikling med Node.js