RabbitMQ 非阻塞消费者

Question

我在 Python 中使用 RabbitMQ 来管理生产者和多个消费者之间的多个队列。在 RabbitMQ 网站 (routing model) 的示例中，消费者被阻止。这意味着它们在每次队列中有新的“任务”时停止在 start_sumption() 并执行回调函数。

我的问题是：我如何以一种仍在等待任务的方式实现我的消费者（因此，每次队列中有新事物时都会调用回调函数）但同时他可以执行其他任务工作/代码。

谢谢

Answer 1

表格FAQ:

Pika 在代码中没有任何线程概念。如果你想 使用带线程的 Pika，确保每个都有 Pika 连接 线程，在该线程中创建。分享一只鼠兔是不安全的 跨线程连接，

所以让我们在线程内部创建连接：

import pika


class PikaMassenger():

    exchange_name = '...'

    def __init__(self, *args, **kwargs):
        self.conn = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
        self.channel = self.conn.channel()
        self.channel.exchange_declare(
            exchange=self.exchange_name, 
            exchange_type='topic')

    def consume(self, keys, callback):
        result = self.channel.queue_declare('', exclusive=True)
        queue_name = result.method.queue
        for key in keys:
            self.channel.queue_bind(
                exchange=self.exchange_name, 
                queue=queue_name, 
                routing_key=key)

        self.channel.basic_consume(
            queue=queue_name, 
            on_message_callback=callback, 
            auto_ack=True)

        self.channel.start_consuming()


    def __enter__(self):
        return self


    def __exit__(self, exc_type, exc_value, traceback):
        self.conn.close()

def start_consumer():

    def callback(ch, method, properties, body):
        print(" [x] %r:%r consumed" % (method.routing_key, body))

    with PikaMassenger() as consumer:
        consumer.consume(keys=[...], callback=callback)


consumer_thread = threading.Thread(target=start_consumer)
consumer_thread.start()

Answer 2

对于接收者

import pika

messages = []
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='message')

def callback(ch, method, properties, message):
    print(message)
    messages.append(message)

channel.basic_consume(callback,queue='message',no_ack=True)

和

channel.basic_consume(callback,queue='message',no_ack=True)

当你需要时） 或在线程中

import threading

import pika
import time

messages = []

def recieve_messages():
    connection = pika.BlockingConnection(pika.ConnectionParameters(
        'localhost'))
    channel = connection.channel()
    channel.queue_declare(queue='hello')

    def callback(ch, method, properties, body):
        messages.append(body)

    channel.basic_consume(callback,
                          queue='hello',
                          no_ack=True)
    # channel.start_consuming()
    mq_recieve_thread = threading.Thread(target=channel.start_consuming)
    mq_recieve_thread.start()

recieve_messages()
while True:
    print(messages)
    time.sleep(1)