python - Pika RabbitMQ 从消费者发布
问题描述
我有一个 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()
解决方案
你可以实现你的消费者,如果连接关闭,它会自动重新连接到 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)
推荐阅读
- python - 如何将日期时间从 tkcalendar 转换为字符串?
- ios - URL 会话保存图像
- cmake - 如何使用 CMake 构建以 ghdl 作为自定义目标的源?
- laravel - Laravel 视图进入另一个视图
- django - 清理多个 Pipenv 环境
- flutter - FormatException:JSON 中的意外字符
- java - 尝试为 4 个简单的条件制作 RegEx
- excel - 排除组合框中的重复值
- python - 使用 tarfile.extractfile 和 Python3.8 返回存档的名称而不是存档文件
- java - 最大堆插入和排序 Java