Node.js v21.7.2 ドキュメント

仕組み#

ワーカープロセスは、child_process.fork() メソッドを使用して生成されるため、IPC を介して親と通信し、サーバーハンドルをやり取りできます。

cluster モジュールは、着信接続を分散する 2 つの方法をサポートしています。

1 つ目 (および Windows を除くすべてのプラットフォームでのデフォルト) はラウンドロビンアプローチです。プライマリプロセスがポートをリッスンし、新しい接続を受け入れ、ワーカープロセスに過負荷をかけないようにするための組み込みのスマート機能を使用して、ラウンドロビン方式でワーカー全体にそれらを分散します。

2 つ目のアプローチは、プライマリプロセスが listen ソケットを作成し、それを対象のワーカーに送信することです。 次に、ワーカーは着信接続を直接受け入れます。

理論的には、2 つ目のアプローチの方が最高のパフォーマンスを発揮するはずです。 しかし実際には、オペレーティングシステムのスケジューラ変動により、分散が非常に不均衡になる傾向があります。 合計 8 つのプロセスのうち、すべての接続の 70% 以上がわずか 2 つのプロセスで終了するという負荷が観測されています。

server.listen() はほとんどの作業をプライマリプロセスに引き渡すため、通常の Node.js プロセスとクラスタワーカーの間で動作が異なる 3 つのケースがあります。

1. server.listen({fd: 7}) メッセージがプライマリに渡されるため、親のファイル記述子 7 がリッスンされ、ワーカーがファイル記述子番号 7 が参照するものをリッスンするのではなく、ハンドルがワーカーに渡されます。
2. server.listen(handle) 明示的にハンドルをリッスンすると、ワーカーはプライマリプロセスと通信するのではなく、指定されたハンドルを使用します。
3. server.listen(0) 通常、これによりサーバーはランダムなポートでリッスンします。ただし、クラスタでは、各ワーカーは listen(0) を実行するたびに同じ「ランダム」ポートを受信します。本質的に、ポートは最初はランダムですが、その後は予測可能です。一意のポートをリッスンするには、クラスタワーカー ID に基づいてポート番号を生成します。

Node.js はルーティングロジックを提供しません。したがって、セッションやログインなどのためにメモリ内のデータオブジェクトに過度に依存しないようにアプリケーションを設計することが重要です。

ワーカーはすべて個別のプロセスであるため、他のワーカーに影響を与えることなく、プログラムのニーズに応じて停止または再起動できます。 少なくとも一部のワーカーが生きている限り、サーバーは接続を受け入れ続けます。 生きているワーカーがない場合、既存の接続は切断され、新しい接続は拒否されます。 ただし、Node.js はワーカーの数を自動的に管理しません。 ワーカープールを独自のニーズに基づいて管理するのはアプリケーションの責任です。

node:cluster モジュールの主な使用例はネットワーキングですが、ワーカープロセスを必要とする他のユースケースにも使用できます。

クラス: Worker#

• 継承: <EventEmitter>

Worker オブジェクトには、ワーカーに関するすべての公開情報とメソッドが含まれています。 プライマリでは、cluster.workers を使用して取得できます。 ワーカーでは、cluster.worker を使用して取得できます。

イベント: 'disconnect'#

cluster.on('disconnect') イベントに似ていますが、このワーカーに固有です。

cluster.fork().on('disconnect', () => {
  // Worker has disconnected
}); copy

イベント: 'error'#

このイベントは、child_process.fork() によって提供されるイベントと同じです。

ワーカー内では、process.on('error') も使用できます。

イベント: 'exit'#

• code <number> 通常に終了した場合の終了コード。
• signal <string> プロセスを強制終了させたシグナル名 (例: 'SIGHUP')。

cluster.on('exit') イベントに似ていますが、このワーカーに固有です。

import cluster from 'node:cluster';

if (cluster.isPrimary) {
  const worker = cluster.fork();
  worker.on('exit', (code, signal) => {
    if (signal) {
      console.log(`worker was killed by signal: ${signal}`);
    } else if (code !== 0) {
      console.log(`worker exited with error code: ${code}`);
    } else {
      console.log('worker success!');
    }
  });
}const cluster = require('node:cluster');

