Node.js clusterモジュールについて

シングルスレッドで動作するNode.jsにおいて、マルチコアCPUを持っているマシンの能力を最大限引き出すために、
複数のワーカープロセスを起動して処理を分散させたいといったニーズがあると思います。
そのときに重要なclusterモジュールについて、いつも業務で使っていますが、さらなる理解のため、整理してみます。
この記事は、Node.jsバージョンv6.x系を前提とした話となります。

clusterモジュール

clusterモジュールについて、大まかな処理の流れは、ドキュメントやNode.jsユーザグループのブログ(Node.js v0.6時のもので古いです)が非常に参考になります。
blog.nodejs.jp

clusterモジュールによるプロセス間通信には2つの方法があり、それぞれ紹介していきます。

マスタープロセスによるロードバランシング

この方法では、マスタープロセスは指定したポートで待ち受けて、新たなコネクションを受け付けます。
リクエストがあるたびに、マスタープロセスはラウンドロビン方式でワーカープロセスに処理を委譲します。
この方法の場合、リクエストが来るたびにマスタープロセスがロードバランサーとして仕事をするために、同時アクセス数が増加したときに、サーバ全体のスループットが、
マスタープロセスの上限に依存します。
この方法は、上記の問題から、webサーバにおいてあまり利用されることがなく、複数マシン間で処理を分散する場合のリバースプロキシとして用いられることが多いです。

カーネルによるロードバランシング (v0.11.2以降デフォルト)

マスタープロセスからソケットを作成し、ワーカプロセスに接続済みソケットを渡してクライアントとワーカプロセス間で接続を行う方法です。
このアプローチはNode.js v0.11.2からUnix系プラットフォーム(windows以外のこと)でデフォルトになりました。 理論的には、この方法が最も良いパフォーマンスを提供します。

clusterモジュールを実際に使ってみる

実際にclusterモジュールを使って動かしてみるのが理解しやすいので、公式ドキュメントのサンプルを参考にwebサーバを作ってみました。

// clusterテスト用
// master workerプロセス共に、このプログラムの上から処理が走る

"use strict";

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

if (cluster.isMaster) {
  // masterプロセスの場合の処理

  // Keep track of http requests
  var 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
  // cpu数を取得し、cpu数分だけ、ワーカープロセスをforkする
  const numCPUs = require('os').cpus().length;
  for (var i = 0; i < numCPUs; i++) {
    cluster.fork();
  }

  Object.keys(cluster.workers).forEach((id) => {
    // クラスター毎にメッセージハンドラー(リクエスト数カウント処理)を設定する
    // messageイベントを受信したときに発火する
    cluster.workers[id].on('message', messageHandler);
  });

  cluster.on('exit', (worker, code, signal) => {
    console.log('worker %d died (%s). restarting...', worker.process.pid, signal || code);
  });
} else {
  // workerプロセスの処理

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

    // notify master about the requests
    process.send({
      cmd: 'notifyRequest'
    });
  }).listen(8000);
}

マスタープロセスでの処理

マスター・ワーカープロセスともに上記のコードを上から実行していくのですが、それぞれのプロセスにおいて cluster.isMasterという値を持っており、 この値でマスター・ワーカープロセスを判断しています。

マスタープロセスの場合、CPUコア数分のワーカープロセスをforkする処理をおこなっており、それぞれのワーカープロセスに対して、
プロセス間メッセージを受信したときのmessageイベントと、ワーカープロセスが死んだときのexitイベントに対するハンドラ関数をそれぞれセットしています。

  // masterプロセスの場合の処理

  // Keep track of http requests
  var 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
  // cpu数を取得し、cpu数分だけ、ワーカープロセスをforkする
  const numCPUs = require('os').cpus().length;
  for (var i = 0; i < numCPUs; i++) {
    cluster.fork();
  }

  Object.keys(cluster.workers).forEach((id) => {
    // クラスター毎にメッセージハンドラー(リクエスト数カウント処理)を設定する
    // messageイベントを受信したときに発火する
    cluster.workers[id].on('message', messageHandler);
  });

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

