Node.jsのストリーム — 4つのタイプとバックプレッシャー機構の詳細な解説

チャーリーには、サーバー上の2GBのアクセスログを分析するという緊急の任務が課せられました。彼がfs.readFileを使ってログを読み込んだ途端、メモリがすぐに不足し、OOMクラッシュが発生してしまいました。チームリーダーは、「ファイルがどんなに大きかろうと、一度に全部処理しようとしないで、Streamを使って少しずつ処理しなさい！」とアドバイスした。チャーリーはcreateReadStreamに切り替えてログを1行ずつ処理したところ、メモリ使用量は50MBで安定した。それ以来、彼は出会う人すべてにこう言うようになった。「Streamは、Node.jsでビッグデータを処理するためのスイスアーミーナイフのような存在だ。」

1. ストリームの基本概念

Streamは、Node.jsにおけるストリーミングデータを処理するための抽象インターフェースです。ストリーミングデータとは、データが一度にメモリに読み込まれるのではなく、水の流れのようにチャンク単位で送信されるものです。

• 重要な概念：データをチャンク単位で処理し、すべてのデータが揃うのを待たずに操作を開始できるようにすること
• 継承関係：すべてのストリームは EventEmitter のインスタンスであり、イベントを使用して状態の変化を通知します。
• 利用例：大容量ファイルの読み書き、ネットワーク通信、データの圧縮および変換
• 組み込みモジュール：fs、zlib、crypto、http などは、いずれもストリームインターフェースを提供しています。
• グローバルオブジェクト：stream モジュールは、基底クラス Readable、Writable、Duplex、および Transform を提供します。

(1) 4種類のフロー

(2) 2種類の流れ

Readableストリームには、2つの動作モードがあります：

2. ストリームイベントの詳細な解説

すべてのストリームは EventEmitter をベースとしており、イベント駆動型の仕組みを用いてデータや状態を伝達しています。

(1) 一般的なイベント

▶ 例：Readableストリームのイベントをリッスンする

const fs = require('fs');
const rs = fs.createReadStream('./access.log', { highWaterMark: 64 * 1024 });

rs.on('data', (chunk) => {
  console.log(`Received ${chunk.length} Byte`);
});

rs.on('end', () => {
  console.log('Finished reading');
});

rs.on('error', (err) => {
  console.error('Output error:', err.message);
});

▶ 例：書き込み可能ストリームのイベントを監視する

const ws = fs.createWriteStream('./output.txt');

ws.on('finish', () => {
  console.log('All data has been written');
});

ws.on('error', (err) => {
  console.error('Write error:', err.message);
});

ws.write('Hello Stream');
ws.end();

• data フローモードではイベントが自動的にトリガーされます
• end と finish の違い：end は読み取りが完了したことを示し、finish は書き込みが完了したことを示します。
• error イベントをリッスンする必要があります。そうしないと、キャッチされなかった例外によってプロセスがクラッシュしてしまいます。
• close すべてのストリームがこのイベントをトリガーするわけではありません。これは、基盤となる実装によって異なります。

3. pipe() との連鎖

pipe() は Stream の最も強力なメソッドであり、読み取り可能なストリームの出力を書き込み可能なストリームの入力に自動的に接続し、バックプレッシャーを自動的に処理します。

(1) 「pipe」の基本的な使い方

readable.pipe(writable);

pipe() ターゲットストリームを返すため、チェーン処理が可能です：

readable.pipe(transform1).pipe(transform2).pipe(writable);

▶ 例：ファイルのコピー

const fs = require('fs');
fs.createReadStream('./source.txt')
  .pipe(fs.createWriteStream('./dest.txt'));

▶ 例：HTTPレスポンスのストリーミング

const http = require('http');
const fs = require('fs');

http.createServer((req, res) => {
  res.writeHead(200, { 'Content-Type': 'text/plain' });
  fs.createReadStream('./large.txt').pipe(res);
}).listen(3000);

(2) パイプデータのフロー図

flowchart LR
    A[Readable<br/>Data Source] -->|chunk| B[Transform<br/>Data Transformation]
    B -->|chunk| C[Writable<br/>Data Endpoint]
    C -.->|backpressure| A
    B -.->|backpressure| A

    style A fill:#4CAF50,color:#fff
    style B fill:#FF9800,color:#fff
    style C fill:#2196F3,color:#fff

flowchart LR
    A[Readable<br/>Data Source] -->|chunk| B[Transform<br/>Data Transformation]
    B -->|chunk| C[Writable<br/>Data Endpoint]
    C -.->|backpressure| A
    B -.->|backpressure| A

    style A fill:#4CAF50,color:#fff
    style B fill:#FF9800,color:#fff
    style C fill:#2196F3,color:#fff

• pipe() データ転送レートを自動的に管理します。書き込み側がビジー状態になると、読み取り側は一時停止します。
• pipe() エラーは処理されない。各ストリームごとにイベントを個別に監視する必要がある error
• stream.pipeline() を使用すると、クリーンアップとエラーの伝播が自動的に処理されるため、pipe() よりも安全です。

