使用 Quart 运行 RabbitMQ Pika

问题描述

我正在使用 Quart 框架，但我也需要使用 RabbitMQ pika 连接器，但我无法让它们发挥良好的作用，因为它们都有无限循环。

入口点：

from quart import Quart
from .service import Service

app = Quart(__name__)

@app.before_serving
async def startup():
app.service_task = asyncio.ensure_future(service.start())

if not service.initialise():
    sys.exit()

服务类：

class Service:
def __init__(self,new_instance):
    self._connection = None
    self._channel = None
    self._messaging_thread = None

def initialise(self):

    credentials = pika.PlainCredentials('username','password')
    parameters = pika.ConnectionParameters('localhost',credentials=credentials)
    self._connection = pika.BlockingConnection(parameters)
    self._channel = self._connection.channel()
    self._channel.queue_declare(queue='to_be_processed_queue')

    self._channel.basic_consume(queue='to_be_processed_queue',auto_ack=True,on_message_callback=self.callback)
  
    print('creating thread')
    self._messaging_thread = Thread(target=self.run_consume())
    #self._messaging_thread.start()
    print('Thread created...')

def run_consume(self):
    try:
        self._channel.start_consuming()
    except KeyboardInterrupt:
        self._shutdown()

代码甚至没有进入print('Thread created...')，我不明白。从 this question 我确实明白 RabbitMQ 不是线程安全的，但我不明白如何运行 RabbitMQ。

解决方法

正如您已经发现的那样，Pika 不是线程安全的，但这不是您的程序阻塞的原因。

您的问题可能出在这里：

print('creating thread')
self._messaging_thread = Thread(target=self.run_consume())
#self._messaging_thread.start()

如果去掉 run_consume 中的括号，效果会更好吗？现在您实际上不是在创建线程而是当场执行 self.run_consume() 并且它不会退出。

self._messaging_thread = Thread(target=self.run_consume)

将是我的第一次尝试。

然而，由于 Pika 不是线程安全的，您还必须将您的频道创建和内容移动到您的线程中，而不是在主程序中进行。如果您不在其他任何地方使用它，它可能会起作用，但使用 Pika 的正确方法是将所有内容完全包含在线程中，而不是像您现在在这里所做的那样在线程之间共享任何 Pika 结构。

pika python quart