启用持久传递模式后，强制工作人员立即接收失败的消息

Question

我有一个RabbitMQ Server 设置，我使用Python-Pika 获取消息。问题是，如果我启用了persistent delivery mode，并且工作人员无法处理消息。它不会释放消息，而是一直保留到消息，直到 RabbitMQ 连接被重置。

有没有办法确保无法处理的消息在合理的时间范围内从可用的工作人员（包括同一个工作人员）中再次被拾取？

这是我当前的代码

if success:
    ch.basic_ack(delivery_tag=method.delivery_tag)
else:
    syslog.syslog('Error (Callback) -- Failed to process payload: %s' % body)

我的想法是我永远不想丢失一条消息，而是希望它被重新发布，或者如果它失败了，我希望它再次被拾起。在工作人员成功处理消息之前，应该始终如此。这通常发生在其中一名工作人员无法打开与 HTTP 服务器的连接时。

Answer 1

我终于明白为什么会这样了。我没有意识到当您完成消息时仅仅确认是不够的，而且还必须拒绝使用channel.basic_reject 无法处理的任何消息。这似乎很明显，但这不是AMQP 的默认行为。

基本上我们必须使用basic_reject 发布消息，并将requeue 设置为True。这里的重要因素是 requeue 关键字，它可以防止消息被丢弃，而是将其再次排队，以便我们可用的工作人员之一可以处理它。

if success:
    # On Success - Mark message as processed.
    ch.basic_ack(delivery_tag=method.delivery_tag)
else:
    # Else - Mark message as rejected and move it back to the queue.
    ch.basic_reject(delivery_tag=method.delivery_tag, requeue=True)

我在this 文章中发现了一些非常有用的信息，reject 关键字在this 博客文章中有更多技术细节。