4. fs ストリームとファイル操作

fs モジュールは、ファイルシステムに関連する読み取りおよび書き込みストリームを提供します。

(1) createReadStream / createWriteStream

▶ 例：指定した範囲内のファイルセグメントを読み込む

const fs = require('fs');
const rs = fs.createReadStream('./big.bin', {
  start: 100,
  end: 199,
  highWaterMark: 32
});

rs.on('data', (chunk) => {
  console.log(chunk.length);
});

(2) readFile と createReadStream の比較

▶ 例：大容量ファイルをブロック単位で処理する

const fs = require('fs');
let totalBytes = 0;

const rs = fs.createReadStream('./2gb.log');
rs.on('data', (chunk) => {
  totalBytes += chunk.length;
});
rs.on('end', () => {
  console.log(`Total ${totalBytes} Byte`);
});

• 大きなファイルは、OOMエラーを回避するためにストリームを使用して処理する必要があります
• highWaterMark メモリが小さければ小さいほど、電力効率は良くなりますが、システムコールの頻度は高くなります。
• start/end オプションを指定すると、ファイルをチャンク単位で読み込むことができます

5. 背圧メカニズム

バックプレッシャーはフロー制御の中核をなすものです。書き込み側が読み出し側のプッシュレートに追いつけない場合、読み出し側を一時停止させ、メモリのバックログを防ぐために、書き込み側に「バックプレッシャー」がかけられます。

(1) pipe() はバックプレッシャーを自動的に処理する

pipe() を使用する場合、背圧は Node.js によって内部で自動的に管理されるため、手動での操作は不要です。

(2) 手動背圧制御

pipe() を使用しない場合は、write() の戻り値を手動で評価する必要があります：

const fs = require('fs');
const rs = fs.createReadStream('./source.txt');
const ws = fs.createWriteStream('./dest.txt');

rs.on('data', (chunk) => {
  const canContinue = ws.write(chunk);
  if (!canContinue) {
    rs.pause();
    ws.once('drain', () => {
      rs.resume();
    });
  }
});

rs.on('end', () => {
  ws.end();
});

▶ 例：背圧の影響の観察

const rs = fs.createReadStream('./big.log', { highWaterMark: 1024 });
const ws = fs.createWriteStream('./out.log', { highWaterMark: 512 });

let paused = 0;
rs.on('data', (chunk) => {
  const ok = ws.write(chunk);
  if (!ok) {
    paused++;
    rs.pause();
    ws.once('drain', () => rs.resume());
  }
});
rs.on('end', () => {
  console.log(`Backpressure triggered ${paused} times`);
  ws.end();
});

• write() Return false は、内部バッファが満杯であり、読み取りを一時停止すべきであることを示しています
• drain このイベントは、バッファがフラッシュされ、書き込みを再開できることを示しています。
• バックプレッシャーを無視すると、メモリ使用量が増え続け、最終的にはOOM（メモリ不足）を引き起こします
• pipeline() は pipe() よりも強く推奨されます。エラーの伝播やリソースのクリーンアップを自動的に処理してくれるからです。

6. ストリームの変換

TransformストリームはDuplexのサブクラスであり、その出力は入力の変換結果となります。一般的に、データの圧縮、暗号化、および形式変換に使用されます。

(1) 「トランスフォーム」と「デュプレックス」の違い

▶ 例：カスタム変換ストリーム（大文字・小文字の変換）

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

const upper = new Transform({
  transform(chunk, encoding, callback) {
    callback(null, chunk.toString().toUpperCase());
  }
});

process.stdin.pipe(upper).pipe(process.stdout);

▶ 例：zlibで圧縮されたファイル

const fs = require('fs');
const zlib = require('zlib');

fs.createReadStream('./access.log')
  .pipe(zlib.createGzip())
  .pipe(fs.createWriteStream('./access.log.gz'));

▶ 例：zlibファイルの解凍

const fs = require('fs');
const zlib = require('zlib');

fs.createReadStream('./access.log.gz')
  .pipe(zlib.createGunzip())
  .pipe(fs.createWriteStream('./access_restored.log'));

• _transform(chunk, encoding, callback) において、callback(null, data) は変換結果を送信する
• callback(err) 許容範囲内のエラー
• zlib.createGzip() / createGunzip() は、最も一般的に使用される組み込みのトランスフォームストリームです
• Transformストリームをカスタマイズする際、残りのデータに対して_flush(callback)処理を実装することも可能です。

7. ストリームの停止と再開

「Readable」ストリームはデフォルトで一時停止モードになっており、pause() および resume() を使用して切り替えることができます。

▶ 例：一時停止と再開の操作

const fs = require('fs');
const rs = fs.createReadStream('./big.log');
let count = 0;

rs.on('data', (chunk) => {
  count++;
  if (count % 10 === 0) {
    rs.pause();
    console.log(`Processed ${count} chunks, pause 1 second`);
    setTimeout(() => rs.resume(), 1000);
  }
});

