Streams & I/O

ปัญหาของการโหลดทุกอย่างในครั้งเดียว

ลองนึกภาพการอ่านไฟล์ log ขนาด 4 GB ถ้าคุณโหลดมันทั้งหมดเข้า memory ก่อน Node ต้องจอง 4 GB ก่อนที่คุณจะดูบรรทัดเดียวได้ด้วยซ้ำ บนเครื่องที่มี RAM 2 GB process จะ crash ก่อนที่มันจะเริ่มด้วยซ้ำ

Stream แก้ปัญหานี้ด้วยการประมวลผลข้อมูล เป็น chunk — คุณเริ่มอ่าน chunk แรกในขณะที่ส่วนที่เหลือยังมาไม่ถึงจากดิสก์หรือ network memory จะคงที่ไม่ว่าต้นทางจะมีขนาดใหญ่แค่ไหน

แนวคิดเดียวกันนี้ใช้กับการเขียนด้วย: แทนที่จะ buffer response ทั้งหมดใน memory แล้วส่งทีเดียว stream จะส่งแต่ละ chunk ไปยังปลายทางทันทีที่มันมาถึง

Stream สี่ประเภท

Node.js กำหนด abstract stream สี่ประเภทไว้ใน module node:stream:

ประเภท	ทิศทาง	ตัวอย่าง
`Readable`	ข้อมูลไหล ออก	`fs.createReadStream`, `process.stdin`
`Writable`	ข้อมูลไหล เข้า	`fs.createWriteStream`, `process.stdout`
`Duplex`	สองทิศทาง	TCP socket (`net.Socket`)
`Transform`	สองทิศทางพร้อมการแปลงข้อมูล	`zlib.createGzip` (บีบอัดระหว่างทาง)

ทุก stream ก็เป็น EventEmitter ด้วย stream base class จะ emit event มาตรฐาน (data, end, error, drain, finish) ที่คุณผูก handler เข้าไปได้

โมดูลนี้ครอบคลุมอะไรบ้าง

บทเรียน	หัวข้อ
Readable Streams	event `data`/`end`/`error`, `Readable.from`, async iteration
Writable Streams	`write()`, `end()` และ event `drain`
Pipe & Pipeline	การเชื่อม stream ด้วย `.pipe()` และ `stream.pipeline()`
Backpressure	มันคืออะไร ทำไมสำคัญ และ pipeline จัดการมันอย่างไร

มองภาพแรกของ streaming

demo ด้านล่างใช้ Readable.from เพื่อสร้าง readable stream จาก array ของ string แล้วบริโภคมันทีละ chunk ด้วย for await สังเกตว่าไม่มี array เต็ม ๆ ถูกเก็บไว้ใน buffer เลย — แต่ละ item ถูก yield ตามความต้องการ

Node.js

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

async function main() {
  const source = Readable.from(['hello', ' ', 'streaming', ' ', 'world']);

  let result = '';
  for await (const chunk of source) {
    result += chunk;
    console.log('chunk: ' + JSON.stringify(chunk));
  }
  console.log('assembled: ' + result);
}

main();

Output (verified on Node 24)

chunk: "hello"
chunk: " "
chunk: "streaming"
chunk: " "
chunk: "world"
assembled: hello streaming world

Mental Model

flowchart LR
  Readable["Readable — data flows OUT"] -->|".pipe() / pipeline()"| Writable["Writable — data flows IN"]
  Readable -->|"through a"| Transform["Transform — reshapes each chunk"]
  Transform --> Writable
  Duplex["Duplex — both directions independently"]
  All["Readable, Writable, Duplex, Transform"] -->|"all extend"| EE["EventEmitter"]

ข้อมูลไหลจาก Readable ผ่าน Transform (ถ้ามี) เข้าสู่ Writable — stream ทุกประเภทก็เป็น EventEmitter ด้วย

Trade-offs