if (cluster.isPrimary) {
  const worker = cluster.fork();
  worker.on('exit', (code, signal) => {
    if (signal) {
      console.log(`worker was killed by signal: ${signal}`);
    } else if (code !== 0) {
      console.log(`worker exited with error code: ${code}`);
    } else {
      console.log('worker success!');
    }
  });
}copy

イベント: 'listening'#

• address <Object>

cluster.on('listening') イベントに似ていますが、このワーカーに固有です。

cluster.fork().on('listening', (address) => {
  // Worker is listening
});cluster.fork().on('listening', (address) => {
  // Worker is listening
});copy

ワーカー内では発行されません。

イベント: 'message'#

• message <Object>
• handle <undefined> | <Object>

cluster の 'message' イベントに似ていますが、このワーカーに固有です。

ワーカー内では、process.on('message') も使用できます。

process event: 'message' を参照してください。

以下はメッセージシステムを使用した例です。これは、ワーカーが受信した HTTP リクエストの数をプライマリプロセスでカウントします

import cluster from 'node:cluster';
import http from 'node:http';
import { availableParallelism } from 'node:os';
import process from 'node:process';

if (cluster.isPrimary) {

  // Keep track of http requests
  let numReqs = 0;
  setInterval(() => {
    console.log(`numReqs = ${numReqs}`);
  }, 1000);

  // Count requests
  function messageHandler(msg) {
    if (msg.cmd && msg.cmd === 'notifyRequest') {
      numReqs += 1;
    }
  }

  // Start workers and listen for messages containing notifyRequest
  const numCPUs = availableParallelism();
  for (let i = 0; i < numCPUs; i++) {
    cluster.fork();
  }

  for (const id in cluster.workers) {
    cluster.workers[id].on('message', messageHandler);
  }

} else {

  // Worker processes have a http server.
  http.Server((req, res) => {
    res.writeHead(200);
    res.end('hello world\n');

    // Notify primary about the request
    process.send({ cmd: 'notifyRequest' });
  }).listen(8000);
}const cluster = require('node:cluster');
const http = require('node:http');
const process = require('node:process');

if (cluster.isPrimary) {

  // Keep track of http requests
  let numReqs = 0;
  setInterval(() => {
    console.log(`numReqs = ${numReqs}`);
  }, 1000);

  // Count requests
  function messageHandler(msg) {
    if (msg.cmd && msg.cmd === 'notifyRequest') {
      numReqs += 1;
    }
  }

  // Start workers and listen for messages containing notifyRequest
  const numCPUs = require('node:os').availableParallelism();
  for (let i = 0; i < numCPUs; i++) {
    cluster.fork();
  }

  for (const id in cluster.workers) {
    cluster.workers[id].on('message', messageHandler);
  }

} else {

  // Worker processes have a http server.
  http.Server((req, res) => {
    res.writeHead(200);
    res.end('hello world\n');

    // Notify primary about the request
    process.send({ cmd: 'notifyRequest' });
  }).listen(8000);
}copy

イベント: 'online'#

cluster.on('online') イベントに似ていますが、このワーカーに固有です。

cluster.fork().on('online', () => {
  // Worker is online
}); copy

ワーカー内では発行されません。

worker.disconnect()#

• 戻り値: <cluster.Worker> worker への参照。

ワーカーでは、この関数はすべてのサーバーを閉じ、それらのサーバーの 'close' イベントを待機してから、IPC チャネルを切断します。

プライマリでは、ワーカーに内部メッセージが送信され、それ自体で .disconnect() を呼び出すようにします。

.exitedAfterDisconnect が設定されます。

サーバーが閉じられると、新しい接続は受け付けられなくなりますが、他のリスニング中のワーカーは接続を受け付けることができます。既存の接続は通常通り閉じることができます。接続がなくなった場合、server.close()を参照してください。ワーカーへのIPCチャネルが閉じられ、正常に終了できるようになります。

上記はサーバー接続にのみ適用され、クライアント接続はワーカーによって自動的に閉じられることはなく、切断は終了前にそれらが閉じるのを待機しません。

ワーカーでは、process.disconnectが存在しますが、これはこの関数ではありません。disconnect()です。

