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

Question

我正在尝试使 this 包工作。

redis 版本：稳定 4.0.6

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

pubsub.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 文件在下面。

test.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 添加到队列的每条新消息，都会向rsmq:rt:{qname} 发送一条内容为{msg} 的PUBLISH 消息。在您的情况下，sendMessage 将发出一个事件，即 rsmq:rt:myqueue

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

1.第一个使用redis客户端，可以通过redis提供的subscribe方法订阅这个发布的事件，实现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 会为您处理这些事情，您也可以提供自己的轮询间隔