如何在RabbitMQ中设置多次重试？

Question

我正在使用RabbitMQ,我有一个包含电子邮件的队列.我的消费者服务将消息排队并尝试发送消息.如果由于任何原因,我的消费者无法发送消息,我想重新排队消息再次发送.我意识到我可以做一个basicNack并将requeue标志设置为true,但是,我不想无限期地重新排列该消息(例如,如果我们的电子邮件系统出现故障,我不想连续重新排队未发送的消息).我想定义有限次数,我可以重新排列要再次发送的消息.但是,当我将它出列并发送一个nack时,我无法在电子邮件消息对象上设置字段.更新的字段不存在于队列中的消息上.还有其他方法可以解决这个问题吗？提前致谢.

Answer 1

在RabbitMQ(以及AMQP协议)中没有像重试尝试这样的功能.

实现重试尝试的可能解决方案限制行为:

1. Redeliver消息,如果它以前没有重新传递(检查方法的redelivered参数basic.deliver- 你的库应该有一些接口)并删除它然后捕获死信交换,然后以某种方式处理.

2. 每次无法处理消息时再次发布它,但设置或增加/减少标题字段,比如说x-redelivered-count(你可以选择任何你喜欢的名字).要在这种情况下控制重新传输,您必须检查您设置的字段是否达到某个限制(顶部或底部 - 0是我的选择,ttl在tcp/ip的ip标头中是a-la ).

3. 在Redis,memcache或其他存储中存储消息唯一键(比如uuid,但是你必须在发布消息时手动设置它),甚至在mysql旁边还有重新计数,然后在每次重新传递时增加/减少此值,直到达到限制.

4. (对于真正的极客)编写插件,将实现您想要的这种行为.

#3的亲是重新传递的消息留在队列头部.如果你有很长的队列或者消息顺序对你很重要,那么这很重要(请注意,重新发送会破坏严格的消息顺序,有关详细信息或SO上的此问题,请参阅官方文档).

PS:

在这个主题中有类似的答案,但在PHP中.看看它,也许它可以帮助你一点(从单词"有多种技术来处理循环再生问题"开始阅读它.

Answer 2

虽然这是一个老问题，但我认为您现在可以通过死信交换和x-death标题数组的组合轻松地做到这一点，一旦消息被死信：

死信过程将一个数组添加到每个名为 x-death 的死信消息的标题中。此数组包含每个死字事件的条目，由一对 {queue, reason} 标识。每个这样的条目都是一个由几个字段组成的表：

queue：消息在死信之前所在队列的名称

原因：死字的原因，见下文

时间：消息被死信的日期和时间，为 64 位 AMQP 0-9-1 时间戳

交换 - 消息发布到的交换（请注意，如果消息多次死信，这将是死信交换）

路由密钥：消息发布时使用的路由密钥（包括 CC 密钥，但不包括 BCC 密钥）

计数：由于这个原因，这条消息在这个队列中被死信多少次

original-expiration（如果消息由于每条消息的 TTL 是死信的）：消息的原始到期属性。过期属性从死信消息中删除，以防止它在路由到的任何队列中再次过期。

阅读这篇很棒的文章了解更多信息

检查这个抽奖：

Answer 3

您应该使用Quorum 队列类型。该队列类型有一个传递限制参数，用于指定删除消息之前传递消息的重试次数。

创建队列时指定下一个参数：

x-queue-type: quorum
x-delivery-limit: 3 // it means rabbitmq will make 3 attempts to deliver a message before deleting it