長時間接続を維持するサーバー接続は、ワーカーの切断を妨げる可能性があるため、メッセージを送信して、それらを閉じるためにアプリケーション固有のアクションを実行できるようにすると便利な場合があります。また、'disconnect'イベントが一定時間後に発行されない場合は、ワーカーを強制終了するタイムアウトを実装すると便利な場合があります。

if (cluster.isPrimary) {
  const worker = cluster.fork();
  let timeout;

  worker.on('listening', (address) => {
    worker.send('shutdown');
    worker.disconnect();
    timeout = setTimeout(() => {
      worker.kill();
    }, 2000);
  });

  worker.on('disconnect', () => {
    clearTimeout(timeout);
  });

} else if (cluster.isWorker) {
  const net = require('node:net');
  const server = net.createServer((socket) => {
    // Connections never end
  });

  server.listen(8000);

  process.on('message', (msg) => {
    if (msg === 'shutdown') {
      // Initiate graceful close of any connections to server
    }
  });
} copy

worker.exitedAfterDisconnect#

• <boolean>

このプロパティは、ワーカーが.disconnect()によって終了した場合はtrueです。ワーカーが他の方法で終了した場合はfalseです。ワーカーが終了していない場合はundefinedです。

ブール値のworker.exitedAfterDisconnectを使用すると、自発的な終了と偶発的な終了を区別でき、プライマリはこの値に基づいてワーカーを再起動しないことを選択できます。

cluster.on('exit', (worker, code, signal) => {
  if (worker.exitedAfterDisconnect === true) {
    console.log('Oh, it was just voluntary – no need to worry');
  }
});

// kill worker
worker.kill(); copy

worker.id#

• <integer>

新しいワーカーにはそれぞれ固有のIDが付与され、このIDはidに格納されます。

ワーカーが稼働中は、これはcluster.workersでワーカーをインデックスするキーです。

worker.isConnected()#

この関数は、ワーカーがIPCチャネルを介してプライマリに接続されている場合はtrueを返し、それ以外の場合はfalseを返します。ワーカーは作成されるとプライマリに接続されます。'disconnect'イベントが発行された後、切断されます。

worker.isDead()#

この関数は、ワーカーのプロセスが終了した場合（終了またはシグナルが送られたため）はtrueを返します。それ以外の場合は、falseを返します。

import cluster from 'node:cluster';
import http from 'node:http';
import { availableParallelism } from 'node:os';
import process from 'node:process';

const numCPUs = availableParallelism();

if (cluster.isPrimary) {
  console.log(`Primary ${process.pid} is running`);

  // Fork workers.
  for (let i = 0; i < numCPUs; i++) {
    cluster.fork();
  }

  cluster.on('fork', (worker) => {
    console.log('worker is dead:', worker.isDead());
  });

  cluster.on('exit', (worker, code, signal) => {
    console.log('worker is dead:', worker.isDead());
  });
} else {
  // Workers can share any TCP connection. In this case, it is an HTTP server.
  http.createServer((req, res) => {
    res.writeHead(200);
    res.end(`Current process\n ${process.pid}`);
    process.kill(process.pid);
  }).listen(8000);
}const cluster = require('node:cluster');
const http = require('node:http');
const numCPUs = require('node:os').availableParallelism();
const process = require('node:process');

if (cluster.isPrimary) {
  console.log(`Primary ${process.pid} is running`);

  // Fork workers.
  for (let i = 0; i < numCPUs; i++) {
    cluster.fork();
  }

  cluster.on('fork', (worker) => {
    console.log('worker is dead:', worker.isDead());
  });

  cluster.on('exit', (worker, code, signal) => {
    console.log('worker is dead:', worker.isDead());
  });
} else {
  // Workers can share any TCP connection. In this case, it is an HTTP server.
  http.createServer((req, res) => {
    res.writeHead(200);
    res.end(`Current process\n ${process.pid}`);
    process.kill(process.pid);
  }).listen(8000);
}copy

worker.kill([signal])#

• signal <string> ワーカープロセスに送信するkillシグナルの名前。デフォルト: 'SIGTERM'

