52

似乎我让 rabbitmq 服务器运行的时间越长，未确认的消息就越麻烦。我很想重新排队。实际上，似乎有一个 amqp 命令可以执行此操作，但它仅适用于您的连接正在使用的通道。我构建了一个小鼠兔脚本至少可以尝试一下，但是我要么遗漏了一些东西，要么无法以这种方式完成（rabbitmqctl 怎么样？）

import pika

credentials = pika.PlainCredentials('***', '***')
parameters = pika.ConnectionParameters(host='localhost',port=5672,\
    credentials=credentials, virtual_host='***')

def handle_delivery(body):
    """Called when we receive a message from RabbitMQ"""
    print body

def on_connected(connection):
    """Called when we are fully connected to RabbitMQ"""
    connection.channel(on_channel_open)    

def on_channel_open(new_channel):
    """Called when our channel has opened"""
    global channel
    channel = new_channel
    channel.basic_recover(callback=handle_delivery,requeue=True)    

try:
    connection = pika.SelectConnection(parameters=parameters,\
        on_open_callback=on_connected)    

    # Loop so we can communicate with RabbitMQ
    connection.ioloop.start()
except KeyboardInterrupt:
    # Gracefully close the connection
    connection.close()
    # Loop until we're fully closed, will stop on its own
    connection.ioloop.start()
4

3 回答 3

82

未确认消息是那些已通过网络传递给消费者但尚未被确认或拒绝的消息——但消费者尚未关闭最初接收它们的通道或连接。因此，代理无法确定消费者是否只是花了很长时间来处理这些消息，或者它是否已经忘记了它们。因此，它使他们处于未确认状态，直到消费者死亡或他们被确认或拒绝。

由于这些消息将来仍然可以由最初消费它们的仍然活着的消费者有效处理，因此（据我所知）您不能将另一个消费者插入组合中并尝试对它们做出外部决策。您需要修复您的消费者，以便在处理每条消息时对它们做出决定，而不是让旧消息不被确认。

于 2011-08-15T18:32:37.013 回答
28

如果消息未被确认，则只有两种方法可以将它们重新放入队列：

  1. 基本的.nack

    此命令将导致消息被放回队列并重新传递。

  2. 断开与代理的连接

    此操作将强制将此通道中的所有未确认消息放回队列中。

注意：basic.recover 将尝试在同一个频道（给同一个消费者）上重新发布未确认的消息，这有时是所需的行为。

用于 basic.recover 和 basic.nack 的 RabbitMQ 规范

真正的问题是：为什么消息未被确认？

导致未确认消息的可能场景：

  1. 消费者获取了太多消息，然后没有足够快地处理和确认它们。

    解决方案：预取尽可能少的消息。

  2. 有缺陷的客户端库（我目前在pika 0.9.13上遇到了这个问题。如果队列中有很多消息，那么一定数量的消息会被卡住，即使是几个小时后也是如此。

    解决方案：我必须多次重启消费者，直到所有未确认的消息都从队列中消失。

于 2013-09-25T06:37:40.653 回答
6

一旦所有工作人员/消费者停止，所有未确认的消息将进入就绪状态。

通过确认输出来确保停止所有工作人员，并grep在ps aux发现时停止/杀死他们。

如果您使用主管管理工人，显示工人已停止，您可能需要检查僵尸。Supervisor 报告该工作程序已停止，但在 ps aux 输出上 grepped 时，您仍然会发现僵尸进程正在运行。杀死僵尸进程将使消息恢复到就绪状态。

于 2016-11-25T07:05:54.500 回答