来自消费者的Pika RabbitMQ发布

Question

我有一个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()

Answer 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)