rabbitmq - 如何从连接自己的 channel 以外的其他 channel 恢复未确认的 AMQP 消息？

Question

似乎我的rabbitmq服务器运行的时间越长，未确认消息带来的麻烦就越多。我很乐意重新排队。事实上，似乎有一个 amqp 命令可以执行此操作，但它仅适用于您的连接正在使用的 channel 。我构建了一个小 pika 脚本来至少尝试一下，但我要么遗漏了一些东西，要么无法以这种方式完成(使用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()

Answer 1

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

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