Bootcamp i backendudvikling med Node.js · Lektion

Interne detaljer i Readable-, Writable-, Duplex- og Transform-streams

Forstå de fire streamtyper, og hvordan den interne buffer og highWaterMark styrer deres adfærd.

Lektion 1 af 413 trin

Interne detaljer i Readable-, Writable-, Duplex- og Transform-streams er en gratis Bootcamp i backendudvikling med Node.js-lektion på CoddyKit. Dette er lektion 1 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 streams findes

Node.js-streams lader dig behandle data stykke for stykke i stedet for at indlæse alt i hukommelsen på én gang. Det er afgørende i backendarbejde som levering af store filer, videresendelse af HTTP-indhold eller gennemkørsel af databaseeksporter.

Readable — en kilde, du læser FRA (fillæsning, HTTP-anmodning)
Writable — et mål, du skriver TIL (filskrivning, HTTP-svar)
Duplex — både læsbar og skrivbar med uafhængige kanaler (TCP-socket)
Transform — en Duplex, hvor outputtet er en funktion af inputtet (gzip, kryptering)

Alle disse understøttes af en intern buffer, der styres af ét tal: highWaterMark.

Den interne buffer og highWaterMark

Hver stream har en intern buffer i sin _readableState eller _writableState. highWaterMark (HWM) er tærsklen, ikke en hård grænse, hvor streamen signalerer, at den har bufferet "nok".

Standard-HWM for byte-streams er 16 KB (16384 bytes).
I objekttilstand tæller HWM objekter, som standard 16.

Når en Readables buffer fyldes til HWM, holder den op med at hente fra kilden. Når en Writables buffer overskrider HWM, returnerer write() false – signalet, der kaldes backpressure.

const fs = require('fs');

const rs = fs.createReadStream('/etc/hostname', { highWaterMark: 4 });
console.log('configured HWM:', rs.readableHighWaterMark);

rs.on('data', (chunk) => {
  console.log('chunk of', chunk.length, 'bytes:', JSON.stringify(chunk.toString()));
});
rs.on('end', () => console.log('done'));

Readable-streams: flydende eller sat på pause

En Readable fungerer i én af to tilstande:

Paused (standard): Du skal kalde read() eksplicit for at hente data.
Flowing: Data skubbes til dig via 'data'-hændelser, så hurtigt som de ankommer.

Hvis du tilføjer en 'data'-lytter eller kalder .pipe(), skifter streamen til flydende tilstand. Hvis du kalder .pause(), skifter den tilbage. Det er vigtigt at forstå dette for at styre hukommelsen.

const { Readable } = require('stream');

const r = Readable.from(['a', 'b', 'c']);

// Paused mode: pull explicitly
r.on('readable', () => {
  let chunk;
  while ((chunk = r.read()) !== null) {
    console.log('pulled:', chunk);
  }
});
r.on('end', () => console.log('stream finished'));

Implementering af en brugerdefineret Readable

Hvis du vil bygge din egen kilde, skal du udvide Readable og implementere _read(size). I den kalder du this.push(chunk) for at fylde bufferen og this.push(null) for at signalere slutningen af streamen (EOF).

Den afgørende detalje er, at den interne buffer har nået HWM, når push() returnerer false. En velopdragen producent holder op med at skubbe, indtil _read kaldes igen.

const { Readable } = require('stream');

class Counter extends Readable {
  constructor(max) {
    super({ objectMode: true, highWaterMark: 2 });
    this.max = max;
    this.current = 1;
  }
  _read() {
    if (this.current > this.max) {
      this.push(null); // EOF
      return;
    }
    const keepGoing = this.push({ n: this.current++ });
    console.log('pushed, buffer wants more:', keepGoing);
  }
}

Readable.from([]); // noop
const c = new Counter(5);
c.on('data', (obj) => console.log('consumed:', obj.n));
c.on('end', () => console.log('all consumed'));

Writable-streams og returværdien fra write()

