Node.js 流式處理實戰：Readable、Writable、Transform 與背壓機制全解析

為什麼用流？記憶體從 GB 降到 KB

方式	處理 1GB 檔案	記憶體佔用	耗時
fs.readFile	全量讀入記憶體	1GB+	長且阻塞
fs.createReadStream	分塊處理	~64KB	立即開始

核心思想：不等待全部資料就緒，邊讀邊處理——資料像水流一樣經過管道。

---

一、四種流型別

型別	方向	典型用途
Readable	輸入	檔案讀取、HTTP 請求體
Writable	輸出	檔案寫入、HTTP 回應
Duplex	雙向	TCP Socket、WebSocket
Transform	轉換	壓縮、加密、格式轉換

繼承關係

Stream
├── Readable
├── Writable
├── Duplex (Readable + Writable)
└── Transform (Duplex，輸出基於輸入)

---

二、Readable 流

兩種模式

模式	觸發方式	特點
Paused（暫停）	預設	需手動呼叫 read()
Flowing（流動）	繫結 data 事件	自動推送資料

建立自訂 Readable

import { Readable } from 'stream';

class NumberStream extends Readable {
  constructor(max) {
    super({ objectMode: true });
    this.max = max;
    this.current = 0;
  }

  _read() {
    if (this.current < this.max) {
      this.push({ value: this.current, timestamp: Date.now() });
      this.current++;
    } else {
      this.push(null); // 結束流
    }
  }
}

const stream = new NumberStream(5);
stream.on('data', (chunk) => console.log(chunk));
// { value: 0, timestamp: ... }
// { value: 1, timestamp: ... }
// ...

從迭代器建立

import { Readable } from 'stream';

async function* generateLogs() {
  for (let i = 0; i < 100; i++) {
    await sleep(100);
    yield `[${new Date().toISOString()}] Event ${i}\n`;
  }
}

const logStream = Readable.from(generateLogs());
logStream.pipe(process.stdout);

---

三、Writable 流

建立自訂 Writable

import { Writable } from 'stream';

class BatchWriter extends Writable {
  constructor(options) {
    super({ objectMode: true, highWaterMark: 10 });
    this.batch = [];
    this.batchSize = options.batchSize || 5;
  }

  _write(chunk, encoding, callback) {
    this.batch.push(chunk);

    if (this.batch.length >= this.batchSize) {
      this.flush()
        .then(() => callback())
        .catch(callback);
    } else {
      callback();
    }
  }

  _final(callback) {
    if (this.batch.length > 0) {
      this.flush()
        .then(() => callback())
        .catch(callback);
    } else {
      callback();
    }
  }

  async flush() {
    console.log('寫入批次:', this.batch);
    await db.batchInsert(this.batch);
    this.batch = [];
  }
}

write() 返回值與背壓

const writable = fs.createWriteStream('output.txt');

for (let i = 0; i < 1e6; i++) {
  const canContinue = writable.write(`Line ${i}\n`);
  if (!canContinue) {
    // 緩衝區已滿，等待 drain 事件
    await once(writable, 'drain');
  }
}
writable.end();

---

四、Transform 流

基礎 Transform

import { Transform } from 'stream';

class JsonLineParser extends Transform {
  constructor() {
    super({ objectMode: true });
  }

  _transform(chunk, encoding, callback) {
    try {
      const data = JSON.parse(chunk.toString());
      this.push(data);
      callback();
    } catch (err) {
      callback(err);
    }
  }
}

實戰：CSV 轉 JSON 流

import { Transform } from 'stream';

class CsvToJson extends Transform {
  constructor() {
    super({ writableObjectMode: false, readableObjectMode: true });
    this.headers = null;
    this.partialLine = '';
  }

  _transform(chunk, encoding, callback) {
    const lines = (this.partialLine + chunk.toString()).split('\n');
    this.partialLine = lines.pop(); // 保留不完整的行

    for (const line of lines) {
      if (!this.headers) {
        this.headers = line.split(',');
        continue;
      }

      const values = line.split(',');
      const obj = {};
      this.headers.forEach((h, i) => obj[h.trim()] = values[i]?.trim());
      this.push(obj);
    }

    callback();
  }

