70

用这个把我的头发扯掉......有没有人设法将Socket.IO 扩展到由 Node.js 的集群模块产生的多个“工作”进程？

假设我在四个工作进程（伪）上有以下内容：

// on the server
var express = require('express');
var server = express();
var socket = require('socket.io');
var io = socket.listen(server);

// socket.io
io.set('store', new socket.RedisStore);

// set-up connections...
io.sockets.on('connection', function(socket) {

  socket.on('join', function(rooms) {
    rooms.forEach(function(room) {
      socket.join(room);
    });
  });

  socket.on('leave', function(rooms) {
    rooms.forEach(function(room) {
      socket.leave(room);
    });
  });

});

// Emit a message every second
function send() {
  io.sockets.in('room').emit('data', 'howdy');
}

setInterval(send, 1000);

而在浏览器...

// on the client
socket = io.connect();
socket.emit('join', ['room']);

socket.on('data', function(data){
  console.log(data);
});

问题：由于四个单独的工作进程发送消息，我每秒都会收到四条消息。

如何确保消息只发送一次？

4

4 回答 4

110

编辑：在 Socket.IO 1.0+ 中，现在可以使用更简单的 Redis 适配器模块，而不是设置具有多个 Redis 客户端的存储。

var io = require('socket.io')(3000);
var redis = require('socket.io-redis');
io.adapter(redis({ host: 'localhost', port: 6379 }));

下面显示的示例看起来更像这样：

var cluster = require('cluster');
var os = require('os');

if (cluster.isMaster) {
  // we create a HTTP server, but we do not use listen
  // that way, we have a socket.io server that doesn't accept connections
  var server = require('http').createServer();
  var io = require('socket.io').listen(server);
  var redis = require('socket.io-redis');

  io.adapter(redis({ host: 'localhost', port: 6379 }));

  setInterval(function() {
    // all workers will receive this in Redis, and emit
    io.emit('data', 'payload');
  }, 1000);

  for (var i = 0; i < os.cpus().length; i++) {
    cluster.fork();
  }

  cluster.on('exit', function(worker, code, signal) {
    console.log('worker ' + worker.process.pid + ' died');
  }); 
}

if (cluster.isWorker) {
  var express = require('express');
  var app = express();

  var http = require('http');
  var server = http.createServer(app);
  var io = require('socket.io').listen(server);
  var redis = require('socket.io-redis');

  io.adapter(redis({ host: 'localhost', port: 6379 }));
  io.on('connection', function(socket) {
    socket.emit('data', 'connected to worker: ' + cluster.worker.id);
  });

  app.listen(80);
}

如果您有一个需要发布到其他 Socket.IO 进程的主节点，但它本身不接受套接字连接，请使用socket.io-emitter而不是socket.io-redis。

如果您在扩展时遇到问题，请使用DEBUG=*. Socket.IO 现在实现了调试，它还将打印出 Redis 适配器调试消息。示例输出：

socket.io:server initializing namespace / +0ms
socket.io:server creating engine.io instance with opts {"path":"/socket.io"} +2ms
socket.io:server attaching client serving req handler +2ms
socket.io-parser encoding packet {"type":2,"data":["event","payload"],"nsp":"/"} +0ms
socket.io-parser encoded {"type":2,"data":["event","payload"],"nsp":"/"} as 2["event","payload"] +1ms
socket.io-redis ignore same uid +0ms

如果您的主进程和子进程都显示相同的解析器消息，那么您的应用程序正在正确扩展。

如果您是从单个工作人员发出的，那么您的设置应该没有问题。您正在做的是从所有四个工作人员发出，并且由于 Redis 发布/订阅，消息不会重复，而是写入四次，正如您要求应用程序所做的那样。这是 Redis 功能的简单示意图：

Client  <--  Worker 1 emit -->  Redis
Client  <--  Worker 2  <----------|
Client  <--  Worker 3  <----------|
Client  <--  Worker 4  <----------|

正如你所看到的，当你从一个worker发出时，它会将emit发布到Redis，它会被其他订阅了Redis数据库的worker镜像。这也意味着您可以使用连接同一个实例的多个套接字服务器，并且在一个服务器上的发射将在所有连接的服务器上触发。

使用集群，当客户端连接时，它将连接到您的四个工作人员之一，而不是全部四个。这也意味着您从该工作人员发出的任何内容都只会向客户端显示一次。所以是的，应用程序正在扩展，但是你这样做的方式是，你从所有四个工作人员中发出，而 Redis 数据库正在使它就像你在一个工作人员上调用它四次一样。如果客户端实际连接到所有四个套接字实例，他们将每秒接收 16 条消息，而不是 4 条。

