Node.js、Socket.io、Redis pub/sub 高并发低延迟问题

Question

当将Socket.io/Node.js和Redis pub/sub连接起来，试图创建一个能够处理多种传输方式的基于服务器事件驱动的实时Web广播系统时，有三种方法：

1. 创建redis连接并订阅频道。在socket.io客户端连接时，将客户端加入到socket.io房间中。在redis.on("message",...)事件中，调用io.sockets.in(room).emit("event",data) 将消息分发给相关房间中的所有客户端。如：如何重用socket.io中的redis连接？
2. 创建redis连接。在socket.io客户端连接时，将客户端加入到socket.io房间并订阅相关的redis频道。在客户端连接闭包（closure）内部包含redis.on("message",...)，当接收到消息时，调用client.emit("event",data)在特定客户端上触发事件。如：使用RedisStore在socket.io中的示例
3. 使用内置在socket.io中的RedisStore，并从Redis中的单个“dispatch”频道进行“广播”，遵循socketio-spec协议。

第1种方法可以为所有客户端处理Redis sub及其相关事件。第2种方法提供了更直接的连接到Redis pub/sub的方法。第3种方法更简单，但对消息事件的控制较少。

然而，在我的测试中，所有方法与超过1个连接的客户端表现出意外的低性能。所涉及的服务器事件是尽可能快地发布了1000条消息到Redis频道，并尽可能快地分发这些消息。性能通过连接的客户端（基于socket.io-client），记录时间戳到Redis列表进行分析来测量。我推测的情况是，在选项1中，服务器收到消息后，顺序写入所有连接的客户端。在选项2中，服务器会多次收到每个消息（每个客户端订阅一次），然后将其写入相关的客户端。在任何一种情况下，服务器在把消息传递给所有连接的客户端之前不会进入第二个消息事件。这种情况随着并发性的增加而变得更加明显。

这似乎与该技术栈的能力认知相矛盾。我想相信，但我很困惑。

这种情况（低延迟分发大量消息）是否还不支持这些工具（尚未实现），或者我是错过了什么技巧？

- lebear

---

你尝试过使用0MQ/AMQ等消息服务器进行发布-订阅，是否获得了类似的性能？ - ali haider

祝你好运获得更多的答案，我很好奇专家们的意见会不会加入到这里。我的回答主要基于网上流传的文档。 - Mark Essel

嗨，lebear，我有完全相同的疑虑，在四处寻找后找到了这个链接。如果您已经找到了自己问题的答案，我也想听听。 - andyk

Answer 1

我认为这是一个合理的问题，并且之前花了一些时间进行了研究。我花了一点时间搜索了一些示例，您可能会从中获得一些有用的提示。

示例

我喜欢从简单明了的示例开始：

• 轻量级 IM 示例代码
• Node.js + Redis Pub/Sub + socket.io 演示

轻量级示例是一个单页应用（请注意，您需要将 redis-node-client 替换为 Matt Ranney 的 node_redis之类的东西）：

/*
 * Mclarens Bar: Redis based Instant Messaging
 * Nikhil Marathe - 22/04/2010

 * A simple example of an IM client implemented using
 * Redis PUB/SUB commands so that all the communication
 * is offloaded to Redis, and the node.js code only
 * handles command interpretation,presentation and subscribing.
 * 
 * Requires redis-node-client and a recent version of Redis
 *    http://code.google.com/p/redis
 *    http://github.com/fictorial/redis-node-client
 *
 * Start the server then telnet to port 8000
 * Register with NICK <nick>, use WHO to see others
 * Use TALKTO <nick> to initiate a chat. Send a message
 * using MSG <nick> <msg>. Note its important to do a
 * TALKTO so that both sides are listening. Use STOP <nick>
 * to stop talking to someone, and QUIT to exit.
 *
 * This code is in the public domain.
 */
var redis = require('./redis-node-client/lib/redis-client');

var sys = require('sys');
var net = require('net');

var server = net.createServer(function(stream) {
    var sub; // redis connection
    var pub;
    var registered = false;
    var nick = "";

    function channel(a,b) {
    return [a,b].sort().join(':');
    }

    function shareTable(other) {
    sys.debug(nick + ": Subscribing to "+channel(nick,other));
    sub.subscribeTo(channel(nick,other), function(channel, message) {
        var str = message.toString();
        var sender = str.slice(0, str.indexOf(':'));
        if( sender != nick )
        stream.write("[" + sender + "] " + str.substr(str.indexOf(':')+1) + "\n");
    });
    }

    function leaveTable(other) {
    sub.unsubscribeFrom(channel(nick,other), function(err) {
        stream.write("Stopped talking to " + other+ "\n");
    });
    }

    stream.addListener("connect", function() {
    sub = redis.createClient();
    pub = redis.createClient();
    });

    stream.addListener("data", function(data) {
    if( !registered ) {
        var msg = data.toString().match(/^NICK (\w*)/);
        if(msg) {
        stream.write("SERVER: Hi " + msg[1] + "\n");
        pub.sadd('mclarens:inside', msg[1], function(err) {
            if(err) {
            stream.end();
            }
            registered = true;
            nick = msg[1];
// server messages
            sub.subscribeTo( nick + ":info", function(nick, message) {
            var m = message.toString().split(' ');
            var cmd = m[0];
            var who = m[1];
            if( cmd == "start" ) {
                stream.write( who + " is now talking to you\n");
                shareTable(who);
            }
            else if( cmd == "stop" ) {
                stream.write( who + " stopped talking to you\n");
                leaveTable(who);
            }
            });
        });
        }
        else {
        stream.write("Please register with NICK <nickname>\n");
        }
        return;
    }

    var fragments = data.toString().replace('\r\n', '').split(' ');
    switch(fragments[0]) {
    case 'TALKTO':
        pub.publish(fragments[1]+":info", "start " + nick, function(a,b) {
        });
        shareTable(fragments[1]);
        break;
    case 'MSG':
        pub.publish(channel(nick, fragments[1]),
            nick + ':' +fragments.slice(2).join(' '),
              function(err, reply) {
              if(err) {
                  stream.write("ERROR!");
              }
              });
        break;
    case 'WHO':
        pub.smembers('mclarens:inside', function(err, users) {
        stream.write("Online:\n" + users.join('\n') + "\n");
        });
        break;
    case 'STOP':
        leaveTable(fragments[1]);
        pub.publish(fragments[1]+":info", "stop " + nick, function() {});
        break;
    case 'QUIT':
        stream.end();
        break;
    }
    });

    stream.addListener("end", function() {
    pub.publish(nick, nick + " is offline");
    pub.srem('mclarens:inside', nick, function(err) {
        if(err) {
        sys.debug("Could not remove client");
        }
    });
    });
});

server.listen(8000, "localhost");

文档

有大量的文档可用，而且这些堆栈的API正在快速更改，因此您需要权衡每个文档的时间相关性。

• Redis延迟
• Linkedin通用技巧
• Node Redis模块

相关问题

只有一些相关问题，这是Stack上的热门话题:

• Node.js中的Redis pub/sub聊天服务器
• 如何为即时消息系统设计Redis pub/sub？

值得注意的提示（效果因人而异）

关闭或优化套接字池，使用高效的绑定，监视延迟，并确保您不会重复工作（即无需向所有侦听器两次发布）。

- Mark Essel