Node.js 应用监听消息队列，异步向redis添加消息答案

【问题标题】：Node.js application listens to message queue and add messages to redis asynchronouslyNode.js 应用监听消息队列，异步向redis添加消息
【发布时间】：2017-07-02 11:46:05
【问题描述】：

我正在开发一个node.js组件，它监听一个消息队列（ActiveMQ），并将接收到的消息分批添加到redis中（每批必须20条）。

从 ActiveMQ 接收的消息数为每秒 10 条或更少时没有问题。

我的问题是每 4 毫秒将消息添加到队列中。这导致添加到批次的记录数有时每批次超过 20 条。

const stompit = require('stompit');
var uuid = require('node-uuid');
var Redis = require('ioredis');

var redis = new Redis();
var pipeline = redis.pipeline();
var batchCounter = 0;

stompit.connect({ host: 'localhost', port: 61613 }, function(err1, client) {

client.subscribe({ destination: 'MyQueue' }, function(err2, msg) {
    msg.readString('UTF-8', function(err3, body) {

        if (batchCounter >= 20){
            pipeline.exec(function(err4, results) {
                pipeline = redis.pipeline();
                batchCounter = 0;
                console.log(results);
            });
        }

        batchCounter++;
        pipeline.set(uuid.v1(), JSON.stringify(body));
        //client.disconnect();
      });
  });
  });

如何解决这个问题？谢谢

【问题讨论】：

标签： node.js redis activemq

【解决方案1】：

在调用.exec 方法之前尝试重置管道，我认为这是一个异步方法。因为.exec 在未来某个时间点运行，增量和pipeline.set 可以在它之前运行。

以下代码保存当前管道，并在.exec之前同步创建一个新管道

if (batchCounter >= 20){
    let fullpipeline = pipeline;
    pipeline = redis.pipeline();
    batchCounter = 0; 
    fullpipeline.exec(function(err4, results) {
        console.log(err4, results);
    });
}

那么新消息应该只附加到新管道中。

【讨论】：

您仍然有可能一次将多个管道连接到 redis。如果这也会导致问题，您可能需要一个完整的管道队列来处理。

【解决方案2】：

我最终使用标志控制了每批的记录数。我相信还有另一种可能的，可能更有效的解决方法是控制从 ActiveMQ 读取的过程。但是，在这种情况下，flags 为我完成了这项工作。使用标志的完整代码在以下链接中：

https://github.com/TamerB/ToActiveMQToRedis/blob/master/consumer/app.js

【讨论】：