En Writable bufferer indkommende chunks og tømmer dem via _write(chunk, encoding, callback). Du SKAL kalde callback, når hver chunk er behandlet – det er sådan, streamen ved, at den kan tømme sin buffer og acceptere flere.

Returværdien fra write() er dit backpressure-signal:

true — bufferen er under HWM, så fortsæt med at skrive.
false — bufferen er på eller over HWM, så du BØR stoppe og vente på 'drain'-hændelsen.

const { Writable } = require('stream');

class SlowSink extends Writable {
  constructor() {
    super({ highWaterMark: 8 });
  }
  _write(chunk, enc, cb) {
    console.log('writing', chunk.length, 'bytes');
    setTimeout(cb, 50); // simulate slow I/O
  }
}

const sink = new SlowSink();
const ok = sink.write(Buffer.alloc(16));
console.log('write returned:', ok); // false -> over HWM
sink.once('drain', () => console.log('drained, safe to write again'));
sink.end(() => console.log('finished'));

Manuel håndtering af backpressure

Hvis du ignorerer en false-værdi fra write() og fortsætter med at skrive, vokser den interne buffer uden grænse, og processen kan løbe tør for hukommelse. Det korrekte manuelle mønster er at sætte produktionen på pause, indtil 'drain' udløses.

I praksis skriver du sjældent dette i hånden – .pipe() og pipeline() gør det for dig – men kendskab til mekanikken forklarer, HVORFOR piping er sikkert.

const { Writable } = require('stream');

const sink = new Writable({
  highWaterMark: 4,
  write(chunk, enc, cb) { setTimeout(cb, 20); }
});

let i = 0;
function writeMore() {
  let ok = true;
  while (i < 10 && ok) {
    ok = sink.write(String(i++));
  }
  if (i < 10) {
    console.log('backpressure at i =', i, '-> wait for drain');
    sink.once('drain', writeMore);
  } else {
    sink.end(() => console.log('done'));
  }
}
writeMore();

pipe(): Automatisk flowkontrol

readable.pipe(writable) forbinder en kilde med et mål og respekterer automatisk backpressure: Når destinationen returnerer false, kalder pipe source.pause(); ved 'drain' kalder den source.resume().

Ulempen ved bare .pipe() er fejlhåndtering: Hvis kilden fejler, lukkes destinationen IKKE automatisk, hvilket kan lække fildeskriptorer. Foretræk stream.pipeline() i produktion.

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

// gzip a file: Readable -> Transform -> Writable
fs.createReadStream('input.txt')
  .pipe(zlib.createGzip())
  .pipe(fs.createWriteStream('input.txt.gz'))
  .on('finish', () => console.log('compressed'));

Duplex-streams: To uafhængige kanaler

En Duplex-stream er både Readable og Writable, men de to sider er uafhængige – det, du skriver, vises ikke automatisk på læsesiden. En TCP-socket er det klassiske eksempel: Bytes, du skriver, sendes til modparten, og bytes, du læser, kommer fra modparten.

Hvis du vil implementere en, skal du angive både _read og _write. Hver side har sin egen buffer og sin egen highWaterMark.

const { Duplex } = require('stream');

class Echo extends Duplex {
  constructor() {
    super();
    this.queue = [];
  }
  _write(chunk, enc, cb) {
    this.queue.push(chunk.toString().toUpperCase());
    cb();
  }
  _read() {
    const item = this.queue.shift();
    this.push(item !== undefined ? item : null);
  }
}

const d = new Echo();
d.on('data', (c) => console.log('read side:', c.toString()));
d.write('hello');
d.write('world');
d.end();

Transform-streams: Output afledt af input

En Transform er en særlig Duplex, hvor læsesiden beregnes ud fra skrivesiden. I stedet for separate _read/_write implementerer du en enkelt _transform(chunk, encoding, callback) og sender resultater ud via this.push() eller callbackens andet argument.

En valgfri _flush(callback) kører én gang til sidst – perfekt til at sende afsluttende data ud, f.eks. en endelig kontrolsum eller en afsluttende klamme.

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

