rabbitmq(pika) 使用 RPC 时抛出异常

Posted 2023-02-16

【中文标题】rabbitmq(pika) 使用 RPC 时抛出异常

【英文标题】：rabbitmq(pika) throws an exception when use RPC

【发布时间】：2015-06-07 14:57:13

【问题描述】：

当我构建发布者-消费者模式时，RabbiMQ RPC 抛出异常。这是我的代码： 发件人.py

# coding:utf-8
import pika
import uuid


class MessageSender(object):

    def __init__(self):
        self.connention = pika.BlockingConnection(
            pika.ConnectionParameters(host='localhost'))
        self.channel = self.connention.channel()

        result = self.channel.queue_declare(exclusive=True)
        self.callback_queue = result.method.queue
        print 'MessageSender callback queue is ------', self.callback_queue
        self.channel.basic_consume(
            self.on_response, no_ack=True, queue=self.callback_queue)

    def on_response(self, ch, method, props, body):
        if self.corr_id == props.correlation_id:
            self.respose = body

    def send(self):
        self.respose = None
        self.corr_id = str(uuid.uuid4())
        self.channel.basic_publish(exchange='',
                                   routing_key='rpc_queue',
                                   properties=pika.BasicProperties(
                                       reply_to=self.callback_queue,
                                       correlation_id=self.corr_id,),
                                   body='MESSAGE')
        print '[x]Send sucessful'
        while self.respose is None:
            self.connention.process_data_events()
        print 'MessageSender get callback ------', self.respose


if __name__ == '__main__':
    sender = MessageSender()
    sender.send()

worker.py

# coding:utf-8
import pika
import time


class MessageWorker(object):

    def __init__(self):
        self.connection = pika.BlockingConnection(
            pika.ConnectionParameters(host='localhost'))
        self.channel = self.connection.channel()
        self.channel.queue_declare(queue='rpc_queue')

    def on_request(self, ch, method, props, body):
        print 'MessageWorker get message ------', body
        response = task(body)
        print 'MessageWorker callback on queue ------', props.reply_to
        ch.basic_publish(exchange='',
                         routing_key=props.reply_to,
                         properties=pika.BasicProperties(
                             correlation_id=props.correlation_id),
                         body=str(response))
        print 'MessageWorker send message to MessageSender'
        ch.basic_ask(delivery_tag=method.delivery_tag)

    def work(self):
        self.channel.basic_qos(prefetch_count=2)
        self.channel.basic_consume(self.on_request, queue='rpc_queue')
        print '[x]Waiting message...'
        self.channel.start_consuming()


def task(body):
    time.sleep(3)
    return body


if __name__ == '__main__':
    worker = MessageWorker()
    worker.work()

发送者工作正常，但是worker.py总是报错：

[x]Waiting message...
MessageWorker get message ------ MESSAGE
MessageWorker callback on queue ------ amq.gen-6l5oJapbiKIqG3ZYTRwqpA
MessageWorker send message to MessageSender
Traceback (most recent call last):
  File "python/workers/new_worker.py", line 40, in <module>
    worker.work()
  File "python/workers/new_worker.py", line 30, in work
    self.channel.start_consuming()
  File "/usr/local/lib/python2.7/dist-packages/pika/adapters/blocking_connection.py", line 955, in start_consuming
    self.connection.process_data_events()
  File "/usr/local/lib/python2.7/dist-packages/pika/adapters/blocking_connection.py", line 243, in process_data_events
    raise exceptions.ConnectionClosed()
pika.exceptions.ConnectionClosed

这困扰了我将近一个星期！任何人都可以帮忙吗？谢谢:)

【问题讨论】：

哦，另一个问题，当我想使用 RPC 时，routing_key 参数必须等于'rpc_queue'？因为我将它设置为 'rpc'，worker 无法获取 sender 发送的内容！

【参考方案1】：

好的，我知道发生了什么。在worker.py中，我拼错了

ch.basic_ack(delivery_tag=method.delivery_tag)

到

ch.basic_ask(delivery_tag=method.delivery_tag)

天哪！T.T