▶ 例：ファイルを行単位で読み込む（readline）

const fs = require('fs');
const readline = require('readline');

const rl = readline.createInterface({
  input: fs.createReadStream('./access.log'),
  crlfDelay: Infinity
});

rl.on('line', (line) => {
  if (line.includes('ERROR')) {
    console.log(line);
  }
});

rl.on('close', () => {
  console.log('The file has been read.');
});

• readline.createInterface Readableストリームをラップし、行単位で読み込むためのインターフェースとして機能する
• crlfDelay: Infinity \r\n の改行を正しく処理する
• pause() の後、resume() が呼び出されるまでは、data イベントはトリガーされなくなります。
• pipe() はストリームをライブモードに切り替えます

8. 包括的な例：ログ処理パイプライン

完全なデータ処理パイプラインを構築する：大規模なログファイルを読み込む → 行ごとに解析する → エラー行を除外する → 出力を圧縮する。

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

const filterError = new Transform({
  transform(chunk, encoding, callback) {
    const lines = chunk.toString().split('\n');
    const errors = lines.filter(l => l.includes('ERROR')).join('\n');
    callback(null, errors ? errors + '\n' : '');
  }
});

const src = fs.createReadStream('./app.log');
const dest = fs.createWriteStream('./errors.log.gz');

src
  .pipe(filterError)
  .pipe(zlib.createGzip())
  .pipe(dest);

dest.on('finish', () => {
  console.log('The error log has been written in compressed form.');
});

src.on('error', (err) => {
  console.error('Read failed:', err.message);
});

• パイプラインの各段階はストリームであり、データはチャンク単位で流れるため、メモリ使用量が極めて少なくなります。
• pipe() 隣接するストリーム間の背圧を自動的に処理する
• エラーは各ストリームごとに個別に監視するか、pipeline() を使用して統一的に処理する必要があります。
• ストリーム filterError を変換し、ERROR を含む行のみを残す

❓ よくある質問

• Streamはどのような場合に使用すべきか？ データ量が利用可能なメモリ容量を超える場合、データを読み込みながら処理する必要がある場合、またはリアルタイムのデータソースを扱う場合にはStreamを使用してください。小さなファイルの場合は、readFileを使用したほうが簡単です。
• pipe() は背圧を自動的に処理しますか？ はい、pipe() は書き込み側 write() からの戻り値を内部でチェックし、流量を管理するために pause()/resume() を自動的に呼び出します。
• ファイルを1行ずつ読み取るにはどうすればよいですか？ readline.createInterface({ input: createReadStream(path) }) を使用し、line イベントを監視することで、データを1行ずつ取得できます。
• ****TransformとDuplexの違いは何ですか？ Duplexでは、読み取りと書き込みは独立しており、互いに関連がありません。一方、Transformでは、入力を変換することで出力が生成されるため、_transform()メソッドの実装のみが必要です。
• ストリームのエラーはどのように処理すべきか？ 各ストリームで error イベントを監視するか、stream.pipeline() を使用してエラーを自動的に伝播させ、リソースをクリーンアップすることで、未処理の例外によるプロセスのクラッシュを防ぐことができます。
• highWaterMark の適切な値はどれくらいですか？ デフォルトの 64 KB はほとんどのシナリオに適しています。大容量のバイナリデータを処理する場合は 256 KB に増やし、メモリが限られている場合は 16 KB に減らすことができます。
• pipe() と pipeline()、どちらを選ぶべきでしょうか？ エラーを自動的に伝播し、リソースをクリーンアップする stream.pipeline() の使用をお勧めします。一方、pipe() はエラーを処理せず、ストリームを自動的に閉じません。

📖 まとめ

• ストリームの基本概念と使い方
• ストリームイベントの中核となる概念と使用方法に関する詳細な解説
• pipe() によるチェーン可能な接続の基本概念と使い方
• FSストリームおよびファイル操作の基本概念と使用方法
• 背圧メカニズムの基本概念と用途
• トランスフォーム・ストリームの基本概念と使用方法
• ストリームの停止と再開に関する主要な概念と使い方
• 包括的な例：ログ処理パイプラインの基本概念と使用方法

📝 練習問題

1. このレッスンにあるすべてのコード例を完成させ、それぞれが正しく動作することを確認してください。
2. 包括的な例を修正し、独自の拡張機能を追加する
3. 公式ドキュメントを確認し、このレッスンで扱われていないAPIを1～2つ見つけ、それらのテストコードを作成してください。
4. 振り返り：このレッスンで学んだことを、実際のプロジェクトにどのように活かしますか？
5. このレッスンで学んだことと、これまでのレッスンの内容を組み合わせて、小さなプロジェクトを作成してみてください。

Web-Tutorial 技術チーム

複数の開発者によって共同維持されているプログラミングチュートリアルプラットフォーム。各チュートリアルは専門分野の開発者が執筆・レビューしています。正確で信頼性の高いコンテンツを目指しています — 問題を見つけた場合はお知らせください。

私たちについて