Socket.ioサーバをスケールアウトする #Socket.io

Socket.ioでコンソールベースのチャットを作ってみるで作ったサーバをスケールアウトしたい。
Node.jsプロセスのスケールアウトといえばclusterなので、これを使って単純にスケールアウトしてみる。

index.js

const os = require('os');
const cluster = require('cluster');
const io = require('socket.io')();

if (cluster.isMaster) {
  os.cpus().forEach(() => {
    const worker = cluster.fork();
    console.log('CLUSTER: Worker %d started', worker.id);
  });

  cluster.on('exit', () => {
    const worker = cluster.fork();
    console.log('CLUSTER: Worker %d started', worker.id);
  });
} else {
  io.listen(3000);

  io.on('connection', (socket) => {
    console.log(`connected, id: ${socket.id}`);

    socket.on('chat message', (user, message) => {
      data = `${message} from ${user}`;
      console.log(data);
      socket.broadcast.emit('chat message', data);
    });

    socket.on('disconnect', () => {
      console.log(`disconnected, id: ${socket.id}`);
    });
  });
}

サーバを起動する。

$ node index.js
CLUSTER: Worker 1 started
CLUSTER: Worker 2 started
CLUSTER: Worker 3 started
CLUSTER: Worker 4 started

クライアントから接続する。
※クライアントのプログラムはこちら。

$ node client.js UserA
Error: { Error: xhr poll error
    at XHR.Transport.onError (/Users/takehiro/Documents/git/socketio-app/node_modules/engine.io-client/lib/transport.js:64:13)
    at Request.<anonymous> (/Users/takehiro/Documents/git/socketio-app/node_modules/engine.io-client/lib/transports/polling-xhr.js:129:10)
    at Request.Emitter.emit (/Users/takehiro/Documents/git/socketio-app/node_modules/engine.io-client/node_modules/component-emitter/index.js:133:20)
    at Request.onError (/Users/takehiro/Documents/git/socketio-app/node_modules/engine.io-client/lib/transports/polling-xhr.js:307:8)
    at Timeout._onTimeout (/Users/takehiro/Documents/git/socketio-app/node_modules/engine.io-client/lib/transports/polling-xhr.js:254:18)
    at ontimeout (timers.js:365:14)
    at tryOnTimeout (timers.js:237:5)
    at Timer.listOnTimeout (timers.js:207:5) type: 'TransportError', description: 400 }
disconnected

XHR Pollingエラーになってしまう。
Socket.ioクライアントはコネクション確立のハンドシェイク時にいくつかのアクセスを行うが、clusterによってこれらのアクセスが毎回違うサーバプロセスに振り分けられてしまうため、コネクションが確立できないのである。

この問題を解決する方法として、Socket.IOの公式ドキュメントではsticky-sessionモジュールを使用する方法を紹介している。

sticky-sessionモジュールを使用したサーバの実装がこちら。

const cluster = require('cluster');
const io = require('socket.io')();
const sticky = require('sticky-session');
const http = require('http');

const server = http.createServer((req, res) => {
  res.end('worker: ' + cluster.worker.id);
});

io.attach(server);
isWorker = sticky.listen(server, 3000);

if (isWorker) {
  io.on('connection', (socket) => {
    console.log(`connected, id: ${socket.id}`);

    socket.on('chat message', (user, message) => {
      data = `${message} from ${user}`;
      console.log(data);
      socket.broadcast.emit('chat message', data);
    });

    socket.on('disconnect', () => {
      console.log(`disconnected, id: ${socket.id}`);
    });
  });
}

sticky-sessionモジュールの内部でclusterが使われていて、sticky.listenというAPIを呼べば複数のプロセスを自動でForkしてくれる。
デフォルトではCPUの数だけForkするが、オプションで数を指定することもできる。

また、sticky.listenの戻り値はbooleanで、自身がMasterであればfalse、Workerであればtrueを返す。
これを利用し、Workerとしてこのプログラムが実行された場合はSocket.IOサーバとして機能するようにしている。

上記の実装により、XHR Pollingエラーの問題は解消される。
毎回同じプロセスに割り振られている様子は以下から確認できる。

$ curl localhost:3000
worker: 1

$ curl localhost:3000
worker: 1

...

なお、sticky-sessionは送信元IPを元に、サーバプロセスへの振り分け先を固定するようである。