Node.js マルチプロセス

Node.js はシングルスレッドモードで動作することで知られていますが、イベント駆動方式で並行処理を行っています。これにより、マルチコア CPU のシステム上で複数の子プロセスを作成し、パフォーマンスを向上させることができます。

各子プロセスには常に3つのストリームオブジェクト（child.stdin、child.stdout、child.stderr）があります。これらは親プロセスの stdio ストリームを共有するか、あるいは独立したパイプされたストリームオブジェクトにすることもできます。

Node は子プロセスを作成するために child_process モジュールと cluster モジュールを提供しています。

child_processモジュール

  • exec- child_process.exec は子プロセスでコマンドを実行し、子プロセスの出力をキャッシュし、その出力をコールバック関数の引数として返します。

  • execFile- 実行可能ファイルを直接実行するため、やや安全です。

  • spawn- child_process.spawn は指定されたコマンドライン引数を使って新規プロセスを作成します。

  • fork- child_process.fork は spawn() の特殊な形式で、子プロセス内で実行するモジュール用です。例えば fork('./son.js') は spawn('node', ['./son.js']) と同等です。spawn メソッドと異なり、fork は親プロセスと子プロセスの間に通信パイプを確立し、プロセス間通信に使用します。

clusterモジュール

  • 1つのポート上で複数のワーカープロセス共有リスニングを実現し、1コア1プロセスの HTTP サービス。
  • Web サービスを複数コアに水平展開してフル活用するのに適しています。
  • 注記: 公式は新規プロジェクトでは慎重に評価することを推奨しています（例: リバースプロキシ + マルチプロセスを直接使う、またはworker_threadsCPU 集約的な処理を行う）、しかしcluster依然として広く使用できます。

exec() メソッド

child_process.exec は子プロセスでコマンドを実行し、子プロセスの出力をキャッシュし、その出力をコールバック関数の引数として返します。

構文は次のとおりです：

child_process.exec(command[, options], callback)

パラメータ

パラメータの説明は次のとおりです：

command：文字列。実行するコマンド。引数はスペースで区切ります。

options: オブジェクト。次のものを指定できます：

  • cwd: 文字列。子プロセスの現在の作業ディレクトリ。
  • env: オブジェクト。環境変数のキーと値のペア。
  • encoding: 文字列。文字エンコーディング（デフォルト: 'utf8'）
  • shell: 文字列。コマンドを実行するシェル（デフォルト: UNIX では /bin/sh/bin/sh、Windows では cmd.execmd.exe、シェルは -c を認識できる必要があります-cスイッチ（UNIX では）、または /s /c/s /cWindows では。Windows では、コマンドライン解析は cmd.exe と互換性がある必要があります。cmd.exe）
  • timeout: 数値。タイムアウト時間（デフォルト: 0）
  • maxBuffer: 数値。stdout または stderr 内で許容される最大バッファ（バイナリ）。超えた場合、子プロセスは強制終了されます（デフォルト: 200*1024）
  • killSignal: 文字列。終了シグナル（デフォルト: 'SIGTERM'）
  • uid: 数値。ユーザープロセスの ID を設定します。
  • gid: 数値。プロセスグループの ID を設定します。

callback ：コールバック関数。3つのパラメータ error、stdout、stderr を含みます。

exec() メソッドは最大バッファを保持し、プロセスの終了を待ってから、バッファの内容を一度に返します。

例

2つの js ファイル support.js と master.js を作成しましょう。

support.js ファイルのコード：

console.log("プロセス" + process.argv[2] + "実行します。" );

master.js ファイルのコード：

const fs = require('fs'); const child_process = require('child_process'); for(var i=0; i<3; i++) { var workerProcess = child_process.exec('node support.js '+i, function (error, stdout, stderr) { if (error) { console.log(error.stack); console.log('Error code: '+error.code); console.log('Signal received: '+error.signal); } console.log('stdout: ' + stdout); console.log('stderr: ' + stderr); }); workerProcess.on('exit', function (code) { console.log('子プロセスが終了しました。終了コード'+code); }); }

