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

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

>’createClient’a 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的答案
>根据socketio-spec协议，使用RedisStore烘焙到socket.io和redis中的单个“dispatch”通道的’broadcast’。

数字1允许为所有客户端处理一次Redis子和相关事件。数字2提供了一个更直接的钩到Redis pub / sub。数字3更简单，但对消息事件几乎没有控制。

然而，在我的测试中，所有表现出意想不到的低性能与超过1个连接的客户端。所讨论的服务器事件是尽可能快地发布到redis信道的1,000个消息，以尽可能快地分发。性能通过连接的客户端(socket.io-client，基于日志时间戳记到Redis列表进行分析)中的计时来衡量。

我推测在选项1中，服务器接收消息，然后顺序地将其写入所有连接的客户端。在选项2中，服务器多次接收每个消息(每个客户端预订一次)，并将其写入相关的客户端。在任一种情况下，服务器不会得到第二个消息事件，直到它被传递到所有连接的客户端。随着并发性的提高，情况明显恶化。

这似乎与堆栈能力的感知智慧不一致。我想相信，但我在努力。

这种情况(大量消息的低延迟分布)不是这些工具的一个选项(还是？)，或者我缺少一个招数？

解决方法

我认为这是一个合理的问题，并一直在研究它一会儿。我花了一些时间搜索例子，你可以从中提取一些有用的提示。

例子

我喜欢从直接的例子开始：

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

光样是一个单页面(注意，你需要替换redis-node-client像node_redis由Matt Ranney：

/*
 * 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,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',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

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

解决方法

相关推荐