我正在尝试使用RabbitMQ将python词典从python生产者发送到python消费者。生产者首先建立与本地RabbitMQ服务器的连接。然后,它创建一个将消息传递到的队列,最后发送该消息。使用者首先连接到RabbitMQ服务器,然后通过创建相同的队列来确保该队列存在。然后,它在回调函数中接收来自生产者的消息,并输出'id'值(1)。这是生产者和消费者的脚本:
producer.py脚本:
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.queue_declare(queue='task_queue', durable=True)
message = {'id': 1, 'name': 'name1'}
channel.basic_publish(exchange='',
routing_key='task_queue',
body=message,
properties=pika.BasicProperties(
delivery_mode = 2, # make message persistent
))
print(" [x] Sent %r" % message)
connection.close()
consumer.py脚本:
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)
print(body['id'])
print(" [x] Done")
ch.basic_ack(delivery_tag = method.delivery_tag)
channel.basic_qos(prefetch_count=1)
channel.basic_consume(callback,
queue='task_queue')
channel.start_consuming()
但是,当我运行producer.py时,出现此错误:
line 18, in <module>
delivery_mode = 2, # make message persistent
File "/Library/Python/2.7/site-packages/pika/adapters/blocking_connection.py", line 1978, in basic_publish
mandatory, immediate)
File "/Library/Python/2.7/site-packages/pika/adapters/blocking_connection.py", line 2064, in publish
immediate=immediate)
File "/Library/Python/2.7/site-packages/pika/channel.py", line 338, in basic_publish
(properties, body))
File "/Library/Python/2.7/site-packages/pika/channel.py", line 1150, in _send_method
self.connection._send_method(self.channel_number, method_frame, content)
File "/Library/Python/2.7/site-packages/pika/connection.py", line 1571, in _send_method
self._send_message(channel_number, method_frame, content)
File "/Library/Python/2.7/site-packages/pika/connection.py", line 1596, in _send_message
content[1][s:e]).marshal())
TypeError: unhashable type
有人可以帮我吗?谢谢!
您无法发送本机Python类型作为有效负载,您必须先对其进行序列化。我建议使用JSON:
import json
channel.basic_publish(exchange='',
routing_key='task_queue',
body=json.dumps(message),
properties=pika.BasicProperties(
delivery_mode = 2, # make message persistent
))
和
def callback(ch, method, properties, body):
print(" [x] Received %r" % json.loads(body))