Python如何操作RabbitMQ实现direct关键字发布订阅模式？有录播直播私教课视频教程-拓冰建站

direct关键字发布订阅模式

基本用法

发布者

import json
from rabbitmq import pika
import rabbitmq# 建立连接
credentials = rabbitmq.PlainCredentials('zhangdapeng','zhangdapeng520',
)  # mq用户名和密码
connection_target = rabbitmq.ConnectionParameters(host='127.0.0.1',port=5672,virtual_host='/',credentials=credentials,
)
connection = rabbitmq.BlockingConnection(connection_target)# 队列信息
exchange_name = "user_manager_direct"# 创建管道
channel = connection.channel()# 声明一个交换机
channel.exchange_declare(exchange=exchange_name, exchange_type=rabbitmq.ExchangeType.direct)# 向队列中写入数据
user = {"id": 1, "name": "张三", "age": 23}
message = json.dumps(user, ensure_ascii=True)
channel.basic_publish(exchange=exchange_name,routing_key="error",  # 这里不再是队列名了，而是关键字body=message.encode('utf8'),properties=pika.BasicProperties(delivery_mode=2),  # 声明消息在队列中持久化
)
channel.basic_publish(exchange=exchange_name,routing_key="info",  # 这里不再是队列名了，而是关键字body=message.encode('utf8'),properties=pika.BasicProperties(delivery_mode=2),  # 声明消息在队列中持久化
)
print(message)# 关闭连接
connection.close()

消费者

import rabbitmq
import json# 创建连接
credentials = rabbitmq.PlainCredentials('zhangdapeng','zhangdapeng520',
)
target = rabbitmq.ConnectionParameters(host='127.0.0.1',port=5672,virtual_host='/',credentials=credentials,
)
connection = rabbitmq.BlockingConnection(target)# 创建管道
channel = connection.channel()# 队列信息
exchange_name = "user_manager_direct"# 绑定交换机
channel.exchange_declare(exchange=exchange_name,exchange_type=rabbitmq.ExchangeType.direct,
)# 绑定队列
result = channel.queue_declare(queue="",  # 这里不要指定队列名exclusive=True,
)
queue_name = result.method.queue  # 通过result获取队列名
channel.queue_bind(exchange=exchange_name,queue=queue_name,routing_key="error",  # 可以通过routing_key绑定多个关键字
)
channel.queue_bind(exchange=exchange_name,queue=queue_name,routing_key="info",  # 可以通过routing_key绑定多个关键字
)def callback(ch, method, properties, body):"""每次接收到消息的消费回调方法"""ch.basic_ack(delivery_tag=method.delivery_tag)data = body.decode("utf8")print(json.loads(data))# 开始消费
channel.basic_consume(queue=queue_name,on_message_callback=callback,auto_ack=False,
)
try:channel.start_consuming()
finally:connection.close()

简化代码

生产者

import json
from rabbitmq import pika
import rabbitmq# 建立连接
connection = rabbitmq.get_connection()# 队列信息
exchange_name = "user_manager_direct"# 创建管道
channel = connection.channel()# 声明一个交换机
channel.exchange_declare(exchange=exchange_name, exchange_type=rabbitmq.ExchangeType.direct)# 向队列中写入数据
user = {"id": 1, "name": "张三", "age": 23}
rabbitmq.send_json(channel, user, exchange_name, "error")
rabbitmq.send_json(channel, user, exchange_name, "info")# 关闭连接
connection.close()

消费者

import rabbitmq
import json# 创建连接
connection = rabbitmq.get_connection()# 创建管道
channel = connection.channel()# 队列信息
exchange_name = "user_manager_direct"# 绑定交换机
channel.exchange_declare(exchange=exchange_name,exchange_type=rabbitmq.ExchangeType.direct,
)# 绑定队列
result = channel.queue_declare(queue="",  # 这里不要指定队列名exclusive=True,
)
queue_name = result.method.queue  # 通过result获取队列名
channel.queue_bind(exchange=exchange_name,queue=queue_name,routing_key="error",  # 可以通过routing_key绑定多个关键字
)
channel.queue_bind(exchange=exchange_name,queue=queue_name,routing_key="info",  # 可以通过routing_key绑定多个关键字
)def callback(ch, method, properties, body):"""每次接收到消息的消费回调方法"""print(rabbitmq.receive_json(ch, method, body))# 开始消费
rabbitmq.consume(connection, queue_name, callback)

进一步简化代码

生产者

import rabbitmq# 建立连接
connection = rabbitmq.get_connection()# 队列信息
exchange_name = "user_manager_direct"# 创建管道
channel, _ = rabbitmq.get_direct_channel(connection, exchange_name)# 向队列中写入数据
user = {"id": 1, "name": "张三", "age": 23}
rabbitmq.send_json(channel, user, exchange_name, "error")
rabbitmq.send_json(channel, user, exchange_name, "info")# 关闭连接
connection.close()

消费者

import rabbitmq
import json# 创建连接
connection = rabbitmq.get_connection()# 队列信息
exchange_name = "user_manager_direct"# 创建管道
channel, queue_name = rabbitmq.get_direct_channel(connection, exchange_name, ["error", "info"])def callback(ch, method, properties, body):"""每次接收到消息的消费回调方法"""print(rabbitmq.receive_json(ch, method, body))# 开始消费
rabbitmq.consume(connection, queue_name, callback)