如何使用 Pika 发送和接收 RabbitMQ 消息？

Question

我在让 Pika 以与 AMQP 或 RabbitMQ 文档一致的方式使用路由密钥或交换时遇到一些问题。我知道 RabbitMQ 文档使用旧版本的 Pika，所以我忽略了他们的示例代码。

我想做的是定义一个队列“订单”并有两个消费者，一个处理交换或routing_key“生产”，另一个处理“测试”。从 RabbitMQ 文档来看，通过使用直接交换和路由键或使用主题交换应该很容易做到。

然而，Pika 似乎不知道如何处理交换和路由密钥。使用 RabbitMQ 管理工具检查队列，很明显 Pika 没有正确对消息进行排队，或者 RabbitMQ 只是将其丢弃。

在消费者方面，我不太清楚应该如何将消费者绑定到交换或处理路由密钥，并且文档并没有真正的帮助。

如果我放弃所有想法或交换和路由密钥，消息会很好地排队并且很容易被我的消费者处理。

人们拥有的任何指针或示例代码都会很好。

Answer 1

事实证明，我对AMQP的理解并不完整。

其思想如下：

客户端：

获得连接后的客户端除了交换器的名称和路由密钥之外不应该关心任何其他内容。也就是说，我们不知道这将最终进入哪个队列。

channel.basic_publish(exchange='order',
                      routing_key="order.test.customer",
                      body=pickle.dumps(data),
                      properties=pika.BasicProperties(
                          content_type="text/plain",
                          delivery_mode=2))

Consumer

当通道打开时，我们声明交换和队列

channel.exchange_declare(exchange='order', 
                         type="topic", 
                         durable=True, 
                         auto_delete=False)

channel.queue_declare(queue="test", 
                      durable=True, 
                      exclusive=False, 
                      auto_delete=False, 
                      callback=on_queue_declared)

当队列准备好时，在“on_queue_declared”回调中是一个很好的选择地方，我们可以使用我们想要的路由密钥将队列绑定到交换器。

channel.queue_bind(queue='test', 
                   exchange='order', 
                   routing_key='order.test.customer')

#handle_delivery is the callback that will actually pickup and handle messages
#from the "test" queue
channel.basic_consume(handle_delivery, queue='test') 

使用路由键“order.test.customer”发送到“order”交换的消息现在将被路由到“test”队列，消费者可以在其中接收它。

Answer 2

虽然西蒙的答案总体上似乎是正确的，但您可能需要交换参数来使用

channel.basic_consume(queue='test', on_message_callback=handle_delivery) 

基本设置就像

credentials = pika.PlainCredentials("some_user", "some_password")
parameters = pika.ConnectionParameters(
    "some_host.domain.tld", 5672, "some_vhost", credentials
)
connection = pika.BlockingConnection(parameters)
channel = connection.channel()

开始使用：

channel.start_consuming()