Node.js Stream 深度解析：從背壓控制到管道模式的生產實踐

一次 OOM 排查引出的認知升級

兩年前接手一個日誌處理服務，功能很簡單——讀取 Nginx 訪問日誌，解析後寫入 ClickHouse。舊程式碼長這樣：

const fs = require('fs');

async function processLogs(filePath) {
  const content = await fs.promises.readFile(filePath, 'utf-8');
  const lines = content.split('\n');

  for (const line of lines) {
    const parsed = parseLogLine(line);
    await insertToClickHouse(parsed);
  }
}

日常 200MB 的日誌檔案跑得沒問題。直到某天維運把日誌輪轉週期從 1 天改成了 1 週，單個檔案膨脹到 1.8GB。服務剛啟動 3 秒，readFile 就把整個檔案讀進記憶體——直接 OOM Kill。

這是 Node.js Stream 最經典的使用場景。Stream 不是「流式處理」的花俏名詞，而是當資料量超過可用記憶體時，你唯一的選擇。

理解 Stream 的四種類型

類型	職責	類比	核心方法
Readable	生產資料	水龍頭	`read()`, `pipe()`
Writable	消費資料	下水道	`write()`, `end()`
Transform	邊讀邊寫邊轉換	淨水器	`_transform()`
Duplex	可讀可寫（雙向獨立）	對講機	`_read()` + `_write()`

Readable Stream

const fs = require('fs');

// 檔案再大也不會爆記憶體——每次只讀 64KB
const stream = fs.createReadStream('1.8gb-logfile.log', {
  encoding: 'utf-8',
  highWaterMark: 64 * 1024,
});

Writable Stream

const fs = require('fs');

const writable = fs.createWriteStream('output.log', {
  flags: 'a',
  highWaterMark: 16 * 1024,
});

// write() 返回 false 表示內部緩衝區滿了，需要等待 drain
const canContinue = writable.write('這是一行日誌\n');
if (!canContinue) {
  writable.once('drain', () => {
    console.log('緩衝區已排空，可以繼續寫入');
  });
}

背壓（Backpressure）——Stream 最核心的概念

背壓是指消費速度跟不上生產速度時的流量控制機制。不用背壓，資料會在記憶體中堆積直到 OOM。

// ✅ 正確：pipe() 自動處理背壓
readStream.pipe(gzipStream).pipe(writeStream);

// 或者手動實作
readStream.on('data', (chunk) => {
  const canWrite = writeStream.write(chunk);
  if (!canWrite) {
    readStream.pause();    // 暫停讀取
  }
});

writeStream.on('drain', () => {
  readStream.resume();    // 恢復讀取
});

highWaterMark——背壓的閾值

場景	預設值	建議值
檔案讀取流	64 KB (65536)	保持預設或調大到 256KB
網路 Socket	16 KB (16384)	取決於頻寬
物件模式	16 個物件	根據單物件大小調整
壓縮流（Gzip）	16 KB	調大到 64KB 減少上下文切換

pipe vs pipeline

pipe() 從 Node.js 誕生之初就存在，但有一個致命缺陷——它不會轉發錯誤：

// ❌ pipe() 的錯誤陷阱
fs.createReadStream('nonexistent.log')
  .pipe(transformStream)
  .pipe(writeStream);
// 如果中間某個流出錯，錯誤不會沿著 pipe 鏈傳播

Node.js 10 引入的 pipeline() 解決了這個問題：

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

async function processFile(inputPath, outputPath) {
  try {
    await pipeline(
      fs.createReadStream(inputPath),
      zlib.createGzip(),
      fs.createWriteStream(outputPath),
    );
    console.log('壓縮完成');
  } catch (err) {
    console.error('壓縮失敗:', err.message);
  }
}

pipeline 的三個好處

自動錯誤傳播——任一流報錯，整個管道報錯
自動銷毀——報錯後清理所有流，不會洩漏資源
回呼/Promise 雙支援——舊程式碼用回呼，新程式碼用 async/await