class UpperCase extends Transform {
  _transform(chunk, enc, cb) {
    cb(null, chunk.toString().toUpperCase());
  }
  _flush(cb) {
    this.push('\n-- END --\n');
    cb();
  }
}

const t = new UpperCase();
t.on('data', (c) => process.stdout.write(c.toString()));
t.write('node ');
t.write('streams');
t.end();

Objekttilstand og optælling med HWM

Som standard flytter streams Buffers/strenge, og HWM tæller bytes. Angiv { objectMode: true }, så streamen flytter vilkårlige JS-værdier, mens HWM i stedet tæller objekter.

Standard-HWM i byte-tilstand: 16384 bytes
Standard-HWM i objekttilstand: 16 objekter

Det er vigtigt i backend-pipelines: En Transform, der parser NDJSON, kan læse rå bytes (skrivesiden i byte-tilstand), men udsende fortolkede objekter (læsesiden i objekttilstand) ved hjælp af readableObjectMode.

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

// Bytes in, objects out
class NdjsonParse extends Transform {
  constructor() {
    super({ writableObjectMode: false, readableObjectMode: true });
    this.buf = '';
  }
  _transform(chunk, enc, cb) {
    this.buf += chunk.toString();
    const lines = this.buf.split('\n');
    this.buf = lines.pop();
    for (const line of lines) {
      if (line.trim()) this.push(JSON.parse(line));
    }
    cb();
  }
}

const p = new NdjsonParse();
p.on('data', (o) => console.log('parsed object:', o));
p.write('{"id":1}\n{"id":2}\n');
p.end();

pipeline(): Sammensætning klar til produktion

stream.pipeline() kæder et vilkårligt antal streams sammen og videresender fejl og rydder op i alle streams (ved at destruere dem), når en af dem fejler eller afsluttes. Det forhindrer lækkede fildeskriptorer og fastlåste sockets.

Den promise-baserede form (require('stream/promises')) integreres problemfrit med async/await i rutehåndteringer.

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

async function gzipFile(src, dest) {
  await pipeline(
    fs.createReadStream(src),
    zlib.createGzip(),
    fs.createWriteStream(dest)
  );
  console.log('pipeline complete:', dest);
}

gzipFile('access.log', 'access.log.gz').catch((err) => {
  console.error('pipeline failed, all streams destroyed:', err.message);
});

Hurtigt tjek: Backpressure-signal

Du skriver et stort datasæt til en brugerdefineret Writable-stream i en løkke. Du vil undgå ukontrolleret hukommelsesvækst ved at respektere backpressure. Hvilket signal fortæller dig, at du skal stoppe med at skrive og vente?

Besvar den i den interaktive lektion

Opsummering

Du forstår nu de fire strømtyper og buffermekanikken bag dem:

Readable — kilde; implementér _read, indsæt data og push(null) for EOF; tilstandene flowing og paused.
Writable — modtager; implementér _write, og kald dens callback; når write() returnerer false, betyder det modtryk, så vent på 'drain'.
Duplex — uafhængige læse- og skrivekanaler, hver med sin egen buffer og HWM (f.eks. en TCP-socket).
Transform — output, der udledes af input via _transform, med valgfri _flush.

highWaterMark (16 KB bytes / 16 objekter som standard) er en tærskel, ikke en hård grænse, som styrer, hvornår buffere signalerer "fuld". Foretræk altid pipeline() frem for en ubeskyttet .pipe() i produktion, så fejlvideregivelse og oprydning fungerer korrekt.

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 “Interne detaljer i Readable-, Writable-, Duplex- og Transform-streams” gratis?

Ja — hele teksten til “Interne detaljer i Readable-, Writable-, Duplex- og Transform-streams” 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 “Interne detaljer i Readable-, Writable-, Duplex- og Transform-streams”?

Forstå de fire streamtyper, og hvordan den interne buffer og highWaterMark styrer deres adfærd. 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 1 af 4.

Hvor lang tid tager lektionen “Interne detaljer i Readable-, Writable-, Duplex- og Transform-streams”?

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