上記のコードを実行すると、出力結果は次のとおりです：

$ node master.js 
子进程已退出，退出码 0
stdout: 进程 1 执行。

stderr: 
子进程已退出，退出码 0
stdout: 进程 0 执行。

stderr: 
子进程已退出，退出码 0
stdout: 进程 2 执行。

stderr:

解説：

exec起動し、3つの子プロセス、それぞれ実行します：

  • node support.js 0
  • node support.js 1
  • node support.js 2

各子プロセスは実行後すぐに出力します：

进程 0 执行。
进程 1 执行。
进程 2 执行。

この部分は子プロセスの標準出力（stdout）、バッファリングされ、子プロセスの終了時にまとめてexecのコールバックに返されます。

子プロセスが終了すると、最初にexitイベントが発生し、次にexecのコールバックが実行されます（注: Node.js ドキュメントではexitイベントはプロセス終了時に発生すると説明されていますが、execのコールバックはstdioストリームが閉じた後にトリガーされるため、通常 exit はコールバックより先に出力されます）。

出力順序は決まっていません。3つの子プロセスが並行して実行されるため、どれが先に終了するかは OS のスケジューリングに依存します。

spawn() メソッド

child_process.spawn は指定されたコマンドライン引数を使用して新規プロセスを作成します。構文は次のとおりです：

child_process.spawn(command[, args][, options])

パラメータ

パラメータの説明は次のとおりです：

command：実行するコマンド

args：Array 文字列の引数配列

options Object

  • cwd String 子プロセスの現在の作業ディレクトリ
  • env Object 環境変数のキーと値のペア
  • stdio Array|String 子プロセスの stdio 設定
  • detached Boolean この子プロセスはプロセスグループのリーダーになります
  • uid Number ユーザープロセスの ID を設定します
  • gid Number プロセスグループの ID を設定します

spawn() メソッドはストリーム（stdout & stderr）を返し、プロセスが大量のデータを返す場合に使用します。プロセスが実行を開始するとすぐに spawn() は応答の受信を開始します。

例

2つの js ファイル support.js と master.js を作成しましょう。

support.js ファイルのコード：

console.log("プロセス" + process.argv[2] + "実行します。" );

master.js ファイルのコード：

const fs = require('fs'); const child_process = require('child_process'); for(var i=0; i<3; i++) { var workerProcess = child_process.spawn('node', ['support.js', i]); workerProcess.stdout.on('data', function (data) { console.log('stdout: ' + data); }); workerProcess.stderr.on('data', function (data) { console.log('stderr: ' + data); }); workerProcess.on('close', function (code) { console.log('子プロセスが終了しました。終了コード'+code); }); }

上記のコードを実行すると、出力結果は次のとおりです：

$ node master.js stdout: 进程 0 执行。

子进程已退出，退出码 0
stdout: 进程 1 执行。

子进程已退出，退出码 0
stdout: 进程 2 执行。

子进程已退出，退出码 0

fork メソッド

child_process.fork は spawn() メソッドの特殊な形式で、プロセスを作成するために使用します。構文は次のとおりです：

child_process.fork(modulePath[, args][, options])

パラメータ

パラメータの説明は次のとおりです：

modulePath： String、子プロセス内で実行するモジュール

args： Array 文字列の引数配列

options：Object

  • cwd String 子プロセスの現在の作業ディレクトリ
  • env Object 環境変数のキーと値のペア
  • execPath String 子プロセスを作成する実行可能ファイル
  • execArgv Array 子プロセスの実行可能ファイルの文字列引数配列（デフォルト: process.execArgv）
  • silent Boolean が true の場合true、子プロセスのstdin，stdoutとstderrは親プロセスにパイプされ、そうでなければそれらは親プロセスから継承されます。（デフォルト: false）false）
  • uid Number ユーザープロセスの ID を設定します
  • gid Number プロセスグループの ID を設定します

返されるオブジェクトは、ChildProcessインスタンスのすべてのメソッドに加えて、組み込みの通信チャネルも備えています。