ทางเลือก	ข้อดี	ต้นทุน
Streaming (แบบ chunk)	การใช้ memory คงที่ไม่ว่าข้อมูลทั้งหมดจะมีขนาดเท่าไหร่; เริ่มประมวลผลได้ก่อนที่ต้นทางจะเสร็จ	มีชิ้นส่วนที่ต้องจัดการมากกว่าการเรียก `readFile()` ครั้งเดียว — event หรือ async iteration, backpressure, การส่งต่อ error
โหลดทุกอย่างเข้า memory ก่อน	code ง่ายที่สุดเท่าที่จะทำได้ — ไม่ต้อง wire event ผลลัพธ์ทั้งหมดพร้อมใช้ทันที	peak memory scale ตามขนาด input; input ที่ใหญ่พอจะทำให้ process crash ตรง ๆ

Common Misconceptions

“Stream เร็วกว่าการอ่านทั้งไฟล์เสมอ” — ความจริง: stream ประหยัด memory ไม่จำเป็นต้องเร็วกว่าในแง่ wall-clock สำหรับไฟล์ที่พอดีกับ RAM readFile() อาจง่ายกว่าและเร็วพอ ๆ กัน
“stream ทุกตัวต้องมี data event listener เพื่อให้ output ออกมา” — ความจริง: for await...of (async iteration) บริโภค Readable ได้โดยไม่แตะ event data เลย และจัดการ backpressure ให้อัตโนมัติไปด้วย
“Transform stream กับ Duplex stream คือสิ่งเดียวกัน” — ความจริง: ทั้งคู่เป็นสองทิศทาง แต่ output ของ Transform ได้มาจาก input ของมัน (ฝั่งหนึ่งป้อนอีกฝั่ง); ฝั่ง readable และ writable ของ Duplex เป็นอิสระต่อกัน — TCP socket อ่านและเขียนข้อมูลที่ไม่เกี่ยวข้องกันในแต่ละฝั่ง

อ่านโค้ดที่ AI เขียน

code เรียก fs.readFileSync() หรือ fs.readFile() เพื่อโหลดไฟล์แค่จะเขียนมันกลับออกไปหรือส่งต่อไปที่อื่นทันทีหรือเปล่า ทั้งที่ stream (หรือ pipeline) จะไม่เก็บทั้งไฟล์ไว้ใน memory เลย?
snippet ผูก data listener กับ Readable โดยไม่เคยตรวจ backpressure ปลายทางเลย สมมติว่าพฤติกรรม pause/resume อัตโนมัติของ .pipe() จะเกิดขึ้นทั้งที่ไม่ได้เรียก .pipe() จริง ๆ หรือเปล่า?
code ปฏิบัติต่อ Transform และ Duplex ราวกับใช้แทนกันได้ — extend Duplex ทั้งที่เจตนาจริงคือ “รับ X เข้ามา สร้าง Y ที่ได้มาออกไป” (ซึ่งคือ Transform) หรือเปล่า?

🐛 หา bug ในโค้ดนี้

const fs = require('node:fs');

// Loads the whole file into memory just to copy it elsewhere
var data = fs.readFileSync('./huge-input.log');
fs.writeFileSync('./huge-output.log', data);

เฉลย

โค้ดนี้อ่านทั้งไฟล์เข้า Buffer เดียวใน memory ก่อนที่จะเขียนมันกลับออกไป ทั้งที่ไม่มีอะไรตรงนี้ต้องตรวจสอบหรือแปลงเนื้อหาเลย — สำหรับไฟล์ log ขนาดหลายกิกะไบต์ นี่อาจทำให้ memory ที่มีอยู่หมดหรือแค่เปลืองเปล่า ๆ fs.createReadStream('./huge-input.log').pipe(fs.createWriteStream('./huge-output.log')) (หรือ pipeline ของ stream/promises ที่ครอบคลุมทีหลังในโมดูลนี้) จะ copy ไฟล์เป็น chunk ที่มีขอบเขตโดยไม่เคยถือมากกว่าหนึ่ง chunk ใน memory เลย