自訂流實戰

場景一：CSV 按行轉換流

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

class CsvTransformer extends Transform {
  constructor(delimiter = ',') {
    super({ readableObjectMode: true });
    this.delimiter = delimiter;
    this.buffer = '';
  }

  _transform(chunk, encoding, callback) {
    this.buffer += chunk.toString();
    const lines = this.buffer.split('\n');
    this.buffer = lines.pop();  // 最後一行可能不完整

    for (const line of lines) {
      if (line.trim()) {
        this.push(line.split(this.delimiter));
      }
    }

    callback();
  }

  _flush(callback) {
    if (this.buffer.trim()) {
      this.push(this.buffer.split(this.delimiter));
    }
    callback();
  }
}

場景二：分塊聚合流（Batch Transform）

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

class BatchTransform extends Transform {
  constructor(batchSize = 100) {
    super({ readableObjectMode: true, writableObjectMode: true });
    this.batchSize = batchSize;
    this.buffer = [];
  }

  _transform(item, encoding, callback) {
    this.buffer.push(item);

    if (this.buffer.length >= this.batchSize) {
      this.push([...this.buffer]);
      this.buffer = [];
    }

    callback();
  }

  _flush(callback) {
    if (this.buffer.length > 0) {
      this.push([...this.buffer]);
    }
    callback();
  }
}

效能對比：Stream vs 全量載入

實測結果（500MB 日誌檔案，MacBook Pro M1，Node.js 22）：

方式	耗時	峰值記憶體	說明
全量載入	0.8s	580 MB	檔案 + 字串開銷
readline	1.5s	45 MB	逐行解析有開銷
Transform Stream	1.2s	42 MB	塊級處理，效能與記憶體平衡

Stream 版本比全量載入略慢 0.4 秒，但記憶體只用了 7%。

生產級場景

場景一：大檔案上傳（前後端協作）

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

http.createServer(async (req, res) => {
  if (req.method !== 'PUT') {
    res.writeHead(405);
    return res.end();
  }

  const fileId = crypto.randomUUID();
  const filePath = `./uploads/${fileId}`;

  try {
    await pipeline(
      req,                                    // 直接從 HTTP 請求讀取
      fs.createWriteStream(filePath),         // 邊收邊寫磁碟
    );
    res.writeHead(200);
    res.end(JSON.stringify({ fileId }));
  } catch (err) {
    res.writeHead(500);
    res.end('Upload failed');
  }
}).listen(3000);

場景二：分頁 API 的資料匯出

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

function createOrderExportStream(userIds) {
  return new Readable({
    objectMode: true,

    async read() {
      if (userIds.length === 0) {
        this.push(null);
        return;
      }

      const userId = userIds.shift();

      try {
        let page = 1;
        let hasMore = true;

        while (hasMore) {
          const response = await fetch(
            `/api/users/${userId}/orders?page=${page}&size=100`
          );
          const { data, total } = await response.json();

          for (const order of data) {
            if (!this.push(order)) {
              userIds.unshift(userId);
              return;
            }
          }

          hasMore = (page * 100) < total;
          page++;
        }
      } catch (err) {
        this.destroy(err);
      }
    },
  });
}

常見陷阱

陷阱 1：忘記處理 error 事件

// ❌ 錯誤：未處理的 error 會 crash 程序
const stream = fs.createReadStream('maybe-not-exist.txt');
stream.pipe(process.stdout);

// ✅ 正確
stream.on('error', (err) => {
  console.error('讀取失敗:', err.message);
});

陷阱 2：在 Transform 的 flush 中忘記 callback

// ❌ 流永遠不會結束
_flush(callback) {
  this.push('剩餘資料');
  // 忘記呼叫 callback()！
}

// ✅ 正確
_flush(callback) {
  this.push('剩餘資料');
  callback();
}