ภายในสตรีมแบบอ่านได้ เขียนได้ สองทาง และแปลงข้อมูล
ทำความเข้าใจสตรีมทั้งสี่ประเภท และวิธีที่บัฟเฟอร์ภายในกับ highWaterMark ควบคุมพฤติกรรมของสตรีม
ภายในสตรีมแบบอ่านได้ เขียนได้ สองทาง และแปลงข้อมูล เป็นบทเรียน Node.js Backend Development Bootcamp ฟรีบน CoddyKit นี่คือบทเรียนที่ 1 จากทั้งหมด 4 บทเรียน คุณสามารถอ่านบทเรียนทั้งหมดด้านล่างฟรี — จากนั้นลองปฏิบัติด้วยตัวคุณเองในเบราว์เซอร์พร้อมตัวแก้ไขโค้ดในตัวและติวเตอร์ AI ตลอด 24/7 บทเรียนนี้เป็นส่วนหนึ่งของเส้นทางการเรียน Node.js Backend Development Bootcamp และความก้าวหน้าของคุณจะซิงค์ข้ามเว็บและแอป CoddyKit คอร์ส Node.js Backend Development Bootcamp มีบทเรียนทั้งหมด 4 บทเรียน
บางส่วนของบทเรียนนี้ยังไม่ได้รับการแปล และแสดงเป็นภาษาอังกฤษ
Why Streams Exist
Node.js streams let you process data piece by piece instead of loading everything into memory at once. This is essential for backend work like serving large files, proxying HTTP bodies, or piping database exports.
- Readable — a source you read FROM (file read, HTTP request)
- Writable — a sink you write TO (file write, HTTP response)
- Duplex — both readable and writable, independent channels (TCP socket)
- Transform — a Duplex where the output is a function of the input (gzip, encryption)
Every one of these is backed by an internal buffer governed by a single number: highWaterMark.
The Internal Buffer & highWaterMark
Each stream keeps an internal buffer in its _readableState or _writableState. The highWaterMark (HWM) is the threshold, not a hard limit, at which the stream signals it has buffered "enough".
- Default HWM for byte streams is 16 KB (16384 bytes).
- In object mode the HWM counts objects, defaulting to 16.
When a Readable's buffer fills to the HWM it stops pulling from the source. When a Writable's buffer exceeds the HWM, write() returns false — the signal known as 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 vs Paused
A Readable operates in one of two modes:
- Paused (default): you must call
read()explicitly to pull data. - Flowing: data is pushed at you via
'data'events as fast as it arrives.
Attaching a 'data' listener or calling .pipe() switches the stream into flowing mode. Calling .pause() switches it back. Understanding this is key to controlling memory.
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'));Implementing a Custom Readable
To build your own source, extend Readable and implement _read(size). Inside it you call this.push(chunk) to feed the buffer, and this.push(null) to signal end-of-stream (EOF).
The crucial detail: when push() returns false, the internal buffer has hit the HWM. A well-behaved producer stops pushing until _read is called again.
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 & the write() Return Value
A Writable buffers incoming chunks and flushes them via _write(chunk, encoding, callback). You MUST call the callback when each chunk is processed — that is how the stream knows to drain its buffer and accept more.
The return value of write() is your backpressure signal:
true— buffer is below HWM, keep writing.false— buffer is at/over HWM, you SHOULD stop and wait for the'drain'event.
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'));Respecting Backpressure Manually
If you ignore a false from write() and keep writing, the internal buffer grows without bound and your process can run out of memory. The correct manual pattern is to pause production until 'drain' fires.
In practice you rarely write this by hand — .pipe() and pipeline() do it for you — but knowing the mechanics explains WHY piping is safe.
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(): Automatic Flow Control
readable.pipe(writable) wires a source to a sink and automatically honors backpressure: when the destination returns false, pipe calls source.pause(); on 'drain' it calls source.resume().
The downside of bare .pipe() is error handling: if the source errors, the destination is NOT closed automatically, which can leak file descriptors. Prefer stream.pipeline() in production.
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: Two Independent Channels
A Duplex stream is both Readable and Writable, but the two sides are independent — what you write does not automatically appear on the read side. A TCP socket is the canonical example: bytes you write go out to the peer, bytes you read come in from the peer.
To implement one, provide both _read and _write. Each side has its own buffer and its own 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 Derived from Input
A Transform is a special Duplex where the readable side is computed from the writable side. Instead of separate _read/_write, you implement a single _transform(chunk, encoding, callback) and emit results via this.push() or the callback's second argument.
An optional _flush(callback) runs once at the end — perfect for emitting trailing data (e.g. a final checksum or closing bracket).
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();Object Mode & HWM Counting
By default streams move Buffers/strings and the HWM counts bytes. Pass { objectMode: true } and the stream moves arbitrary JS values, with the HWM counting objects instead.
- Byte mode default HWM: 16384 bytes
- Object mode default HWM: 16 objects
This matters for backend pipelines: a Transform parsing NDJSON might read raw bytes (writable side, byte mode) but emit parsed objects (readable side, object mode) using 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(): Production-Grade Composition
stream.pipeline() chains any number of streams and, unlike .pipe(), it propagates errors and cleans up every stream (destroying them) when any one fails or finishes. This prevents leaked file descriptors and hung sockets.
The promise-based form (require('stream/promises')) integrates cleanly with async/await in 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);
});Quick Check: Backpressure Signal
You are writing a large dataset to a custom Writable stream in a loop. You want to avoid unbounded memory growth by respecting backpressure. Which signal tells you to stop writing and wait?
Recap
You now understand the four stream types and the buffer mechanics behind them:
- Readable — source; implement
_read, push data andpush(null)for EOF; flowing vs paused modes. - Writable — sink; implement
_writeand call its callback;write()returningfalsemeans backpressure, wait for'drain'. - Duplex — independent read and write channels, each with its own buffer and HWM (e.g. TCP socket).
- Transform — output derived from input via
_transform, with optional_flush.
The highWaterMark (16 KB bytes / 16 objects by default) is a threshold, not a hard cap, governing when buffers signal "full". Always prefer pipeline() over bare .pipe() in production for correct error propagation and cleanup.
คำถามที่พบบ่อย
บทเรียน “ภายในสตรีมแบบอ่านได้ เขียนได้ สองทาง และแปลงข้อมูล” ฟรีหรือไม่
ใช่ — ข้อความเต็มของ “ภายในสตรีมแบบอ่านได้ เขียนได้ สองทาง และแปลงข้อมูล” ฟรีให้อ่านที่นี่บนเว็บ เพื่อปฏิบัติแบบโต้ตอบ (ตัวแก้ไขโค้ดในตัวและติวเตอร์ AI ตลอด 24/7) และปลดล็อคส่วนที่เหลือของคอร์ส Node.js Backend Development Bootcamp ให้อัปเกรดเป็น CoddyKit PRO คอร์ส Node.js Backend Development Bootcamp มีบทเรียนทั้งหมด 4 บทเรียน
คุณจะเรียนรู้อะไรในบทเรียน “ภายในสตรีมแบบอ่านได้ เขียนได้ สองทาง และแปลงข้อมูล”
ทำความเข้าใจสตรีมทั้งสี่ประเภท และวิธีที่บัฟเฟอร์ภายในกับ highWaterMark ควบคุมพฤติกรรมของสตรีม คุณปฏิบัติ Node.js Backend Development Bootcamp ด้วยโค้ดที่ใช้งานได้จริงที่คุณเรียกใช้โดยตรงในเบราว์เซอร์ และติวเตอร์ AI ตลอด 24/7 ตอบคำถามของคุณขณะที่คุณไปผ่านบทเรียน
คุณต้องมีประสบการณ์ก่อนที่จะเริ่มเรียน Node.js Backend Development Bootcamp หรือไม่
ไม่จำเป็นต้องมีประสบการณ์มาก่อน Node.js Backend Development Bootcamp บน CoddyKit ออกแบบมาสำหรับผู้เริ่มต้นไปจนถึงผู้เรียนขั้นสูง คุณสามารถเริ่มต้นที่นี่หรือเริ่มจากตัวแรกและเรียนด้วยความเร็วของคุณเอง นี่คือบทเรียนที่ 1 จากทั้งหมด 4 บทเรียน
บทเรียน “ภายในสตรีมแบบอ่านได้ เขียนได้ สองทาง และแปลงข้อมูล” ใช้เวลานานแค่ไหน
บทเรียน CoddyKit ส่วนใหญ่ใช้เวลาประมาณ 5–10 นาที แต่ละบทเรียนจึงสั้นและเป็นแบบโต้ตอบ คุณสามารถก้าวหน้าอย่างต่อเนื่องและกลับมาเรียนต่อจากตรงที่เพิ่งหยุดบนเว็บและแอปได้เลย
ฉันเขียนและรันโค้ดในบทเรียน Node.js Backend Development Bootcamp นี้ได้ไหม
ได้ บทเรียน Node.js Backend Development Bootcamp ทุกบทมีตัวแก้ไขโค้ดในตัว คุณจึงเขียนและรันโค้ดจริงได้เลยในเบราว์เซอร์ และได้รับข้อเสนอแนะจาก AI ในทันที — ไม่ต้องติดตั้งในเครื่องของคุณ
บทเรียนทั้งหมดในหลักสูตรนี้
- ภายในสตรีมแบบอ่านได้ เขียนได้ สองทาง และแปลงข้อมูล
- การสร้างสตรีมแปลงข้อมูลแบบกำหนดเองด้วย _transform และ _flush
- แรงดันย้อนกลับ pipe() และยูทิลิตี pipeline()
- ตัววนซ้ำแบบอะซิงโครนัสและ for-await-of บนสตรีม