Node.js Stream 徹底解説：バックプレッシャーからパイプラインパターンまでの本番実践

OOM インシデントから得た教訓

2 年前、ログ処理サービスを引き継ぎました。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.8GB に膨れ上がりました。起動から 3 秒後、readFile がファイル全体をメモリに読み込み——即座に OOM Kill。

これが Node.js Stream の最も典型的な使用場面です。Stream は単なる「ストリーミング処理」のおしゃれな用語ではなく、データが利用可能なメモリを超える場合の唯一の選択肢です。

Stream の 4 つの型を理解する

型	役割	例え	主要メソッド
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 まで
ネットワークソケット	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 が自動的にすべての中間ストリームを破棄
  }
}

pipeline の 3 つの利点

自動エラー伝播——どのストリームのエラーもパイプライン全体のエラーに
自動破棄——エラー時に全ストリームをクリーンアップ、リソースリークなし
コールバック/Promise の両対応——レガシーコードはコールバック、新コードは async/await

カスタムストリームの実装

パターン 1：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();
  }
}

パターン 2：バッチ集約変換

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%。ファイルがさらに大きくなると、全ロードはクラッシュし、Stream は動作し続けます。

本番パターン

パターン A：大規模ファイルアップロード

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('アップロード失敗');
  }
}).listen(3000);

パターン B：ページネーション 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 イベントハンドラの忘れ

// ❌ 未処理のエラーはプロセスをクラッシュさせる
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();
}

落とし穴 3：オブジェクトモードとバッファモードの混在

// ✅ 正しい
const t = new Transform({
  readableObjectMode: true,
  writableObjectMode: true,
  transform(chunk, encoding, callback) {
    // chunk は JS オブジェクト
    callback();
  },
});