Node.js rsmq - 在 Node.js 应用程序重新启动之前，新消息不可见

Question

我正在努力使这个包工作。

redis 版本：稳定 4.0.6

我像这样连接Redis，那里没有问题。

发布订阅.js

var redis = require("redis");
var psRedis = redis.createClient();

psRedis.auth("mypasswordishere", function (callback) {
  console.log("connected");
});

module.exports.psRedis = psRedis;

启动 Node.js 应用程序后，我可以在控制台上看到“已连接”并执行操作，我已经检查过了。

我的test.js文件在下面。

测试.js

var express = require('express');
var router = express.Router();
var path = require("path");
var bodyParser = require('body-parser');
var async1 = require("async");
var client = require("../databases/redis/redis.js").client;

var RedisSMQ = require("rsmq");

var psRedis = require("./realtime/operations/pubsub").psRedis;

var rsmq = new RedisSMQ({client: psRedis});

rsmq.createQueue({qname: "myqueue"}, function (err, resp) {
  if (resp === 1) {
    console.log("queue created");
  }
});

rsmq.receiveMessage({qname: "myqueue"}, function (err, resp) {
  if (resp) {
    console.log(resp);
  }
});

router.get('/pubsubTest', function (req, res, next) {

  async1.waterfall([
    function (callback) {

      rsmq.sendMessage({qname: "myqueue", message: "Hello World 1"}, function (err, resp) {
        if (resp) {
          console.log("Message sent. ID:", resp);
        }
      });

      callback(null, 'done!');

    }
  ], function (err, result) {
    res.sendStatus(200);
  });

});

module.exports = router;

但是，当我访问时/pubsubTest，控制台上只显示消息 ID。

消息已发送。编号：exb289xu0i7IaQPEy1wA4O7xQQ6n0Cap

如果我重新启动我的 Node.js 应用程序，我会看到下面的结果，这是预期的。为什么不立即出现？

{ id: 'exb289xu0i7IaQPEy1wA4O7xQQ6n0CAp',
  message: 'Hello World 1',
  rc: 1,
  fr: 1515802884138,
  sent: 1515802880098 }

谢谢你。

Answer 1

receiveMessage不会“火”。您需要在发送消息后调用它。您正在寻找的是rsmq 提供的实时选项。

var rsmq = new RedisSMQ({client: psRedis}, ns: "rsmq",realtime :true});

现在，在通过 将每条新消息添加到队列时sendMessage，都会将 PUBLISH 消息rsmq:rt:{qname}与内容一起发送到{msg}。在你的情况下sendMessage会发出一个事件，即rsmq:rt:myqueue

对此可以有两种解决方案，都将使用该事件 rsmq:rt:myqueue

1.第一个会使用redis客户端，它可以subscribe通过redis提供的方法订阅这个发布的事件来实现PUB/SUB。

 var redis = require('redis');
    const subscribe = redis.createClient();
    subscribe.subscribe('rsmq:rt:myqueue');
    subscribe.on('message', function(msg) {     //msg=>'rsmq:rt:myqueue'
        rsmq.receiveMessage({qname: "myqueue"}, function (err, resp) {
            if (resp) {
                console.log(resp);
            }
        });
    });

整个代码看起来像：

var express = require('express');
var router = express.Router();
var path = require("path");
var bodyParser = require('body-parser');
var async1 = require("async");
var client = require("../databases/redis/redis.js").client;
var psRedis = require("./realtime/operations/pubsub").psRedis;
var rsmq = new RedisSMQ({client: psRedis}, ns: "rsmq",realtime :true});

rsmq.createQueue({qname: "myqueue"}, function (err, resp) {
  if (resp === 1) {
    console.log("queue created");
  }
});

const subscribe = redis.createClient( 6379,"127.0.0.1"); //creating new 
worker and pass your credentials
subscribe.subscribe('rsmq:rt:myqueue');
subscribe.on('message', function(msg) {     //msg=>'rsmq:rt:myqueue'
    rsmq.receiveMessage({qname: "myqueue"}, function (err, resp) {
        if (resp) {
            console.log(resp);
        }
    });
});

router.get('/pubsubTest', function (req, res, next) {
  async1.waterfall([
    function (callback) {
      rsmq.sendMessage({qname: "myqueue", message: "Hello World 1"}, 
function (err, 
      resp) {
        if (resp) {
          console.log("Message sent. ID:", resp);
        }});
      callback(null, 'done!');
    }
  ], function (err, result) {
    res.sendStatus(200);
  });
});
module.exports = router;

2.第二个解决方案是使用rsmq-worker它会为您提供一个消息事件，您可以使用.on方法收听该事件。

var RSMQWorker = require( "rsmq-worker" );
var worker = new RSMQWorker( "myqueue" ,{interval:.1}); // this worker 
will poll the queue every .1 second.

worker.on( "message", function( message, next, msgid ){
     if(message){
         console.log(message);
     }
    next();
 });
worker.start();

整个代码看起来像：

var express = require('express');
var router = express.Router();
var path = require("path");
var bodyParser = require('body-parser');
var async1 = require("async");
var client = require("../databases/redis/redis.js").client;
var psRedis = require("./realtime/operations/pubsub").psRedis;
var rsmq = new RedisSMQ({client: psRedis},{ ns: "rsmq",realtime :true});

rsmq.createQueue({qname: "myqueue"}, function (err, resp) {
    if (resp === 1) {
        console.log("queue created");
    }
});

var RSMQWorker = require( "rsmq-worker" );
var worker = new RSMQWorker( "myqueue" ,{interval:.1});
worker.on( "message", function( message, next, msgid ){
        if(message){
            console.log(message);
        }
    next();
});


// optional error listeners
worker.on('error', function( err, msg ){
    console.log( "ERROR", err, msg.id );
});
worker.on('exceeded', function( msg ){
console.log( "EXCEEDED", msg.id );
});
worker.on('timeout', function( msg ){
    console.log( "TIMEOUT", msg.id, msg.rc );
});
worker.start();


router.get('/pubsubTest', function (req, res, next) {
    async1.waterfall([
        function (callback) {
            rsmq.sendMessage({qname: "myqueue", message: "Hello World1"}
            ,function (err, resp) {
                if (resp) {
                    console.log("Message sent. ID:", resp);
                }});
        callback(null, 'done!');
        }
    ], function (err, result) {
    res.sendStatus(200);
    });
});
module.exports = router;

注意：在第一个解决方案中，您需要使用删除从队列中收到的消息，deleteMessage或者您也可以使用popMessage它将接收最后一条消息并为您删除它。如果您不删除该消息，您将再次获得所有消息，直到该特定消息的超时结束。

出于这个原因，我更喜欢使用第二种解决方案，rsmq 会为您处理这些事情，您也可以提供自己的轮询间隔