为 Pika ioloop async (RabbitMQ) 设置超时 [英] Set Timeout for Pika ioloop async (RabbitMQ)
问题描述
我需要能够优雅地停止在 Pika ioloop 中工作的消费者(工人).工作人员应在 60 秒后停止.当前处理的消息应该已完成.
I need to be able to gracefully stop a consumer (worker) who works in a Pika ioloop. The worker should stop after 60 seconds. Currently processed messages should be finished.
我试图在回调函数中放入一个 connection.close()
但这只会停止当前线程而不是完整的 ioloop.它给出了一个可怕的错误输出.
I tried to put a connection.close()
inside the callback function but that only stopped the current thread and not the complete ioloop. And it gave a terrible error output.
请参阅我的代码中的第 16 行及以下内容:我使用了(关于 Pika ioloop 的基本示例 http://pika.github.com/connecting.html#cps-example:
Please see line 16 and following in my code: I used the (basic example about Pika ioloop http://pika.github.com/connecting.html#cps-example:
from pika.adapters import SelectConnection
channel = None
def on_connected(connection):
connection.channel(on_channel_open)
def on_channel_open(new_channel):
global channel
channel = new_channel
channel.queue_declare(queue="test", durable=True, exclusive=False, auto_delete=False, callback=on_queue_declared)
def on_queue_declared(frame):
channel.basic_consume(handle_delivery, queue='test')
def handle_delivery(channel, method, header, body):
print body
# timer stuff which did NOT work
global start_time, timeout, connection
time_diff = time.time()-start_time
if time_diff > timeout:
#raise KeyboardInterrupt
connection.close()
timeout = 60
start_time = time.time()
connection = SelectConnection(parameters, on_connected)
try:
connection.ioloop.start()
except KeyboardInterrupt:
connection.close()
connection.ioloop.start()
推荐答案
您可以在打开的连接上附加超时回调函数.这是您的示例的额外代码.
You can attach a timeout call-back function on the opened connection. Here is the extra code for your example.
timeout = 60
def on_timeout():
global connection
connection.close()
connection.add_timeout(timeout, on_timeout)
这篇关于为 Pika ioloop async (RabbitMQ) 设置超时的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!