如何使用Redis / Node / Rails来延迟和合并pub / sub

我有一个RubyOnRails应用程序,它使用Node.js / Socket.io服务器将交易信息推送到所有连接的客户端。 无论何时执行交易,客户端屏幕都会根据上次交易的信息进行更新。

随着交易频率的增加,更新每秒更新或者更频繁地变得相当烦人。 我正在寻找一种方法,例如推送更新到客户端只有最大。 每5秒一次,即如果没有交易发生,则什么都不推。

到目前为止,我所拥有的是:我通过以下方式将交易信息从Rails应用程序推送到Redis:

REDIS.publish('tradeupdate', ..... ) 

和节点服务器做了这样的事情:

 cli_sub.subscribe("tradeupdate"); cli_sub.on("message",function(channel,message) { io.sockets.emit('ablv', message); }); 

然后客户做

 socket.on('ablv', function (data) { obj = JSON.parse(data); ..... }); 

其目的是在给定的时间段内(例如5秒)只将最后一条消息从Rails发送到Node或从Node发送到客户端。

       

网上收集的解决方案 "如何使用Redis / Node / Rails来延迟和合并pub / sub"

它看起来像你在这里需要的是帮助你的函数调用。 例如:

 var makeThrottler = function() { var toRun = null, timeout = null; function doRun() { if (!toRun) { // nothing to run; we set timeout to null so that the // next function to execute knows to run immediately timeout = null; return; } // set a timeout of 5s before timeout = setTimeout(function() { doRun(); }, 5000); // we need to do this temp thing to protect against // calling executeThrottled again within toRun var temp = toRun; toRun = null; temp(); } function executeThrottled(fn) { // this is the function we want to execute next; it // overwrites any function we've stored earlier toRun = fn; // if we already ran within the last 5 seconds, don't do // anything now (our function will be called later) if (timeout) return; // execute the function right away doRun(); } return executeThrottled; } 

这里有一个如何使用它的例子:

 var throttled = makeThrottler(), x = 0; function increment() { throttled(function() { console.log(x); }); x++; setTimeout(increment, 1000); } increment(); 

增量函数将x每秒增加一次。 logging被扼杀,所以你会看到的输出是0,5,10等(他们可能偶尔会因为一些时间不准确而被closures)。

你原来的代码会变成这样:

 cli_sub.subscribe("tradeupdate"); cli_sub.on("message",function(channel,message) { throttled(function() { io.sockets.emit('ablv', message); }); }); 

什么阻止你缓冲消息,并使用一个简单的计时器来执行每五秒发射一次?

 var last_message = null; cli_sub.subscribe("tradeupdate"); cli_sub.on("message",function(channel,message) { last_message = message; }); setInterval(function() { io.sockets.emit('ablv', last_message); }, 5000);