ワーカープロセスでの処理

ワーカープロセスでは、8000番ポートでhttpリクエストを受付、hello worldを返すhttpサーバの処理を行っています。 このプログラムでは、リクエストを受け付けた際に、 process.sendで、マスタープロセスに対してリクエストを受け付けたことを通知しています。
マスタープロセスでは、このnotifyRequestコマンドを受け取ると、リクエストカウントをインクリメントして、1000ミリ秒毎にリクエストカウントを表示させています。

  // workerプロセスの処理

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

    // notify master about the requests
    process.send({
      cmd: 'notifyRequest'
    });
  }).listen(8000);

サーバの起動

実際にサーバを起動してみて、挙動を確認します。
先程書いたプログラムを/Users/testuser/Documents/nodejs-sandbox配下のcluster-test.jsに作成し、ターミナルで実行します。

$ node cluster-test.js
numReqs = 0     // 1000ミリ秒毎にリクエストカウント表示
numReqs = 0

これでlocalhostの8000番ポートでリクエストを待ち受けている状態になりました。 ためしに、起動されているnodeプロセスを確認すると、しっかりとCPUコア数分(8コア)ワーカープロセスが立ち上がっていることが確認できます。

testuser     81767   0.0  0.1  3050252  21320 s003  S+    2:31AM   0:00.15 /Users/testuser/.nodebrew/node/v6.9.2/bin/node /Users/testuser/Documents/nodejs-sandbox/cluster-test.js
testuser     81766   0.0  0.1  3068684  21408 s003  S+    2:31AM   0:00.16 /Users/testuser/.nodebrew/node/v6.9.2/bin/node /Users/testuser/Documents/nodejs-sandbox/cluster-test.js
testuser     81765   0.0  0.1  3041036  20876 s003  S+    2:31AM   0:00.16 /Users/testuser/.nodebrew/node/v6.9.2/bin/node /Users/testuser/Documents/nodejs-sandbox/cluster-test.js
testuser     81764   0.0  0.1  3068684  21580 s003  S+    2:31AM   0:00.16 /Users/testuser/.nodebrew/node/v6.9.2/bin/node /Users/testuser/Documents/nodejs-sandbox/cluster-test.js
testuser     81763   0.0  0.1  3049228  20980 s003  S+    2:31AM   0:00.16 /Users/testuser/.nodebrew/node/v6.9.2/bin/node /Users/testuser/Documents/nodejs-sandbox/cluster-test.js
testuser     81762   0.0  0.1  3068684  21652 s003  S+    2:31AM   0:00.16 /Users/testuser/.nodebrew/node/v6.9.2/bin/node /Users/testuser/Documents/nodejs-sandbox/cluster-test.js
testuser     81761   0.0  0.1  3068684  21604 s003  S+    2:31AM   0:00.15 /Users/testuser/.nodebrew/node/v6.9.2/bin/node /Users/testuser/Documents/nodejs-sandbox/cluster-test.js
testuser     81760   0.0  0.1  3068684  21464 s003  S+    2:31AM   0:00.16 /Users/testuser/.nodebrew/node/v6.9.2/bin/node /Users/testuser/Documents/nodejs-sandbox/cluster-test.js
testuser     81759   0.0  0.1  3059468  21468 s003  S+    2:31AM   0:00.13 node cluster-test.js

この状態のまま、クライアント用にターミナルをもう１つ立ち上げ、curlでhttpリクエストを送るとレスポンスが返ってくるのを確認できます。

$ curl localhost:8000
hello world 

さらに、先程から起動していたサーバ側のターミナルには、リクエストカウントが1増えていることが確認できます。

numReqs = 0
---Request header. { 'user-agent': 'curl/7.37.0',
  host: 'localhost:8000',
  accept: '*/*' }
numReqs = 1

以上のように、clusterモジュールを使って、複数プロセスでwebサーバを立ち上げることができました。
clusterはNode.jsにおける負荷分散で欠かせないものなのでしっかりと理解することが重要かと思います。