发布者等待消费消息，而它应该做这两件事（发布和订阅来自订阅者的回复）。）答案

【问题标题】：Publisher waits for consuming messages while it should do both things(publishing and subscribing replies from subscriber).)发布者等待消费消息，而它应该做这两件事（发布和订阅来自订阅者的回复）。）
【发布时间】：2019-07-19 11:53:51
【问题描述】：

发布者在使用第一个 message_reply 后被阻止。它没有向订阅者发送任何消息。

我已经尝试过，如果我不使用 'start_sumption()' 方法，发布者会连续发送数据，但它不会打印来自订阅者的回复消息。如果我使用 'start_sumption()' 方法，它只会阻止发布者并等待。

#publisher.py
#!/usr/bin/env python
import pika
import sys
import uuid

connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

channel.queue_declare(queue='task_queue', durable=True)
channel.queue_declare(queue='reply_queue', durable=True)

message = "Hello World!"
corr_id = str(uuid.uuid4())

def on_response(ch, method, properties, body):
    print("----- On_response -----")
    print("Received CORR_ID : ", properties.correlation_id)
    if(corr_id == properties.correlation_id):
        resp = body.decode('utf-8')
        print("RESPONSE : ", resp)
    ch.basic_ack(delivery_tag=method.delivery_tag)

def consume_response(channel):
    channel.basic_consume(queue='reply_queue', on_message_callback=on_response)
    channel.start_consuming()

while True:
    channel.basic_publish(
        exchange='',
        routing_key='task_queue',
        body=message,
        properties=pika.BasicProperties(
            reply_to='reply_queue',
            correlation_id=corr_id
        ))
    print(" [x] Sent %r" % message)
    consume_response(channel)

#subscriber.py
#!/usr/bin/env python
import pika
import time

connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

channel.queue_declare(queue='task_queue', durable=True)
print(' [*] Waiting for messages. To exit press CTRL+C')


def callback(ch, method, properties, body):
    print(" [x] Received %r" % body.decode('utf-8'))
    #time.sleep(body.count(b'.'))
    print(" [x] Done")
    print("CORR_ID : ",str(properties.correlation_id))
    print("Reply : ", str(properties.reply_to))
    ch.basic_ack(delivery_tag=method.delivery_tag)
    res = "Received "+str(body.decode('utf-8'))
    ch.basic_publish(
        exchange='',
        routing_key='reply_queue',
        properties = pika.BasicProperties(
            correlation_id=\
                properties.correlation_id
        ),
        body=res
    )


channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue='task_queue', on_message_callback=callback)

channel.start_consuming()

Actual result is-
publisher                                           subscriber
   (1)----------publish message 'Hello'-------------->  |
    | <---------reply_to_publisher_queue---------------(2)
   (3)prints_the message
   (4)publisher waits for consuming messages

我想设计这个-

(expected)
publisher                                           subscriber
   (1)----------publish message 'Hello'-------------->  |
    | <---------reply_to_publisher_queue---------------(2)
   (3)prints_the message
   (4)----------publish message 'Hello'-------------->  |
    | <---------reply_to_publisher_queue---------------(5)
    .
    .
    .
   continues....

请帮忙。

【问题讨论】：

我看到您正在尝试实现类似 RPC 调用。你检查rabbitmq.com/tutorials/tutorial-six-python.html了吗？
是的，我已经检查了这段代码。在 'start_sumption()' 调用后，它会在控制台上打印收到的消息并仅开始消费。它不会进一步发布任何消息。如果我评论 'start_sumption()' 那么它只是默默地消耗。

标签： python python-3.x rabbitmq pika python-pika

【解决方案1】：

我将publisher.py 中的channel.connection.process_data_events(time_limit=1) 替换为channel.start_consuming() 行，它按预期工作。

我也关注了https://github.com/pika/pika/issues/770。

【讨论】：