nodejs推送订阅redis的通知

Question

对于仅登录用户,我想以某种方式通知他们,如果他们有任何新的通知.

例如,假设一个成员向他们发送了一条私人消息,我想告诉用户他们有一条要查看的新消息(假设他们没有刷新页面).

使用Nodejs和redis,我将如何做到这一点？

注意:我只需要nodejs向用户发送一个小json,说他们有一条新消息.

我正在考虑的工作流程如下:

    1. user is logged in, a new message is sent to them.
    2. somehow using nodejs and redis and long-polling, nodejs communicates back to the logged in users browser they have a pending message.
    3. when nodejs sends this push notification, I then call another javascript function that will call a rest service to pull down additional json with the message.

我正在将nodejs集成到现有的应用程序中,所以我希望尽可能简单地使用nodejs来负责通知而不做任何额外的逻辑.

有人可以概述一下我应该如何解决这个问题吗？我应该以某种方式使用redis http://redis.io/topics/pubsub吗？
即使阅读了有关它的页面,我也不确定它是如何工作的.

Answer 1

如果要将node服务集成到现有应用程序中,我宁愿使用某种消息传递系统来传递来自该应用程序的消息,node而不是数据库,甚至是内存数据库.为清楚起见,我假设您可以使用rabbitmq.如果你确实需要使用redis,你只需要找到一种方法来使用它的发布而不是rabbitmq发布和相应的节点端订阅,但我认为整体解决方案是相同的.

您需要以下模块:

• rabbitmq服务器(安装复杂性大致相同redis)
• rabbitmq 在您的外部应用程序库中发送消息,支持大多数语言
• rabit.js用于node订阅消息或与外部应用程序通信的模块
• socket.io用于node在node服务器和客户端之间建立实时连接的模块

我还假设您的外部应用程序和node服务器都可以访问某些共享数据库(可以是redis),其中node存储了客户端会话信息(例如,redis-session-store用于node).这将允许用于sessionId验证消息的用户,会话中的用户是否已登录以及是否需要向某些用户发送通知(通过外部应用程序).

这就是你的堆栈的样子(未抛光):

定义发布者node以通知您的外部应用程序它需要开始/停止发送给定的消息sessionId.我将假设,对于给定sessionId的用户信息可以node从共享数据库的任一侧(或外部应用程序)恢复,并且可以验证用户(这里为了简单起见,通过检查session.authenticated_user).还定义订阅者以侦听用户的传入消息:

var context = require('rabbit.js').createContext();

var pub = context.socket('PUB');
var sub = context.socket('SUB');

定义socket.io从node服务器到客户端的连接.一旦客户的网页被(重新)加载并被io.connect()调用,下面的代码将被执行(见答案末尾的clinet方面).在建立新连接时,验证用户是否已登录(意味着其凭据在会话中),注册套接字处理程序并向外部应用程序发布通知以开始为此发送消息sessionId.这里的代码假设登录/注销时页面重新加载(因此新socket.io会话).如果不是这种情况,只需socket.io从客户端发出相应的消息,node并在下面的方法中注册处理程序,方法与新连接相同(这超出了本示例的范围):

var sessionStore = undefined; // out-of-scope: define redis-session-store or any other store

var cookie = require("cookie"),
    parseSignedCookie = require('connect').utils.parseSignedCookie;

// will store a map of all active sessionIds to sockets
var sockets = {};

// bind socket.io to the node http server
var io = require('socket.io').listen(httpServer);

// assumes some config object with session secrect and cookie sid
io.sockets.on("connection", function(socket) {
    if (socket.handshake.headers.cookie) {
        var cks = cookie.parse(socket.handshake.headers.cookie);
        var sessionId = parseSignedCookie(cks[config.connectSid], config.sessionSecret);

        // retrieve session from session store for sessionId
        sessionStore.get(sessionId, function(err, session) {
            // check if user of this session is logged in, 
            // define your elaborate method here
            if (!err && session.authenticated_user) { 
                // define cleanup first for the case when user leaves the page
                socket.on("disconnect", function() {
                    delete sockets[sessionId];
                    // notify external app that it should STOP publishing
                    pub.connect('user_exchange', function() {
                        pub.write(JSON.stringify({sessionId: sessionId, action: 'stop', reason: 'user disconnected'}), 'utf8');
                    });
                });

                // store client-specific socket for emits to the client
                sockets[sessionId] = socket;

                // notify external app that it should START publishing
                pub.connect('user_exchange', function() {
                    pub.write(JSON.stringify({sessionId: sessionId, action: 'start'}), 'utf8');
                });
            }
        });
    }
});

将订户连接到rabbitmq交换机以捕获消息并将其发送给客户端:

sub.connect('messages_exchange', function() {
    sub.on("readable", function() {
        // parse incoming message, we need at least sessionId
        var data = JSON.parse(sub.read());
        // get socket to emit for this sessionId
        var socket = sockets[data.sessionId];
        if (socket) {
            socket.emit("message", data.message);
        } else {
            // notify external app that it should STOP publishing
            pub.connect('user_exchange', function() {
                pub.write(JSON.stringify({sessionId: sessionId, action: 'stop', reason: 'user disconnected'}), 'utf8');
            });
            // further error handling if no socket found
        }
    });
});

最后你的客户端看起来大致如此(这里是Jade,但那只是因为我已经在这些行中有了这整个堆栈):

script(src="http://ajax.googleapis.com/ajax/libs/jquery/1.7.2/jquery.min.js")
script(src="/socket.io/socket.io.js")

script(type='text/javascript').
    $(function(){
        var iosocket = io.connect();
        iosocket.on('connect', function () {
            // do whatever you like on connect (re-loading the page)

            iosocket.on('message', function(message) {
                // this is where your client finally gets the message
                // do whatever you like with your new message
            });
        });

        // if you want to communicate back to node, e.g. that user was logged in,
        // do it roughly like this
        $('#btnSend').click(function(event) {
            iosocket.send('a message back to the node server if you need one');
        });
    });