この関数はワーカーを強制終了します。プライマリワーカーでは、worker.processを切断し、切断後にsignalで強制終了します。ワーカーでは、signalでプロセスを強制終了します。

kill()関数は、正常な切断を待たずにワーカープロセスを強制終了します。worker.process.kill()と同じ動作をします。

このメソッドは、後方互換性のためにworker.destroy()としてエイリアスされています。

ワーカーでは、process.kill()が存在しますが、これはこの関数ではありません。kill()です。

worker.process#

• <ChildProcess>

すべてのワーカーはchild_process.fork()を使用して作成されます。この関数から返されるオブジェクトは.processとして格納されます。ワーカーでは、グローバルなprocessが格納されます。

参照: Child Process モジュール。

ワーカーは、processで'disconnect'イベントが発生し、.exitedAfterDisconnectがtrueでない場合、process.exit(0)を呼び出します。これにより、偶発的な切断から保護されます。

worker.send(message[, sendHandle[, options]][, callback])#

• message <Object>
• sendHandle <Handle>
• options <Object> options引数は、存在する場合、特定のタイプのハンドルを送信する際のパラメータ化に使用されるオブジェクトです。optionsは次のプロパティをサポートします。
  • keepOpen <boolean> net.Socketのインスタンスを渡すときに使用できる値です。trueの場合、ソケットは送信プロセスで開いたままになります。デフォルト: false。
• callback <Function>
• 戻り値: <boolean>

ワーカーまたはプライマリに、必要に応じてハンドル付きでメッセージを送信します。

プライマリでは、これは特定のワーカーにメッセージを送信します。ChildProcess.send()と同じです。

ワーカーでは、これはプライマリにメッセージを送信します。process.send()と同じです。

この例では、プライマリからのすべてのメッセージをエコーバックします。

if (cluster.isPrimary) {
  const worker = cluster.fork();
  worker.send('hi there');

} else if (cluster.isWorker) {
  process.on('message', (msg) => {
    process.send(msg);
  });
} copy

イベント: 'disconnect'#

• worker <cluster.Worker>

ワーカーのIPCチャネルが切断された後に発行されます。これは、ワーカーが正常に終了した場合、強制終了された場合、または手動で切断された場合（worker.disconnect()など）に発生する可能性があります。

'disconnect'イベントと'exit'イベントの間には遅延が生じる可能性があります。これらのイベントを使用して、プロセスがクリーンアップでスタックしているか、長時間接続があるかどうかを検出できます。

cluster.on('disconnect', (worker) => {
  console.log(`The worker #${worker.id} has disconnected`);
}); copy

イベント: 'exit'#

• worker <cluster.Worker>
• code <number> 通常に終了した場合の終了コード。
• signal <string> プロセスを強制終了させたシグナル名 (例: 'SIGHUP')。

いずれかのワーカーが終了すると、クラスターモジュールは'exit'イベントを発行します。

これは、.fork()を再度呼び出すことによって、ワーカーを再起動するために使用できます。

cluster.on('exit', (worker, code, signal) => {
  console.log('worker %d died (%s). restarting...',
              worker.process.pid, signal || code);
  cluster.fork();
}); copy

参照: child_process イベント: 'exit'。

イベント: 'fork'#

• worker <cluster.Worker>

新しいワーカーがフォークされると、クラスターモジュールは'fork'イベントを発行します。これは、ワーカーのアクティビティをログに記録し、カスタムタイムアウトを作成するために使用できます。

const timeouts = [];
function errorMsg() {
  console.error('Something must be wrong with the connection ...');
}

cluster.on('fork', (worker) => {
  timeouts[worker.id] = setTimeout(errorMsg, 2000);
});
cluster.on('listening', (worker, address) => {
  clearTimeout(timeouts[worker.id]);
});
cluster.on('exit', (worker, code, signal) => {
  clearTimeout(timeouts[worker.id]);
  errorMsg();
}); copy

イベント: 'listening'#

• worker <cluster.Worker>
• address <Object>

ワーカーからlisten()を呼び出した後、サーバーで'listening'イベントが発行されると、プライマリのclusterでも'listening'イベントが発行されます。

