Node.js、Socket.io、Redis pub/sub 大容量、低延迟困难

Posted 2023-02-16

【中文标题】Node.js、Socket.io、Redis pub/sub 大容量、低延迟困难

【英文标题】：Node.js, Socket.io, Redis pub/sub high volume, low latency difficulties

【发布时间】：2012-05-20 10:22:00

【问题描述】：

当结合 socket.io/node.js 和 redis pub/sub 以尝试创建一个由可以处理多个传输的服务器事件驱动的实时网络广播系统时，似乎有三种方法：

'createClient' 一个 redis 连接并订阅频道。在 socket.io 客户端连接上，将客户端加入到 socket.io 房间。在 redis.on("message", ...) 事件中，调用 io.sockets.in(room).emit("event", data) 分发给相关房间的所有客户端。赞How to reuse redis connection in socket.io?

'createClient' 一个 redis 连接。在 socket.io 客户端连接上，将客户端加入到 socket.io 房间并订阅相关的 redis 频道。在客户端连接关闭中包含 redis.on("message", ...) 并在收到消息调用 client.emit("event", data) 以在特定客户端上引发事件。喜欢Examples in using RedisStore in socket.io中的答案

使用嵌入到 socket.io 中的 RedisStore，并按照 socketio-spec 协议从 Redis 中的单个“调度”通道“广播”。

数字 1 允许为所有客户端处理一次 Redis 子和关联事件。 2 号提供了一个更直接的挂钩到 Redis 发布/订阅。数字 3 更简单，但对消息传递事件几乎没有控制。

但是，在我的测试中，所有连接的客户端都表现出出乎意料的低性能。有问题的服务器事件是尽快发布到 redis 通道的 1,000 条消息，以便尽快分发。性能是通过连接客户端的时间来衡量的（socket.io-client 基于将时间戳记录到 Redis 列表中以供分析）。

我的猜测是，在选项 1 中，服务器接收到消息，然后按顺序将其写入所有连接的客户端。在选项 2 中，服务器多次接收每条消息（每个客户端订阅一次）并将其写入相关客户端。在任何一种情况下，服务器都不会到达第二个消息事件，直到它与所有连接的客户端通信。随着并发性的增加，情况明显恶化。

这似乎与堆栈功能的感知智慧不一致。我想相信，但我很挣扎。

这种情况（大量消息的低延迟分布）是不是这些工具的选项（还没有？），还是我错过了一个技巧？

【问题讨论】：

您是否尝试过使用 0MQ/AMQ 等消息服务器的 pub-sub 并获得类似的性能？

祝您获得更多答案好运，我很想看到专家意见。我的答案主要基于网络上流传的文档。

嗨 lebear，我也有同样的担忧，环顾四周后找到了这个链接。如果您找到了自己问题的答案，我也想听听。

【参考方案1】：

我认为这是一个合理的问题，并在不久前对其进行了简短的研究。我花了一些时间搜索示例，您可以从中获得一些有用的提示。

示例

我喜欢从简单的例子开始：

light im sample code Node.js + Redis Pub/Sub + socket.io demo

轻样本是一个单页（请注意，您需要将 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 正在迅速变化，因此您必须权衡每个文档的时间相关性。

node activity streams cloud foundry example how to node redis pubsub redis latency redis cookbook Using Pub/Sub for Asynchronous Communication linkedin's generic tips node redis bindings google groups nodejs question

相关问题

只是几个相关的问题，这是堆栈上的热门话题：

Redis pub/sub for chat server in node.js How to design redis pub/sub for an instant messaging system?

值得注意的提示 (ymmv)

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