  _flush(callback) {
    if (this.partialLine) {
      const values = this.partialLine.split(',');
      const obj = {};
      this.headers.forEach((h, i) => obj[h.trim()] = values[i]?.trim());
      this.push(obj);
    }
    callback();
  }
}

使用 pipeline 組合

import { pipeline } from 'stream/promises';
import { createReadStream, createWriteStream } from 'fs';
import { CsvToJson } from './csv-to-json.js';
import { Transform } from 'stream';

await pipeline(
  createReadStream('data.csv'),
  new CsvToJson(),
  new Transform({
    objectMode: true,
    transform(obj, _, cb) {
      this.push(JSON.stringify(obj) + '\n');
      cb();
    },
  }),
  createWriteStream('data.jsonl')
);

pipeline 的優勢：自動處理錯誤傳播和流清理，比 .pipe() 更安全。

---

五、背壓（Backpressure）機制

什麼是背壓？

當生產者速度 > 消費者速度時，資料在緩衝區堆積。背壓是消費者通知生產者減速的機制。

Readable (快) → Writable (慢)
  產生資料 100MB/s    只能處理 10MB/s
  
  無背壓：緩衝區爆滿 → 記憶體溢位
  有背壓：Readable 暫停 → 等待 Writable 消化

背壓的工作流程

readable.on('data', (chunk) => {
  const canContinue = writable.write(chunk);
  if (!canContinue) {
    readable.pause(); // 暫停讀取
    writable.once('drain', () => {
      readable.resume(); // 恢復讀取
    });
  }
});

highWaterMark 設定

流型別	預設值	含義
Readable	64KB (16KB objectMode)	內部緩衝區大小
Writable	16KB	寫入緩衝區大小

const readable = fs.createReadStream('big.txt', {
  highWaterMark: 1024 * 1024, // 1MB 緩衝區
});

---

六、實戰案例

案例1：大檔案 Gzip 壓縮

import { pipeline } from 'stream/promises';
import { createReadStream, createWriteStream } from 'fs';
import { createGzip } from 'zlib';

await pipeline(
  createReadStream('large.log'),
  createGzip({ level: 9 }),
  createWriteStream('large.log.gz')
);

案例2：HTTP 流式回應

import { createReadStream } from 'fs';
import { stat } from 'fs/promises';

app.get('/video/:id', async (req, res) => {
  const filePath = `/videos/${req.params.id}.mp4`;
  const { size } = await stat(filePath);
  const range = req.headers.range;

  if (range) {
    const [start, end] = range.replace(/bytes=/, '').split('-').map(Number);
    res.writeHead(206, {
      'Content-Range': `bytes ${start}-${end || size - 1}/${size}`,
      'Content-Length': (end || size - 1) - start + 1,
      'Content-Type': 'video/mp4',
    });
    createReadStream(filePath, { start, end }).pipe(res);
  } else {
    res.writeHead(200, {
      'Content-Length': size,
      'Content-Type': 'video/mp4',
    });
    createReadStream(filePath).pipe(res);
  }
});

案例3：逐行處理日誌

import { pipeline } from 'stream/promises';
import { createReadStream } from 'fs';
import { Transform } from 'stream';
import { createInterface } from 'readline';

const fileStream = createReadStream('app.log');
const rl = createInterface({ input: fileStream });

let errorCount = 0;
for await (const line of rl) {
  if (line.includes('ERROR')) {
    errorCount++;
    if (errorCount <= 10) console.log(line);
  }
}
console.log(`總錯誤數: ${errorCount}`);

案例4：Web Stream 與 Node Stream 互轉

import { Readable } from 'stream';

// Node Readable → Web ReadableStream
const nodeStream = fs.createReadStream('data.bin');
const webStream = Readable.toWeb(nodeStream);

// 在瀏覽器端使用
const reader = webStream.getReader();
while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  processChunk(value);
}

---

七、常見陷阱

陷阱	後果	解決
忘記處理 error 事件	程序靜默崩潰	用 pipeline 代替 .pipe()
.pipe() 不處理背壓	記憶體溢位	用 pipeline 或手動處理
Transform 中不呼叫 callback	流掛起	確保 _transform 中呼叫 callback
objectMode 混用	資料丟失/型別錯誤	確保上下游 objectMode 一致
不呼叫 _final	最後一批資料丟失	實作 _final 處理殘餘資料