RabbitMQ---订阅模型-Direct

在RabbitMQ中，Direct Exchange是一种点对点的模式，它将消息路由到那些binding key与routing key完全匹配的队列中。

以下是一个使用Python和pika库实现的基于Direct Exchange的生产者和消费者的示例代码：

生产者（发送消息）:

import pika
 
# 连接到RabbitMQ服务器
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
 
# 声明Direct Exchange
channel.exchange_declare(exchange='direct_logs', exchange_type='direct')
 
# 发送消息
severity = 'info'
message = 'Hello World!'
channel.basic_publish(exchange='direct_logs',
                      routing_key=severity,
                      body=message)
 
print(f" [x] Sent '{message}'")
 
# 关闭连接
connection.close()

消费者（接收消息）:

import pika
 
# 连接到RabbitMQ服务器
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
 
# 声明Direct Exchange
channel.exchange_declare(exchange='direct_logs', exchange_type='direct')
 
# 声明一个临时队列
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
 
# 根据需求绑定对应的routing key
severities = ['info', 'warning', 'error']
for severity in severities:
    channel.queue_bind(exchange='direct_logs',
                       queue=queue_name,
                       routing_key=severity)
 
# 定义一个回调函数来处理消息
def callback(ch, method, properties, body):
    print(f" [x] Received {body}")
 
# 开始监听队列
channel.basic_consume(queue=queue_name, on_message_callback=callback, auto_ack=True)
print(f" [*] Waiting for messages. To exit press CTRL+C")
channel.start_consuming()

在这个例子中，我们创建了一个名为direct_logs的Direct Exchange，并且定义了三个消费者（对应于info、warning、error三个routing key）来监听队列。生产者发送消息时，它将消息发送到direct_logs Exchange，并通过指定的routing key（例如info、warning、error）来决定消息流向哪个队列。消费者将接收到它所绑定的routing key的消息。