本記事のテーマ

express.jsプロセスのクラスタリング化
セッション情報の格納先の変更

元来、Node.jsはマルチスレッドではないため、全てのリクエストは一つのプロセスにより処理されます。この点が多くのリクエストを処理する必要があるWebサービスにおいてボトルネックとなり、パフォーマンスの劣化につながるとの懸念があります。

その為、Node.js v0.8よりClusterモジュールが導入されました。

この機能によりサーバプロセスをクラスタ化し、処理を並列化することでスループットの向上が見込めます。

しかし、サーバプロセスをクラスタリングする上では、セッション情報の格納先に注意する必要あります。

express-sessionモジュールによりセッション管理を行った場合、セッション情報はデフォルトではメモリー上に格納されます。この状態ではセッションがプロセスを跨った場合に、セッション情報が取得できない、という問題が発生します。従って、全プロセスが参照可能な領域にセッション情報を格納する必要があります。
そもそも、メモリ上での運用は非推奨となっています（以下、https://github.com/expressjs/session）

Warning The default server-side session storage, MemoryStore, is purposely not designed for a production environment. It will leak memory under most conditions, does not scale past a single process, and is meant for debugging and developing.

当記事では、セッション情報の格納先として以下を指定しています。
(1)ファイル
(2)Redis
ぞれぞれの手順を順番に説明します。

※以下、JavaScriptはECMAScriptモジュール形式で記述します。

まずは、シンプルなExpress.jsプロセスから

express.mjs

import express from 'express';
var app = express();

var server = app.listen(3000, function () {
  console.log("Server Process has started:" + server.address().port);
});

app.get("/", function (req, res, next) {
  console.log("/ called. ");
  res.json({"result":"hello"});
});

次に、サーバプロセスのクラスタ化

express.mjs

import express from 'express';

import cluster from "cluster";
// 負荷分散の方法を指定する
//  - cluster.SCHED_RR : ラウンドロビン
//  - cluster.SCHED_NONE : OS依存
cluster.schedulingPolicy = cluster.SCHED_NONE;

// CPU数を取得
import numCPUs from"os";
const cpus = numCPUs.cpus().length;


if (cluster.isPrimary) {

  console.log(`Master ${process.pid} has started`);

  for (let i=0 ; i<cpus ; i++) {
    console.log(`Worker [${i}] fork`);
    cluster.fork();
  }

  cluster.on("exit", (worker) => {
    console.log(`Worker ${worker.process.pid} died`);
    cluster.fork();
  });

}
else {

  console.log(`Worker has [${cluster.worker.id}] ${process.pid} started`);

  const app = express();
  var server = app.listen(3000, function () {
    console.log("Express.js is listening to PORT:" + server.address().port);
  });

  app.get("/", function (req, res, next) {
    console.log(`/ called. Worker [${cluster.worker.id}] ${process.pid} `);
    res.json({"result":"hello"});
  });

}

clusterモジュールを用いて、マスタープロセスから実際にリクエストを処理する子プロセスを複数起動（Fork）します。このサンプルでは、子プロセスが何らかの理由により停止した場合は、再度Forkされます。

セッション上に情報を保持する

express.mjs

import express from 'express';

import cluster from "cluster";
// 負荷分散の方法を指定する
//  - cluster.SCHED_RR : ラウンドロビン
//  - cluster.SCHED_NONE : OS依存
cluster.schedulingPolicy = cluster.SCHED_NONE;

// CPU数を取得
import numCpus from "os";
const cpus = numCpus.cpus().length;

// HTTPセッションを使用
import cookieParser from "cookie-parser";
import session from "express-session";


if (cluster.isPrimary) {

  console.log(`Master ${process.pid} has started`);

  for (let i=0 ; i<cpus ; i++) {
    console.log(`Worker [${i}] fork`);
    cluster.fork();
  }

}
else {

  console.log(`Worker has [${cluster.worker.id}] ${process.pid} started`);

  const app = express();

  app.use(cookieParser());
  app.use(session({
    secret            : 'myApp',
    resave            : false,
    saveUninitialized : false
  }));

  var server = app.listen(3000, function () {
    console.log("Express.js is listening to PORT:" + server.address().port);
  });

  app.get("/kill" , function (req, res, next) {
    const pid = process.pid;
    setTimeout(() => {
      console.log("killed");
      process.kill(pid);
    },10000);
    console.log(`/kill called. Worker [${cluster.worker.id}] ${process.pid} ${req.session.sid}`);
    res.json('{"sessionId":"' + req.session.sid + '"}');
  });

  app.get("/check", function (req, res, next) {
    console.log(`/check called. Worker [${cluster.worker.id}] ${process.pid} ${req.session.sid}`);
    if ( req.session.sid === undefined || req.session.sid === null || req.session.sid === '' ) {
      console.log(`sessionId was null. [${cluster.worker.id}] ${process.pid}`);
    }
    res.json('{"sessionId":"' + req.session.sid + '"}');
  });

  app.get("/start", function (req, res, next) {
    req.session.sid = Math.floor(Math.random() * 100);
    console.log(`/start called. Worker [${cluster.worker.id}] ${process.pid} ${req.session.sid}`);
    res.json('{"sessionId":"' + req.session.sid + '"}');
  });

}

・/start ：セッションの開始
・/check ：セッションに保持した情報の参照
・/kill　：プロセスの停止

※テスト用にプロセスの停止を検知し、再度Forkする処理は削除しています。