イベントハンドラーは、2つの引数で実行されます。workerにはワーカーオブジェクトが含まれ、addressオブジェクトには次の接続プロパティが含まれます：address、port、およびaddressType。これは、ワーカーが複数のアドレスでリッスンしている場合に非常に役立ちます。

cluster.on('listening', (worker, address) => {
  console.log(
    `A worker is now connected to ${address.address}:${address.port}`);
}); copy

addressTypeは次のいずれかです。

• 4 (TCPv4)
• 6 (TCPv6)
• -1 (Unixドメインソケット)
• 'udp4'または'udp6' (UDPv4またはUDPv6)

イベント: 'message'#

• worker <cluster.Worker>
• message <Object>
• handle <undefined> | <Object>

クラスタープライマリが任意のワーカーからメッセージを受信したときに発行されます。

参照: child_process イベント: 'message'。

イベント: 'online'#

• worker <cluster.Worker>

新しいワーカーをフォークした後、ワーカーはオンラインメッセージで応答する必要があります。プライマリがオンラインメッセージを受信すると、このイベントを発行します。'fork'と'online'の違いは、forkはプライマリがワーカーをフォークしたときに発行され、'online'はワーカーが実行中のときに発行されることです。

cluster.on('online', (worker) => {
  console.log('Yay, the worker responded after it was forked');
}); copy

イベント: 'setup'#

• settings <Object>

.setupPrimary()が呼び出されるたびに発行されます。

settingsオブジェクトは、.setupPrimary()が呼び出された時点のcluster.settingsオブジェクトであり、1回のティックで.setupPrimary()が複数回呼び出される可能性があるため、助言のみです。

精度が重要な場合は、cluster.settingsを使用してください。

cluster.disconnect([callback])#

• callback <Function> すべてのワーカーが切断され、ハンドルが閉じられたときに呼び出されます。

cluster.workers内の各ワーカーで.disconnect()を呼び出します。

それらが切断されると、すべての内部ハンドルが閉じられ、他のイベントが待機していない場合、プライマリプロセスが正常に終了できます。

このメソッドは、完了時に呼び出されるオプションのコールバック引数を取ります。

これはプライマリプロセスからのみ呼び出すことができます。

cluster.fork([env])#

• env <Object> ワーカープロセスの環境に追加するキーと値のペア。
• 戻り値: <cluster.Worker>

新しいワーカープロセスを生成します。

これはプライマリプロセスからのみ呼び出すことができます。

cluster.isMaster#

cluster.isPrimary の非推奨のエイリアスです。

cluster.isPrimary#

• <boolean>

プロセスがプライマリである場合は true です。これは process.env.NODE_UNIQUE_ID によって決定されます。process.env.NODE_UNIQUE_ID が未定義の場合、isPrimary は true です。

cluster.isWorker#

• <boolean>

プロセスがプライマリでない場合 (cluster.isPrimary の否定) は true です。

cluster.schedulingPolicy#

スケジューリングポリシーは、ラウンドロビンの場合は cluster.SCHED_RR、オペレーティングシステムに任せる場合は cluster.SCHED_NONE です。これはグローバル設定であり、最初のワーカーが生成されるか、.setupPrimary() が呼び出されるか、どちらか早い方が発生した時点で事実上固定されます。

SCHED_RR は、Windows を除くすべてのオペレーティングシステムでデフォルトです。Windows では、libuv が大きなパフォーマンス低下を招くことなく IOCP ハンドルを効率的に分散できるようになれば、SCHED_RR に変更されます。

cluster.schedulingPolicy は、環境変数 NODE_CLUSTER_SCHED_POLICY を介して設定することもできます。有効な値は 'rr' および 'none' です。

cluster.settings#