例

js ファイル support.js と master.js を 2 つ作成しましょう。

support.js ファイルのコード：

console.log("プロセス" + process.argv[2] + "実行。" );

master.js ファイルのコード：

const fs = require('fs'); const child_process = require('child_process'); for(var i=0; i<3; i++) { var worker_process = child_process.fork("support.js", [i]); worker_process.on('close', function (code) { console.log('子プロセスが終了しました。終了コード' + code); }); }

上記のコードを実行すると、出力結果は次のようになります：

$ node master.js 
进程 0 执行。
子进程已退出，退出码 0
进程 1 执行。
子进程已退出，退出码 0
进程 2 执行。
子进程已退出，退出码 0

cluster: 1ポート・マルチプロセスのHTTPサービス

最小限のマルチコアHTTPサービス

例

// server-cluster.js
import cluster from 'node:cluster';
import os from 'node:os';
import http from 'node:http';
import { pbkdf2Sync } from 'node:crypto';

if (cluster.isPrimary) {
  const cpuCount = Math.max(1, os.cpus().length);
  console.log(`主プロセス ${process.pid}、起動 ${cpuCount}個のワーカープロセス`);
  for (let i = 0; i < cpuCount; i++) cluster.fork();

  cluster.on('exit', (worker, code) => {
    console.warn(`ワーカープロセス ${worker.process.pid}が終了（code=${code}）、再起動中…`);
    cluster.fork();
  });
} else {
  const server = http.createServer((req, res) => {
    // CPU集中処理をシミュレート
    pbkdf2Sync('password', 'salt', 100_000, 64, 'sha512');
    res.writeHead(200, {'content-type':'text/plain; charset=utf-8'});
    res.end(`Handled by worker ${process.pid}\n`);
  });

  server.listen(3000, () => {
    console.log(`ワーカープロセス ${process.pid}がリッスン3000`);
  });
}

実行：node server-cluster.js、複数回リクエストするとhttp://localhost:3000異なるPID応答が表示されます。

注意：本番の複数インスタンスでは通常、リバースプロキシ/ロードバランサー（Nginx、Envoy、K8s Service）を前に置くか、cluster で1つのポートに集約します。

WebSocket/粘着セッションには"sticky session"（同じクライアントが同じ worker に割り当てられること）が必要です。四層ロードバランサーの送信元アドレスハッシュを使うか、アプリケーション層で自分で粘着を実装できます（例：req.socket.remoteAddress に基づいて振り分ける）。

グレースフルリスタート（ゼロダウンタイムの考え方）

考え方：主プロセスがリロードシグナルを受信 → まず新しい worker を fork し、その listening を待つ → その後古い worker を disconnect し、リクエストの処理が完了して終了するのを待つ。

例

// 要約例（cluster.isPrimary 分岐内）
process.on('SIGUSR2', async () => {
  console.log('SIGUSR2 を受信、グレースフルリスタートを開始');
  const workers = Object.values(cluster.workers ?? {});
  // 1) 新しいものを起動
  const fresh = cluster.fork();
  await new Promise(r => fresh.once('listening', r));
  // 2) 古いものを一つずつ停止
  for (const w of workers) {
    w?.disconnect();
    // タイムアウトしても終了しない場合は強制終了
    setTimeout(() => w?.process.kill('SIGKILL'), 5000);
  }
});

Windows には SIGUSR2 がないため、HTTP / RPC 管理インターフェースでリロードをトリガーできます。

独自の「プロセスプール」: 並行処理を制御し、子プロセスを再利用する

大量の独立した CPU タスク（バッチ圧縮、暗号化、クローラー解析など）があり、頻繁に fork/spawn するとコストが高い場合、このとき Process Pool で複数の子プロセスを再利用し、"スレッドプール"のように流量制御する必要があります。

ファイル：pool/worker.js

例

