Pika RabbitMQ 从消费者发布答案

【问题标题】：Pika RabbitMQ Publish from a consumerPika RabbitMQ 从消费者发布
【发布时间】：2020-04-29 03:36:40
【问题描述】：

我有一个 RabbitMQ 消费者。我想让该消费者进行一些消息处理，由time.sleep(10) 模拟，然后将消息发布到不同的队列。我知道消费者回调有一个理论上可以用于发布的通道，但这似乎是一个糟糕的实现，因为如果basic_publish() 以某种方式设法强制关闭通道，那么消费者就会死亡。处理这个问题的最佳方法是什么？

import time
import pika

connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='logs', exchange_type='fanout')

result = channel.queue_declare(queue='original_queue', exclusive=True)

channel.queue_bind(exchange='logs', queue='original_queue')

print(' [*] Waiting for logs. To exit press CTRL+C')

def callback(ch, method, properties, body):
    time.sleep(10)
    ch.basic_publish(exchange='logs', routing_key='different_queue', body='hello_world')

channel.basic_consume(
    queue='original_queue', on_message_callback=callback, auto_ack=True)

channel.start_consuming()

【问题讨论】：

“basic_publish() 以某种方式强制关闭频道”，此信息的来源在哪里？在文档中没有找到。

标签： python python-3.x rabbitmq pika python-pika

【解决方案1】：

您可以实现您的消费者，如果连接关闭，它会自动重新连接到 RabbitMQ 服务器。希望这会有所帮助（我没有在设计部分考虑太多，请随意提出一些建议！）

import time
import pika

reconnect_on_failure = True


def consumer(connection, channel):

    channel.exchange_declare(exchange='logs', exchange_type='fanout')
    result = channel.queue_declare(queue='original_queue', exclusive=True)
    channel.queue_bind(exchange='logs', queue='original_queue')
    print(' [*] Waiting for logs. To exit press CTRL+C')

    def callback(ch, method, properties, body):
        time.sleep(10)
        ch.basic_publish(exchange='logs', routing_key='different_queue', body='hello_world')

    channel.basic_consume(
  queue='original_queue', on_message_callback=callback, auto_ack=True)

    channel.start_consuming()


def get_connection_and_channel():
    connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
    channel = connection.channel()


def start(reconnect_on_failure):
    connection, channel = get_connection_and_channel()
    consumer(connection, channel)
    # the if condition will be executed when the consumer's start_consuming loop exists
    if reconnect_on_failure:
        # cleanly close the connection and channel
        if not connection.is_closed():
            connection.close()
        if not channel.is_close():
            channel.close()
        start(reconnect_on_failure)


start(reconnect_on_failure)

【讨论】：