RabbitMQ python 演示 route

Posted 2022-01-12 quietguoguo

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了RabbitMQ python 演示 route相关的知识，希望对你有一定的参考价值。

之前的例子使用RabbitMQ中fanout模式的exchange实现类似广播的方式发布消息，本次的例子中使用exchange的direct模式实现消息路由。

根据不同的route_key实现消息发往不同的队列中。

RabbitMQ

在此示例中 route_key 为orange black green

生产者代码

import sys
import pika

severity = sys.argv[1] if len(sys.argv) > 1 else info  # 第一个参数 作为消息类型
message =  .join(sys.argv[2:]) or "Direct Queues!"

credentials = pika.PlainCredentials(username="admin", password="admin")  # 连接凭证 如果不指定 则默认使用 guest guest
connection = pika.BlockingConnection(pika.ConnectionParameters(host=192.168.0.191,
                                                               virtual_host=/,
                                                               credentials=credentials))
channel = connection.channel()
channel.exchange_declare(exchange=direct, exchange_type=direct)  # 如果队列没有创建，就创建这个队列
channel.basic_publish(exchange=direct, routing_key=severity, body=message)
connection.close()

msg = "Message  has been sent to mq".format(message)
print(msg)

消费者

import pika
import sys

severities = sys.argv[1:]  # 使用参数作为队列关键字 可以输入多个关键字
if not severities:
    sys.stderr.write("Usage: %s [info] [warning] [error]\\n" % sys.argv[0])
    sys.exit(1)

def callback(ch, method, propertities,body):
    print(" [x] Received %r" % body)


# ########################### 消费者 ###########################
credentials = pika.PlainCredentials(username="admin", password="admin")  # 连接凭证 默认为 guest
connection = pika.BlockingConnection(pika.ConnectionParameters(host=192.168.0.191,
                                                               virtual_host=/,
                                                               credentials=credentials))
channel = connection.channel()
# 声明exchange
channel.exchange_declare(exchange=direct, exchange_type=direct)
# 匿名queue
result = channel.queue_declare(queue=, exclusive=True, )
queue_name = result.method.queue
# queue绑定
for severity in severities:
    channel.queue_bind(
        exchange=direct, queue=queue_name, routing_key=severity)
channel.basic_consume(queue=queue_name, on_message_callback=callback, auto_ack=True)
print( [*] Waiting for message. To exit press CTRL+C)
channel.start_consuming()