// pool/worker.js
process.on('message', async (msg) => {
  if (msg.type === 'task') {
    const { id, payload } = msg;
    // 重いタスクをシミュレート（フィボナッチ）
    const fib = (n) => (n <= 1 ? n : fib(n-1) + fib(n-2));
    const result = fib(payload.n);
    process.send({ type: 'done', id, result });
  }
});

ファイル：pool/index.js

例

// pool/index.js
import { fork } from 'node:child_process';
import os from 'node:os';

export class ProcessPool {
  constructor({ file, size = Math.max(1, os.cpus().length - 1) } = {}) {
    this.file = file;
    this.size = size;
    this.idle = [];
    this.busy = new Map();  // worker -> taskId
    this.queue = [];
    for (let i = 0; i < size; i++) this._spawn();
  }

  _spawn() {
    const w = fork(this.file);
    w.on('message', (m) => {
      if (m?.type === 'done') {
        const cb = this.callbacks.get(m.id);
        if (cb) cb.resolve(m.result);
        this.callbacks.delete(m.id);
        this._markIdle(w);
        this._drain();
      }
    });
    w.on('exit', () => {
      // プールを自動的に補充
      this.busy.delete(w);
      const idx = this.idle.indexOf(w);
      if (idx >= 0) this.idle.splice(idx, 1);
      this._spawn();
    });
    if (!this.callbacks) this.callbacks = new Map();
    this.idle.push(w);
  }

  _markIdle(w) {
    this.busy.delete(w);
    if (!this.idle.includes(w)) this.idle.push(w);
  }

  _acquire() {
    return this.idle.length ? this.idle.shift() : null;
  }

  _drain() {
    while (this.queue.length && this.idle.length) {
      const { id, payload, resolve, reject } = this.queue.shift();
      const w = this._acquire();
      this.busy.set(w, id);
      this.callbacks.set(id, { resolve, reject });
      w.send({ type: 'task', id, payload });
    }
  }

  runTask(payload) {
    const id = Math.random().toString(36).slice(2);
    return new Promise((resolve, reject) => {
      this.queue.push({ id, payload, resolve, reject });
      this._drain();
    });
  }

  close() {
    for (const w of this.idle) w.kill();
    for (const w of this.busy.keys()) w.kill();
  }
}

ファイル：pool/demo.js

例

// pool/demo.js
import { ProcessPool } from './index.js';

const pool = new ProcessPool({ file: './pool/worker.js', size: 4 });

const tasks = Array.from({ length: 10 }, (_, i) => pool.runTask({ n: 35 + (i % 3) }));
const t0 = Date.now();
const results = await Promise.all(tasks);
console.log('結果:', results);
console.log('所要時間(ms):', Date.now() - t0);

await new Promise(r => setTimeout(r, 100)); // メッセージが処理されるのを待つ
pool.close();

説明：

  • 「タスク」をメッセージとしてカプセル化し、プロセスプールが worker の割り当てと再利用を担当します。
  • 実際のビジネスでは、解析/トランスコード/圧縮をカプセル化してworker.jsに入れることができます。
  • 注意タスクのタイムアウト、リトライ、冪等性と背圧（キュー長の上限）。

実践例: HTTP インターフェースが重いタスクをプロセスプールに投げる

例

// app.js
import http from 'node:http';
import { ProcessPool } from './pool/index.js';

const pool = new ProcessPool({ file: './pool/worker.js', size: 4 });
const server = http.createServer(async (req, res) => {
  if (req.url?.startsWith('/fib?')) {
    const url = new URL(req.url, 'http://localhost');
    const n = Number(url.searchParams.get('n') || 35);
    try {
      const result = await pool.runTask({ n });
      res.writeHead(200, {'content-type': 'application/json'});
      res.end(JSON.stringify({ pid: process.pid, n, result }));
    } catch (e) {
      res.writeHead(500); res.end('error');
    }
  } else {
    res.writeHead(404).end('Not Found');
  }
});

server.listen(3000, () => console.log('http://localhost:3000'));

アクセス：/fib?n=38など。主プロセスはブロックされず、タスクは子プロセス内で計算されます。

その他の拡張