前言

使用场景:

我们希望接收端指定接收某个队列的消息的时候,此时为直连交换类型

原理:

每个接收端在绑定交换机的时候可以设置相应路由键,每个发送端在发送消息的时候可以指明路由键,交换机可以根据路由键将数据发送到指定的队列中,这样接收端就能从指定的队列获取到相应的数据

代码

send

import pika

hostname = '127.0.0.1'
parameters = pika.ConnectionParameters(hostname)
connection = pika.BlockingConnection(parameters)

# 创建通道
channel = connection.channel()
# 定义交换机，设置类型为direct
channel.exchange_declare(exchange='change_dir', exchange_type='direct')

# 定义三个路由键
routings = ['info', 'warning', 'error']

# 将消息依次发送到交换机，并设置路由键
for routing in routings:
    message = '%s message.' % routing
    channel.basic_publish(exchange='change_dir', routing_key=routing, body=message)
    print(message)

connection.close()

receive

import sys, pika

hostname = '127.0.0.1'
parameters = pika.ConnectionParameters(hostname)
connection = pika.BlockingConnection(parameters)

# 创建通道
channel = connection.channel()
# 定义交换机，设置类型为direct
channel.exchange_declare(exchange='change_dir', exchange_type='direct')

# 从命令行获取路由键参数，如果没有，则设置为info
routings = sys.argv[1:]
if not routings:
    routings = ['info']

# 生成临时队列，
result = channel.queue_declare(queue='change_dir', exclusive=True)  # exclusive=True 当接收端退出的时候 会销毁那个临时创建的队列
queue_name = result.method.queue
print(queue_name)
for routing in routings:
    # 绑定到交换机上，设置路由键
    channel.queue_bind(exchange='change_dir', queue=queue_name, routing_key=routing)


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


channel.basic_consume(
    queue='change_dir',  # 指定队列名
    on_message_callback=callback,  # 从队列里获取消息
    auto_ack=False  # mq服务器挂掉 防止任务丢失
)

print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()

演示

开启一个发送端一个接收端并且让发送端发送数据

 接收端

 发送者发送了三个数据 但是其只能接收一个数据 属于自己路由匹配的