如何在 RabbitMQ 中重新排队消息

Question

consumer 收到消息后，consumer/worker 进行一些验证，然后调用 web 服务。在这个阶段，如果发生任何错误或验证失败，我们希望将消息放回最初使用它的队列。

我已阅读 RabbitMQ 文档。但我对拒绝、nack 和取消方法之间的区别感到困惑。

Answer 1

简答：

要重新排队特定消息，您可以同时选择 basic.reject 或 basic.nack 并将 multiple 标志设置为 false。

basic.consume 调用也可能导致消息重新传递，如果您正在使用消息确认，并且在特定时间有未确认的消息在消费者身上并且消费者在没有确认的情况下退出。

basic.recover 将重新发送特定频道上所有未确认的消息。

长答案：

basic.reject 和 basic.nack 都用于相同的目的 - 丢弃或重新排队特定消费者无法处理的消息（在给定时刻，在某些条件下或根本不处理）。它们之间的主要区别在于basic.nack 支持批量消息处理，而basic.reject 不支持。

RabbitMQ 官方网站Negative Acknowledgements 文章中描述了这种差异：

AMQP 规范定义了basic.reject 方法，该方法允许客户端拒绝单个已传递的消息，指示代理要么丢弃它们，要么将它们重新排队。不幸的是，basic.reject 不支持批量否定确认消息。

为了解决这个问题，RabbitMQ 支持basic.nack 方法，该方法提供basic.reject 的所有功能，同时还允许批量处理消息。

要批量拒绝邮件，客户端将basic.nack 方法的multiple 标志设置为true。然后，代理将拒绝所有未确认的已传递消息，直到并包括在 basic.nack 方法的 delivery_tag 字段中指定的消息。在这方面，basic.nack 补充了basic.ack 的批量确认语义。

注意，basic.nack 方法是 RabbitMQ 特定的扩展，而 basic.reject 方法是 AMQP 0.9.1 规范的一部分。

basic.cancel 方法用于通知服务器客户端停止消息消费。请注意，客户端可能会在basic.cancel 方法发送接收cancel-ok 回复之间接收任意消息编号。如果客户端使用消息确认并且它有任何未确认的消息，它们将被移回它们最初被消费的队列。

basic.recover 在 RabbitMQ 中有一些限制：它 - basic.recover with requeue=false - basic.recover synchronicity

除了勘误表之外，according to RabbitMQ specs basic.recover 也有部分支持（不支持使用 requeue=false 进行恢复。）

注意basic.consume：

当basic.consume 开始时没有自动确认 (no­ack=false) 并且有一些未确认的消息未确认消息，然后当消费者被取消（死亡、致命错误、异常等）时，将重新传递未决消息。从技术上讲，在消费者释放它们（ack/nack/reject/recover）之前，不会处理待处理的消息（甚至是死信）。只有在那之后，它们才会被处理（例如，死信）。

例如，假设我们最初连续发布 5 条消息：

Queue(main) (tail) { [4] [3] [2] [1] [0] } (head)

然后消费其中3个，但不确认，然后取消消费者。我们会有这样的情况：

Queue(main) (tail) { [4] [3] [2*] [1*] [0*] } (head)

星号 (*) 注意到 redelivered 标志设置为 true。

假设我们有死信交换集和死信消息队列

Exchange(e-main)                                   Exchange(e-dead) 
  Queue(main){x-dead-letter-exchange: "e-dead"}       Queue(dead) 

假设我们发布了 5 条消息，其中 expire 属性设置为 5000（5 秒）：

Queue(main) (tail) { [4] [3] [2] [1] [0] } (head)
Queue(dead) (tail) { }(head)

然后我们从main队列中消费3条消息并保持10秒：

Queue(main) (tail) { [2!] [1!] [0!] } (head)
Queue(dead) (tail) { [4*] [3*] } (head)

感叹号 (!) 代表未确认的消息。此类消息无法传递给任何消费者，并且通常无法在管理面板中查看。但是让我们取消消费者，记住，它仍然持有 3 条未确认的消息：

Queue(main) (tail) { } (head)
Queue(dead) (tail) { [2*] [1*] [0*] [4*] [3*] } (head)

所以现在头部中的 3 条消息被放回原始队列，但是由于它们设置了每个消息的 TTL，它们被死信到死信队列的尾部（当然，通过死信交换)。

附注：

消费消息又名监听新消息在某种程度上不同于直接队列访问（获取一条或多条消息而不关心其他消息）。更多内容见basic.get方法说明。