使用 pika 的 start_sumption 方法中断线程答案

【问题标题】：interrupt thread with start_consuming method of pika使用 pika 的 start_sumption 方法中断线程
【发布时间】：2015-11-20 02:31:39
【问题描述】：

我有一个线程使用 pika 监听来自 rabbitmq 的新消息。使用 BlockingConnection 配置连接后，我开始通过 start_sumption 消费消息。如何中断启动消费方法调用，例如以优雅的方式停止线程？

【问题讨论】：

向您的消费者发送 basic_cancel？让您的消费者从一个控制队列中监听，该队列在消费者需要停止时注入“退出”消息？
channel.stop_consuming()。当您收到“退出”消息时，在使用basic_consume 注册的回调中调用它。在channel.start_consuming() 之后添加优雅的停止指令。

标签： python rabbitmq pika

【解决方案1】：

您可以使用consume generator 代替 start_sumption。

import threading

import pika


class WorkerThread(threading.Thread):
    def __init__(self):
        super(WorkerThread, self).__init__()
        self._is_interrupted = False

    def stop(self):
        self._is_interrupted = True

    def run(self):
        connection = pika.BlockingConnection(pika.ConnectionParameters())
        channel = connection.channel()
        channel.queue_declare("queue")
        for message in channel.consume("queue", inactivity_timeout=1):
            if self._is_interrupted:
                break
            if not message:
                continue
            method, properties, body = message
            print(body)

def main():
    thread = WorkerThread()
    thread.start()
    # some main thread activity ...
    thread.stop()
    thread.join()


if __name__ == "__main__":
    main()

【讨论】：

inactivity_timeout=1 参数和continue 声明是天才！你让我开心，谢谢。
这是一个忙等待循环，但 pika 连接不是线程安全的，这是最好的答案。
这真的很有帮助，不敢相信我以前没有在文档中看到这个。从现在开始要以这种方式使用 pika...
感谢您的回答，为了让它对我有用，我不得不将 if not message 更改为 if not all(message)，因为消息是 tuple (None, None, None)