Node.js, Socket.io, Redis pub/sub alto volumen, baja latencia dificultades

publish-subscribe low-latency

Question

Cuando se combinan socket.io/node.js y redis pub / sub en un intento de crear un sistema de transmisión web en tiempo real impulsado por eventos del servidor que pueden manejar múltiples transportes, parece haber tres enfoques:

1. ''createClient'' una conexión redis y suscribirse a los canales. En la conexión de cliente socket.io, únete al cliente en una habitación socket.io. En el evento redis.on ("message", ...), llame a io.sockets.in (room) .emit ("event", data) para distribuirlo a todos los clientes en la sala correspondiente. ¿Cómo reutilizar la conexión de redis en socket.io?

2. ''createClient'' una conexión redis. En la conexión del cliente de socket.io, únase al cliente en una sala socket.io y suscríbase a los canales redis relevantes. Incluya redis.on ("mensaje", ...) dentro del cierre de la conexión del cliente y al recibir la llamada del mensaje client.emit ("evento", datos) para generar el evento en el cliente específico. Me gusta la respuesta en Ejemplos al usar RedisStore en socket.io

3. Utilice RedisStore horneado en socket.io y ''broadcast'' desde el único canal de "despacho" en Redis siguiendo el protocolo de especificación socketio.

El número 1 permite manejar el sub Redis y el evento asociado una vez para todos los clientes. El número 2 ofrece un gancho más directo en el pub / sub de Redis. El número 3 es más simple, pero ofrece poco control sobre los eventos de mensajería.

Sin embargo, en mis pruebas, todas exhiben un rendimiento inesperadamente bajo con más de 1 cliente conectado. Los eventos del servidor en cuestión son 1,000 mensajes publicados en un canal redis lo más rápido posible, para ser distribuidos lo más rápido posible. El rendimiento se mide por tiempos en los clientes conectados (basado en socket.io-client que registra las marcas de tiempo en una lista de Redis para el análisis).

Lo que supongo es que en la opción 1, el servidor recibe el mensaje, luego lo escribe secuencialmente a todos los clientes conectados. En la opción 2, el servidor recibe cada mensaje varias veces (una vez por suscripción de cliente) y lo escribe en el cliente correspondiente. En cualquier caso, el servidor no llega al segundo evento de mensaje hasta que se comunique a todos los clientes conectados. Una situación claramente exacerbada con el aumento de concurrencia.

Esto parece en desacuerdo con la sabiduría percibida de las capacidades de las pilas. Quiero creer, pero estoy luchando.

¿Este escenario (distribución de baja latencia de alto volumen de mensajes) simplemente no es una opción con estas herramientas (¿todavía?), O ¿me falta un truco?

Answer 1

Pensé que esta era una pregunta razonable y la había investigado brevemente hace un tiempo. Dediqué un poco de tiempo a buscar ejemplos de los que puedas obtener algunos consejos útiles.

Ejemplos

Me gustaría comenzar con ejemplos sencillos:

• código de muestra im luz
• Node.js + Redis Pub / Sub + socket.io demo

La muestra de luz es una sola página (tenga en cuenta que querrá reemplazar redis-node-client con algo así como node_redis por 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,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");

Documentos

Hay un montón de documentación por ahí, y las API están cambiando rápidamente en este tipo de pila, así que tendrás que sopesar la relevancia temporal de cada documento.

• corrientes de actividad de nodo
• ejemplo de fundición en la nube
• cómo nodo redis pubsub
• latencia de redis
• libro de cocina redis utilizando Pub / Sub para comunicación asincrónica
• consejos genéricos de LinkedIn
• node_redis
• Pregunta de google groups nodejs

preguntas relacionadas

Solo unas pocas preguntas relacionadas, este es un tema candente en la pila:

• Redis pub / sub para el servidor de chat en node.js
• ¿Cómo diseñar redis pub / sub para un sistema de mensajería instantánea?

Consejos notables (ymmv)

Desactive u optimice la agrupación de sockets, utilice enlaces eficientes, supervise la latencia y asegúrese de no duplicar el trabajo (es decir, no es necesario publicarlo para todos los oyentes dos veces).