如何为 pika 库上的方法 start_sumption() 添加超时答案

【问题标题】：How to add a timeout to method start_consuming() on pika library如何为 pika 库上的方法 start_sumption() 添加超时
【发布时间】：2019-07-12 02:23:38
【问题描述】：

我有一个BlockingConnection，我关注 pika 文档的the examples。但在所有这些中，开始消费消息的代码示例是：

connection = pika.BlockingConnection()
channel = connection.channel()
channel.basic_consume('test', on_message)
try:
    channel.start_consuming()
except KeyboardInterrupt:
    channel.stop_consuming()
connection.close()

（或多或少的细节）。

我必须编写许多脚本，并且我想一个接一个地运行（用于测试/研究目的）。但是上面的代码要求我每一个都加^C。

我尝试添加一些超时explained in the documentation，但我没有运气。例如，如果我找到一个 set 参数，如果客户端在最后 X 秒内没有消费任何消息，则脚本完成。这在 pika lib 中可行吗？还是我必须改变方法？

【问题讨论】：

您希望您的代码在一定时间后自动杀死消费者。对吗？
@bumblebee 好的，这可能是一个选择。但是这个“时间量”应该是在队列中不存在更多消息之后。例如，在 C++ 客户端 you can to set a timeout.

标签： python rabbitmq pika

【解决方案1】：

如果您不想阻止代码，请不要使用start_consuming。使用SelectConnection 或使用consume 的this method。您可以为传递给consume 的参数添加超时。

_{注意：RabbitMQ 团队监控 rabbitmq-users mailing list 并且仅有时会回答 StackOverflow 上的问题。}

【讨论】：

【解决方案2】：

import pika

parameters = pika.ConnectionParameters(host="localhost")
connection = pika.BlockingConnection(parameters)
channel = connection.channel()

def ack_message(channel, method):
    """Note that `channel` must be the same pika channel instance via which
    the message being ACKed was retrieved (AMQP protocol constraint).
    """
    if channel.is_open:
        channel.basic_ack(method.delivery_tag)
    else:
        # Channel is already closed, so we can't ACK this message;
        # log and/or do something that makes sense for your app in this case.
        pass

def callback(channel,method, properties, body):
    ack_message(channel,method)
    print("body",body, flush=True)

channel.basic_consume(
    queue="hello", on_message_callback=callback)

channel.start_consuming()
connection.close()

我的原始代码是 Luke Bakken 的答案。
但我已经稍微编辑了代码。
:)

【讨论】：