• <Object>
  • execArgv <string[]> Node.js 実行可能ファイルに渡される文字列引数のリスト。デフォルト: process.execArgv。
  • exec <string> ワーカーファイルのファイルパス。デフォルト: process.argv[1]。
  • args <string[]> ワーカーに渡される文字列引数。デフォルト: process.argv.slice(2)。
  • cwd <string> ワーカープロセスのカレントワーキングディレクトリ。デフォルト: undefined (親プロセスから継承)。
  • serialization <string> プロセス間でメッセージを送信するために使用されるシリアル化の種類を指定します。使用可能な値は 'json' および 'advanced' です。詳細については、child_process の高度なシリアル化 を参照してください。デフォルト: false。
  • silent <boolean> 親の標準入出力に出力を送信するかどうか。デフォルト: false。
  • stdio <Array> フォークされたプロセスの標準入出力を構成します。cluster モジュールは IPC に依存して機能するため、この構成には 'ipc' エントリを含める必要があります。このオプションが指定された場合、silent をオーバーライドします。child_process.spawn() の stdio を参照してください。
  • uid <number> プロセスのユーザー ID を設定します。( setuid(2) を参照してください。)
  • gid <number> プロセスのグループ ID を設定します。( setgid(2) を参照してください。)
  • inspectPort <number> | <Function> ワーカーのインスペクターポートを設定します。これは数値、または引数を取らず数値を返す関数にすることができます。デフォルトでは、各ワーカーにはプライマリの process.debugPort からインクリメントされた独自のポートが割り当てられます。
  • windowsHide <boolean> 通常 Windows システムで作成されるフォークされたプロセスのコンソールウィンドウを非表示にします。デフォルト: false。

.setupPrimary() (または .fork()) を呼び出した後、この settings オブジェクトには、デフォルト値を含む設定が含まれます。

このオブジェクトは、手動で変更したり設定したりすることを意図していません。

cluster.setupMaster([settings])#

.setupPrimary() の非推奨のエイリアスです。

cluster.setupPrimary([settings])#

• settings <Object> cluster.settings を参照してください。

setupPrimary は、デフォルトの 'fork' の動作を変更するために使用されます。呼び出されると、設定は cluster.settings に存在します。

設定の変更は、.fork() の今後の呼び出しにのみ影響し、既に実行中のワーカーには影響しません。

.setupPrimary() を介して設定できないワーカーの唯一の属性は、.fork() に渡される env です。

上記のデフォルトは最初の呼び出しにのみ適用されます。後の呼び出しのデフォルトは、cluster.setupPrimary() が呼び出された時点での現在の値です。

import cluster from 'node:cluster';

cluster.setupPrimary({
  exec: 'worker.js',
  args: ['--use', 'https'],
  silent: true,
});
cluster.fork(); // https worker
cluster.setupPrimary({
  exec: 'worker.js',
  args: ['--use', 'http'],
});
cluster.fork(); // http workerconst cluster = require('node:cluster');

cluster.setupPrimary({
  exec: 'worker.js',
  args: ['--use', 'https'],
  silent: true,
});
cluster.fork(); // https worker
cluster.setupPrimary({
  exec: 'worker.js',
  args: ['--use', 'http'],
});
cluster.fork(); // http workercopy

これはプライマリプロセスからのみ呼び出すことができます。

cluster.worker#

• <Object>

現在のワーカーオブジェクトへの参照。プライマリプロセスでは利用できません。

import cluster from 'node:cluster';

if (cluster.isPrimary) {
  console.log('I am primary');
  cluster.fork();
  cluster.fork();
} else if (cluster.isWorker) {
  console.log(`I am worker #${cluster.worker.id}`);
}const cluster = require('node:cluster');

if (cluster.isPrimary) {
  console.log('I am primary');
  cluster.fork();
  cluster.fork();
} else if (cluster.isWorker) {
  console.log(`I am worker #${cluster.worker.id}`);
}copy

cluster.workers#

• <Object>

id フィールドをキーとして、アクティブなワーカーオブジェクトを格納するハッシュ。これにより、すべてのワーカーを簡単にループ処理できます。プライマリプロセスでのみ利用可能です。

ワーカーは、ワーカーが接続を切断してかつ終了した後、cluster.workers から削除されます。これらの 2 つのイベント間の順序を事前に決定することはできません。ただし、cluster.workers リストからの削除は、最後の 'disconnect' または 'exit' イベントが発行される前に行われることが保証されています。

import cluster from 'node:cluster';

for (const worker of Object.values(cluster.workers)) {
  worker.send('big announcement to all workers');
}const cluster = require('node:cluster');

for (const worker of Object.values(cluster.workers)) {
  worker.send('big announcement to all workers');
}copy