套接字处理的类型取决于您将拥有的应用程序的类型。如果您要单独处理客户端，那么您应该没有问题，因为连接事件只会为每个客户端的一个工作人员触发。如果您需要全局“心跳”，那么您可以在主进程中有一个套接字处理程序。由于主进程死亡时工作人员也会死亡，因此您应该抵消主进程的连接负载，让子进程处理连接。这是一个例子：

var cluster = require('cluster');
var os = require('os');

if (cluster.isMaster) {
  // we create a HTTP server, but we do not use listen
  // that way, we have a socket.io server that doesn't accept connections
  var server = require('http').createServer();
  var io = require('socket.io').listen(server);

  var RedisStore = require('socket.io/lib/stores/redis');
  var redis = require('socket.io/node_modules/redis');

  io.set('store', new RedisStore({
    redisPub: redis.createClient(),
    redisSub: redis.createClient(),
    redisClient: redis.createClient()
  }));

  setInterval(function() {
    // all workers will receive this in Redis, and emit
    io.sockets.emit('data', 'payload');
  }, 1000);

  for (var i = 0; i < os.cpus().length; i++) {
    cluster.fork();
  }

  cluster.on('exit', function(worker, code, signal) {
    console.log('worker ' + worker.process.pid + ' died');
  }); 
}

if (cluster.isWorker) {
  var express = require('express');
  var app = express();

  var http = require('http');
  var server = http.createServer(app);
  var io = require('socket.io').listen(server);

  var RedisStore = require('socket.io/lib/stores/redis');
  var redis = require('socket.io/node_modules/redis');

  io.set('store', new RedisStore({
    redisPub: redis.createClient(),
    redisSub: redis.createClient(),
    redisClient: redis.createClient()
  }));

  io.sockets.on('connection', function(socket) {
    socket.emit('data', 'connected to worker: ' + cluster.worker.id);
  });

  app.listen(80);
}

在示例中，有五个 Socket.IO 实例，一个是主实例，四个是子实例。主服务器从不调用listen()，因此该进程没有连接开销。但是，如果您在主进程上调用一个 emit，它将被发布到 Redis，四个工作进程将在它们的客户端上执行 emit。这抵消了工作人员的连接负载，如果工作人员死亡，您的主要应用程序逻辑将在主服务器中保持不变。

请注意，使用 Redis，即使在命名空间或房间中的所有发射都将由其他工作进程处理，就好像您从该进程触发了发射一样。换句话说，如果您有两个 Socket.IO 实例和一个 Redis 实例，emit()则在第一个工作人员中调用套接字会将数据发送到其客户端，而工作人员二将执行与从该工作人员调用发射相同的操作。

于 2013-09-06T04:57:18.963 回答
2

让 master 处理你的心跳（下面的例子）或在内部在不同的端口上启动多个进程，并使用 nginx（它也支持从 V1.3 开始的 websockets）对它们进行负载平衡。

集群与主

// on the server
var express = require('express');
var server = express();
var socket = require('socket.io');
var io = socket.listen(server);
var cluster = require('cluster');
var numCPUs = require('os').cpus().length;

// socket.io
io.set('store', new socket.RedisStore);

// set-up connections...
io.sockets.on('connection', function(socket) {
    socket.on('join', function(rooms) {
        rooms.forEach(function(room) {
            socket.join(room);
        });
    });

    socket.on('leave', function(rooms) {
        rooms.forEach(function(room) {
            socket.leave(room);
        });
    });

});

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

    // Emit a message every second
    function send() {
        console.log('howdy');
        io.sockets.in('room').emit('data', 'howdy');
    }

    setInterval(send, 1000);


    cluster.on('exit', function(worker, code, signal) {
        console.log('worker ' + worker.process.pid + ' died');
    }); 
}
于 2013-09-01T16:41:30.290 回答
1

这实际上看起来像 Socket.IO 在扩展方面取得了成功。您会期望来自一台服务器的消息发送到该房间中的所有套接字，无论它们碰巧连接到哪台服务器。

最好的办法是拥有一个每秒发送一条消息的主进程。例如，您可以通过仅运行它来做到这一点cluster.isMaster。

于 2013-09-01T02:15:27.807 回答
0

进程间通信不足以使 socket.io 1.4.5 与集群一起工作。强制 websocket 模式也是必须的。请参阅Node.JS、Socket.IO 和集群中的 WebSocket 握手不起作用

于 2016-04-09T04:57:48.127 回答