node.js - amqplib - 尽管调用了channel.ack(msg)，但消息仍未确认

Question

我正在编写一个模块，充当 amqplib 的包装器。动机是，我们已经设置了一个现有的且定义良好的交换/队列/绑定(bind)，我只想公开使用方法以允许使用传入数据。

为此，我的模块采用 callback(channel, msg) 形式的回调参数。在模块中，设置交换、队列和绑定(bind)后，我有以下内容

module.exports = function (options, callback) {

  /* connection, exchange and queue set up here */

  // consume messages from primary queue
  ok = ok.then(function() {
    var q = opts.pq;
    console.log('Subscribing to', q);
    return ch.consume(q, function (message) {
      callback(ch, message);
    });
  });

  return ok;

};

在回调中，我正在处理消息，如果成功，则调用 channel.ack(msg)。

一切运行正常，没有错误，但是 RabbitMQ 管理控制台将所有已处理的消息显示为“未确认”。如果我随后终止我的应用程序，未确认的消息将返回到队列中(谢天谢地)。

为什么我的消息没有被确认？我在 ack'ing 中做错了什么吗？我应该在调用 ack 时将 allUpTo 参数设置为 true 吗？

环境详细信息

node -v
v0.8.26

npm ls
...
amqplib@0.1.1

Answer 1

为了完整起见，用我发现的内容回答我自己的问题。这实际上是由于 node-amqp 之间默认设置的差异造成的。和 amqp.node我的消费者中的实现略有缺陷。

订阅队列时，node-amqp 默认使用 prefetch=1，这意味着任何时候只有 1 条消息在传输。在前一个得到确认之前，不会交付另一个。然而，amqp.node 默认为 prefetch=0，这意味着所有消息都会尽快发送给消费者，并且可以在消费者完成后的某个时刻单独进行确认每一个。

这是我在管理控制台中看到的，这引起了警报并导致了这个问题。对此的讨论可以在 this github issue 中看到。 .

ASIDE - 鉴于上述信息，我让我的消费者保持运行状态，并让它在准备好时确认消息。这给我的实现带来了一个单独的问题。考虑到我们传递消息的方式(即一次全部)，消息处理程序正在用新连接的请求轰炸数据库。结果，数据库服务器过载，最终消费者死亡。为了解决这个问题，我像以前一样切换到 prefetch=1，这意味着上面的 Consumer Promise 方法现在如下

module.exports = function (options, callback) {

  /* connection, exchange and queue set up here */

  // consume messages from primary queue
  ok = ok.then(function() {
    var q = opts.pq;
    console.log('Subscribing to', q);

    ch.prefetch(1);   // <-- only get 1 message at a time

    return ch.consume(q, function (message) {
      callback(ch, message);
    }, { noAck: false });
  });


  return ok;

};

如果有人遇到类似情况，请记住。