/startによりセッションを作成します。
/checkを呼び出すと、同一のプロセスがリクエストを処理していることが確認できます。
この時、セッションに格納した情報が参照可能であることを確認します。
/killを実行すると、リクエストを処理したプロセスは停止します。
その後、再度/checkを呼び出すと、前回リクエストを処理したプロセスが存在しないため、別のプロセスがリクエストを処理しますが、この時、セッションに格納された情報が参照不可となっていることがわかります。
次のサンプルで、この問題に対応します。

クラスタリングされたプロセス間でセッション情報を共有する（ファイル編）

express.mjs

import express from 'express';

import cluster from "cluster";
// 負荷分散の方法を指定する
//  - cluster.SCHED_RR : ラウンドロビン
//  - cluster.SCHED_NONE : OS依存
cluster.schedulingPolicy = cluster.SCHED_NONE;

// CPU数を取得
import numCpus from "os";
const cpus = numCpus.cpus().length;

// HTTPセッションを使用
import cookieParser from "cookie-parser";
import session from "express-session";
import sessionFileStore from 'session-file-store';
const FileStore = sessionFileStore(session);
const sessionStoreParam  = new FileStore({ path: "sessions" });


if (cluster.isPrimary) {

  console.log(`Master ${process.pid} has started`);

  for (let i=0 ; i<cpus ; i++) {
    console.log(`Worker [${i}] fork`);
    cluster.fork();
  }

}
else {

  console.log(`Worker has [${cluster.worker.id}] ${process.pid} started`);

  const app = express();

  app.use(cookieParser());
  app.use(session({
    store             : new FileStore(sessionStoreParam),
    secret            : 'myApp',
    resave            : false,
    saveUninitialized : false
  }));

  var server = app.listen(3000, function () {
    console.log("Express.js is listening to PORT:" + server.address().port);
  });

  app.get("/kill" , function (req, res, next) {
    const pid = process.pid;
    setTimeout(() => {
      console.log("killed");
      process.kill(pid);
    },10000);
    console.log(`/kill called. Worker [${cluster.worker.id}] ${process.pid} ${req.session.sid}`);
    res.json('{"sessionId":"' + req.session.sid + '"}');
  });

  app.get("/check", function (req, res, next) {
    console.log(`/check called. Worker [${cluster.worker.id}] ${process.pid} ${req.session.sid}`);
    if ( req.session.sid === undefined || req.session.sid === null || req.session.sid === '' ) {
      console.log(`sessionId is null. [${cluster.worker.id}] ${process.pid}`);
    }
    res.json('{"sessionId":"' + req.session.sid + '"}');
  });

  app.get("/start", function (req, res, next) {
    req.session.sid = Math.floor(Math.random() * 100);
    console.log(`/start called. Worker [${cluster.worker.id}] ${process.pid} ${req.session.sid}`);
    res.json('{"sessionId":"' + req.session.sid + '"}');
  });

}

session-file-storeを使用し、セッション情報をファイルに格納するよう変更しました。

直前のサンプルと同様に /start /check /kill　を順番に呼び出してセッションに格納した情報が参照可能か確認します。
今回は、/killによりプロセスが停止した後、別のプロセスがリクエストを処理する際も、セッションに格納した情報が参照可能であることが確認できました。

クラスタリングされたプロセス間でセッション情報を共有する（Radis編）

express.mjs

import express from 'express';

import cluster from "cluster";
// 負荷分散の方法を指定する
//  - cluster.SCHED_RR : ラウンドロビン
//  - cluster.SCHED_NONE : OS依存
cluster.schedulingPolicy = cluster.SCHED_NONE;

// CPU数を取得
import numCPUs from "os";
const cpus = numCPUs.cpus().length;

// HTTPセッションを使用
import cookieParser from "cookie-parser";
import session from "express-session";
import redis from 'redis';
import RedisStore from "connect-redis"
const redisClient = redis.createClient();
await redisClient.connect();

if (cluster.isPrimary) {

  console.log(`Master ${process.pid} has started`);

  for (let i=0 ; i<cpus ; i++) {
    console.log(`Worker [${i}] fork`);
    cluster.fork();
  }

}
else {

  console.log(`Worker has [${cluster.worker.id}] ${process.pid} started`);

  const app = express();

  app.use(cookieParser());
  app.use(session({
    store             : new RedisStore({ client: redisClient }),
    secret            : 'myApp',
    resave            : false,
    saveUninitialized : false
  }));

  var server = app.listen(3000, function () {
    console.log("Express.js is listening to PORT:" + server.address().port);
  });

  app.get("/kill" , function (req, res, next) {
    const pid = process.pid;
    setTimeout(() => {
      console.log("killed");
      process.kill(pid);
    },10000);
    console.log(`/kill called. Worker [${cluster.worker.id}] ${process.pid} ${req.session.sid}`);
    res.json('{"sessionId":"' + req.session.sid + '"}');
  });

  app.get("/check", function (req, res, next) {
    console.log(`/check called. Worker [${cluster.worker.id}] ${process.pid} ${req.session.sid}`);
    if ( req.session.sid === undefined || req.session.sid === null || req.session.sid === '' ) {
      console.log(`sessionId was null. [${cluster.worker.id}] ${process.pid}`);
    }
    res.json('{"sessionId":"' + req.session.sid + '"}');
  });

  app.get("/start", function (req, res, next) {
    req.session.sid = Math.floor(Math.random() * 100);
    console.log(`/start called. Worker [${cluster.worker.id}] ${process.pid} ${req.session.sid}`);
    res.json('{"sessionId":"' + req.session.sid + '"}');
  });

}

セッション情報の格納先(storeオプション)をインメモリーデータベース（Redis）へ変更しています。
直前のサンプルと同様に、リクエストを処理するプロセスが変化しても、セッション情報を参照可能であることが確認できます。

※当サンプルを動かすためには、事前にRedisのインストールが必要です。

以上です。