nodejs推送订阅redis的通知

对于login用户只,我想以某种方式通知他们,如果他们有任何例如新的通知。

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

使用Nodejs和Redis,我将如何去做这件事?

注意:我只需要nodejs发送一个小的json给用户说他们有一个新的消息。

工作stream程如下,我在想:

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与nodejs保持一致,只负责通知nodejs而不做任何额外的逻辑。

有人可以概述我应该怎么做呢? 我应该使用Redis的http://redis.io/topics/pubsub莫名其妙?
即使阅读了关于它的网页,我也不确定它是如何工作的。

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

您需要以下模块:

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

我还假定你的外部应用程序和你的node服务器都可以访问一些共享的数据库(可以是redis),其中存储了node客户机会话信息(例如, node redis-session-store )。 这将允许使用sessionIdvalidation消息的用途,会话中的用户是否已login,以及某些用户是否需要发送通知(通过外部应用程序)。

这是你的堆栈看起来像(未抛光):

node定义一个发布者来通知你的外部应用程序需要启动/停止给定sessionId消息。 我将假定对于给定的sessionId ,用户信息可以从共享数据库的任一侧( node或外部应用程序)恢复,并且可以validation用户(这里为了简单起见,通过检查session.authenticated_user )。 还要定义一个用户来收听用户的传入消息:

 var context = require('rabbit.js').createContext(); var pub = context.socket('PUB'); var sub = context.socket('SUB'); 

定义从node服务器到客户端的socket.io连接。 一旦客户端的网页被重新加载并调用了io.connect() ,下面的代码就会被执行(见答案结尾处的io.connect()端)。 当build立一个新连接时,validation用户是否login(意味着它的凭证在会话中),注册套接字处理程序并向外部应用程序发布一个通知,以开始为此sessionId发送消息。 这里的代码假设一个页面重新加载login/注销(并因此新的socket.io会话)。 如果不是这种情况,只需从客户端向node发送一个相应的socket.io消息,并在下面的方法中注册一个处理程序,就像新build连接一样(这不在本例的范围之内) :

 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 } }); }); 

最后你的客户会看起来像这样(这里在翡翠,但是这只是因为我已经有这整个堆栈沿着这些线):

 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, eg 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'); }); }); 

Flickr对于如何使用NodeJS和Redis创build高度可用且可扩展的推送通知系统也有非常好的解释。

http://code.flickr.net/2012/12/12/highly-available-real-time-notifications/