Bootcamp i backendutveckling med Node.js · Lektion

Readable-, Writable-, Duplex- och Transform-strömmars interna funktion

Förstå de fyra strömtyperna och hur den interna bufferten och highWaterMark styr deras beteende.

Lektion 1 av 413 steg

Readable-, Writable-, Duplex- och Transform-strömmars interna funktion är en gratis lektion i Bootcamp i backendutveckling med Node.js på CoddyKit. Detta är lektion 1 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 streams finns

Node.js-streams låter dig bearbeta data del för del i stället för att läsa in allt i minnet på en gång. Detta är viktigt i backendarbete, till exempel när du skickar stora filer, vidarebefordrar HTTP-innehåll eller strömmar databasexporter.

Readable — en källa som du läser FRÅN (filläsning, HTTP-begäran)
Writable — ett mål som du skriver TILL (filskrivning, HTTP-svar)
Duplex — både läsbar och skrivbar, med oberoende kanaler (TCP-socket)
Transform — en Duplex där utdata är en funktion av indata (gzip, kryptering)

Alla dessa har en intern buffert som styrs av ett enda tal: highWaterMark.

Den interna bufferten och highWaterMark

Varje stream har en intern buffert i sitt _readableState eller _writableState. highWaterMark (HWM) är tröskeln, inte en hård gräns, där streamen signalerar att den har buffrat "tillräckligt".

Standardvärdet för HWM i byte-streams är 16 KB (16384 byte).
I object mode räknar HWM objekt, med standardvärdet 16.

När en Readables buffert fylls till HWM slutar den att hämta från källan. När en Writables buffert överskrider HWM returnerar write() false – signalen som kallas 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: flowing kontra paused

En Readable arbetar i ett av två lägen:

Paused (standard): du måste uttryckligen anropa read() för att hämta data.
Flowing: data skickas till dig via 'data'-händelser så snabbt som den anländer.

Om du lägger till en 'data'-lyssnare eller anropar .pipe() växlar streamen till flowing-läge. Om du anropar .pause() växlar den tillbaka. Att förstå detta är viktigt för att kunna styra minnesanvändningen.

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

Implementera en egen Readable

Om du vill bygga en egen källa utökar du Readable och implementerar _read(size). Där anropar du this.push(chunk) för att fylla bufferten och this.push(null) för att signalera slutet på streamen (EOF).

Den viktiga detaljen är att den interna bufferten har nått HWM när push() returnerar false. En välfungerande producent slutar då att skicka data tills _read anropas 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 och returvärdet från write()

En Writable buffrar inkommande chunks och tömmer dem via _write(chunk, encoding, callback). Du MÅSTE anropa callback när varje chunk har behandlats – det är så streamen vet att den kan tömma bufferten och ta emot mer.

Returvärdet från write() är din backpressure-signal:

true — bufferten ligger under HWM, fortsätt skriva.
false — bufferten ligger vid eller över HWM, du BÖR sluta och vänta på händelsen 'drain'.

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

Hantera backpressure manuellt

Om du ignorerar false från write() och fortsätter skriva växer den interna bufferten utan gräns, och processen kan få slut på minne. Det korrekta manuella mönstret är att pausa produktionen tills 'drain' utlöses.

I praktiken skriver du sällan detta för hand – .pipe() och pipeline() gör det åt dig – men om du känner till mekaniken förstår du varför piping är säkert.

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 flödeskontroll

readable.pipe(writable) kopplar en källa till ett mål och hanterar automatiskt backpressure: när destinationen returnerar false anropar pipe source.pause(); vid 'drain' anropar den source.resume().

Nackdelen med en fristående .pipe() är felhantering: om källan får ett fel stängs destinationen INTE automatiskt, vilket kan läcka filidentifierare. Föredra 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: två oberoende kanaler

En Duplex-stream är både Readable och Writable, men de två sidorna är oberoende – det du skriver visas inte automatiskt på lässidan. En TCP-socket är det klassiska exemplet: byte som du skriver skickas till motparten, medan byte som du läser kommer från motparten.

För att implementera en sådan tillhandahåller du både _read och _write. Varje sida har sin egen buffert och sitt eget 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: utdata härleds från indata

En Transform är en särskild Duplex där lässidan beräknas från skrivsidan. I stället för separata _read/_write implementerar du en enda _transform(chunk, encoding, callback) och skickar ut resultat via this.push() eller callbackens andra argument.

En valfri _flush(callback) körs en gång i slutet – perfekt för att skicka ut avslutande data (till exempel en slutlig kontrollsumma eller en avslutande hakparentes).

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

Objektläge och HWM-räkning

Som standard hanterar streams Buffers/strängar och HWM räknar byte. Skicka { objectMode: true } så hanterar streamen godtyckliga JS-värden, medan HWM i stället räknar objekt.

Standardvärde för HWM i byteläge: 16384 byte
Standardvärde för HWM i objektläge: 16 objekt

Detta är viktigt i backend-pipelines: en Transform som tolkar NDJSON kan läsa råa byte (skrivsidan, byteläge) men skicka ut tolkade objekt (lässidan, objektläge) med hjälp av 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(): komposition för produktion

stream.pipeline() kedjar ihop valfritt antal streams och, till skillnad från .pipe(), vidarebefordrar fel och städar upp alla streams (genom att förstöra dem) när någon av dem misslyckas eller avslutas. Detta förhindrar läckta filidentifierare och hängande sockets.

Den promise-baserade formen (require('stream/promises')) integreras smidigt med async/await i route handlers.

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

Snabbtest: backpressure-signalen

Du skriver en stor datamängd till en egen Writable-stream i en loop. Du vill undvika obegränsad minnestillväxt genom att respektera backpressure. Vilken signal talar om att du ska sluta skriva och vänta?

Besvara frågan i den interaktiva lektionen

Sammanfattning

Ni förstår nu de fyra strömtyperna och buffertmekaniken bakom dem:

Readable — källa; implementera _read, skicka data och push(null) för EOF; lägena flowing och paused.
Writable — mottagare; implementera _write och anropa dess callback; om write() returnerar false betyder det backpressure, så vänta på 'drain'.
Duplex — oberoende läs- och skrivkanaler, var och en med sin egen buffert och HWM (till exempel ett TCP-uttag).
Transform — utdata som härleds från indata via _transform, med valfri _flush.

highWaterMark (16 KB byte / 16 objekt som standard) är ett tröskelvärde, inte en hård gräns, som avgör när buffertar signalerar "full". Föredra alltid pipeline() framför en fristående .pipe() i produktion för korrekt vidarebefordran av fel och städning.

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 ”Readable-, Writable-, Duplex- och Transform-strömmars interna funktion” gratis?

Ja – hela texten till ”Readable-, Writable-, Duplex- och Transform-strömmars interna funktion” 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 ”Readable-, Writable-, Duplex- och Transform-strömmars interna funktion”?

Förstå de fyra strömtyperna och hur den interna bufferten och highWaterMark styr deras beteende. 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 1 av 4.

Hur lång tid tar lektionen ”Readable-, Writable-, Duplex- och Transform-strömmars interna funktion”?

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