在Python中是否可以将RabbitMQ直接回复功能与Pika生成器使用者一起使用？

Question

我想在Python中的Pika客户端库中使用RabbitMQ的直接回复功能。它适用于基本消费者。但它引发了发电机使用者的以下例外情况：

鼠兔。例外情况。ChannelClosedByBroker:（406，“前提条件_失败-快速回复使用者不存在”）

有没有一种方法可以将直接回复功能用于生成器使用者？

使用基本消费者的示例客户端代码（它的工作原理）：

import pika


def handle(channel, method, properties, body):
    message = body.decode()
    print("received:", message)


connection = pika.BlockingConnection()
channel = connection.channel()

with connection, channel:
    message = "hello"
    channel.basic_consume(queue="amq.rabbitmq.reply-to",
                          on_message_callback=handle, auto_ack=True)
    channel.basic_publish(
        exchange="", routing_key="test", body=message.encode(),
        properties=pika.BasicProperties(reply_to="amq.rabbitmq.reply-to"))
    print("sent:", message)
    channel.start_consuming()

使用生成器消费者的示例客户端代码（它引发了异常）：

import pika


def handle(channel, method, properties, body):
    message = body.decode()
    print("received:", message)


connection = pika.BlockingConnection()
channel = connection.channel()

with connection, channel:
    message = "hello"
    channel.basic_publish(
        exchange="", routing_key="test", body=message.encode(),
        properties=pika.BasicProperties(reply_to="amq.rabbitmq.reply-to"))
    print("sent:", message)

    for (method, properties, body) in channel.consume(
            queue="amq.rabbitmq.reply-to", auto_ack=True):
        handle(channel, method, properties, body)

环境Windows 10、RabbitMQ 3.7.13、CPython 3.7.3、Pika 1.0.1。

注-在使用基本使用者的示例客户端代码中的basic\u publish方法之后调用basic\u consume方法会引发与使用生成器使用者相同的异常：

import pika


def handle(channel, method, properties, body):
    message = body.decode()
    print("received:", message)


connection = pika.BlockingConnection()
channel = connection.channel()

with connection, channel:
    message = "hello"
    channel.basic_publish(
        exchange="", routing_key="test", body=message.encode(),
        properties=pika.BasicProperties(reply_to="amq.rabbitmq.reply-to"))
    print("sent:", message)
    channel.basic_consume(queue="amq.rabbitmq.reply-to",
                          on_message_callback=handle, auto_ack=True)
    channel.start_consuming()

Answer 1

我推荐pika文档中给出的示例。

这对我也很管用。

这就是地点

Answer 2

正如Luke Bakken在这里所建议的，这就是诀窍：

import pika


def handle(channel, method, properties, body):
    message = body.decode()
    print("received:", message)


connection = pika.BlockingConnection()
channel = connection.channel()

with connection, channel:
    message = "hello"
    next(channel.consume(queue="amq.rabbitmq.reply-to", auto_ack=True,
                         inactivity_timeout=0.1))
    channel.basic_publish(
        exchange="", routing_key="test", body=message.encode(),
        properties=pika.BasicProperties(reply_to="amq.rabbitmq.reply-to"))
    print("sent:", message)

    for (method, properties, body) in channel.consume(
            queue="amq.rabbitmq.reply-to", auto_ack=True):
        handle(channel, method, properties, body)