python使用pika库调用rabbitmq交换机模式详解
作者:IT之一小佬 时间:2024-01-01 06:28:18
前言:
交换机模式主要包括:交换机之发布订阅、交换机之关键字和交换机之通配符。
1、交换机之发布订阅
发布订阅和简单的消息队列区别在于,发布订阅会将消息发送给所有的订阅者,而消息队列中的数据被消费一次便消失。所以,RabbitMQ实现发布和订阅时,会为每一个订阅者创建一个队列,而发布者发布消息时,会将消息放置在所有相关队列中。
生产者模式:
示例代码:
import pika
# 1.连接rabbit
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.124.104'))
channel = connection.channel()
# 2.声明一个名为logs类型的fanout的交换机
channel.exchange_declare(exchange='logs', exchange_type='fanout') # 发布订阅模式参数
# 3.向logs交换机中插入数据:"Hello world"
message = 'info:Hello World!'
channel.basic_publish(exchange='logs',
routing_key='',
body=message,
properties=pika.BasicProperties(
delivery_mode=2, # make message persistent
)
)
print(" [x] Sent 'Hello World!'")
运行结果:
消费者模式:
示例代码:
import pika
# 1.连接rabbit
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.124.104'))
channel = connection.channel()
# 2.声明一个名为logs类型的fanout的交换机
channel.exchange_declare(exchange='logs', exchange_type='fanout')
# 3.创建队列
result = channel.queue_declare("", exclusive=True) # 随机生成一个队列名
queue_name = result.method.queue
print(queue_name)
# 4.将指定队列绑定到交换机上
channel.queue_bind(exchange='logs', queue=queue_name)
# 5.确定回调函数
def callback(ch, method, properties, body):
print(" [x] Received %r" % body)
ch.basic_ack(delivery_tag=method.delivery_tag)
# 6.确定监听队列参数
channel.basic_consume(queue=queue_name, # 指定队列
auto_ack=False, # 手动应答方式
on_message_callback=callback)
print(' [*] Waiting for messages. To exit press CTRL+C')
# 7.正式监听
channel.start_consuming()
运行结果:【将程序重复执行三次,三个消费者都收到了同样的消息】
2、交换机之关键字
生产者模式:
示例代码: 【将info分别改为warning、error运行】
import pika
# 1.连接rabbit
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.124.104'))
channel = connection.channel()
# 2.声明一个名为logs类型的direct的交换机
channel.exchange_declare(exchange='logs2', exchange_type='direct') # 发布订阅模式参数
# 3.向logs交换机中插入数据:"Hello world"
message = 'info:Hello World!'
channel.basic_publish(exchange='logs2',
routing_key='info', # info信息
body=message,
properties=pika.BasicProperties(
delivery_mode=2, # make message persistent
)
)
print(" [x] Sent 'Hello World!'")
运行结果:
消费者模式:
示例代码1:
import pika
# 1.连接rabbit
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.124.104'))
channel = connection.channel()
# 2.声明一个名为logs类型的direct的交换机
channel.exchange_declare(exchange='logs2', exchange_type='direct')
# 3.创建队列
result = channel.queue_declare("", exclusive=True) # 随机生成一个队列名
queue_name = result.method.queue
print(queue_name)
# 4.将指定队列绑定到交换机上
channel.queue_bind(exchange='logs2', queue=queue_name, routing_key='info')
channel.queue_bind(exchange='logs2', queue=queue_name, routing_key='waring')
channel.queue_bind(exchange='logs2', queue=queue_name, routing_key='error')
# # 使用for循环将指定队列绑定到交换机上
# for key in ['info', 'waring', 'error']:
# channel.queue_bind(exchange='logs2', queue=queue_name, routing_key=key)
# 5.确定回调函数
def callback(ch, method, properties, body):
print(" [x] Received %r" % body)
ch.basic_ack(delivery_tag=method.delivery_tag)
# 6.确定监听队列参数
channel.basic_consume(queue=queue_name, # 指定队列
auto_ack=False, # 手动应答方式
on_message_callback=callback)
print(' [*] Waiting for messages. To exit press CTRL+C')
# 7.正式监听
channel.start_consuming()
运行结果:
示例代码2:
import pika
# 1.连接rabbit
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.124.104'))
channel = connection.channel()
# 2.声明一个名为logs类型的direct的交换机
channel.exchange_declare(exchange='logs2', exchange_type='direct')
# 3.创建队列
result = channel.queue_declare("", exclusive=True) # 随机生成一个队列名
queue_name = result.method.queue
print(queue_name)
# 4.将指定队列绑定到交换机上
channel.queue_bind(exchange='logs2', queue=queue_name, routing_key='info')
# # 使用for循环将指定队列绑定到交换机上
# for key in ['info', 'waring', 'error']:
# channel.queue_bind(exchange='logs2', queue=queue_name, routing_key=key)
# 5.确定回调函数
def callback(ch, method, properties, body):
print(" [x] Received %r" % body)
ch.basic_ack(delivery_tag=method.delivery_tag)
# 6.确定监听队列参数
channel.basic_consume(queue=queue_name, # 指定队列
auto_ack=False, # 手动应答方式
on_message_callback=callback)
print(' [*] Waiting for messages. To exit press CTRL+C')
# 7.正式监听
channel.start_consuming()
运行结果:
示例代码3:
import pika
# 1.连接rabbit
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.124.104'))
channel = connection.channel()
# 2.声明一个名为logs类型的direct的交换机
channel.exchange_declare(exchange='logs2', exchange_type='direct')
# 3.创建队列
result = channel.queue_declare("", exclusive=True) # 随机生成一个队列名
queue_name = result.method.queue
print(queue_name)
# 4.将指定队列绑定到交换机上
channel.queue_bind(exchange='logs2', queue=queue_name, routing_key='error')
# # 使用for循环将指定队列绑定到交换机上
# for key in ['info', 'waring', 'error']:
# channel.queue_bind(exchange='logs2', queue=queue_name, routing_key=key)
# 5.确定回调函数
def callback(ch, method, properties, body):
print(" [x] Received %r" % body)
ch.basic_ack(delivery_tag=method.delivery_tag)
# 6.确定监听队列参数
channel.basic_consume(queue=queue_name, # 指定队列
auto_ack=False, # 手动应答方式
on_message_callback=callback)
print(' [*] Waiting for messages. To exit press CTRL+C')
# 7.正式监听
channel.start_consuming()
运行结果:
3、交换机之通配符
通配符交换机”与之前的路由模式相比,它将信息的传输类型的key更加细化,以“key1.key2.keyN....”的模式来指定信息传输的key的大类型和大类型下面的小类型,让消费者可以更加精细的确认自己想要获取的信息类型。而在消费者一段,不用精确的指定具体到哪一个大类型下的小类型的key,而是可以使用类似正则表达式(但与正则表达式规则完全不同)的通配符在指定一定范围或符合某一个字符串匹配规则的key,来获取想要的信息。
“通配符交换机”(Topic Exchange)将路由键和某模式进行匹配。此时队列需要绑定在一个模式上。符号“#”匹配一个或多个词,符号“*”仅匹配一个词。因此“audit.#”能够匹配到“audit.irs.corporate”,但是“audit.*”只会匹配到“audit.irs”。(这里与一般的正则表达式的“*”和“#”刚好相反,这里我们需要注意一下。)
生产者模式:
示例代码: 【分别将routing_key改为usa.news、news.usa和usa.weather执行一遍】
import pika
# 1.连接rabbit
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.124.104'))
channel = connection.channel()
# 2.声明一个名为logs类型的topic的交换机
channel.exchange_declare(exchange='logs3', exchange_type='topic') # 发布订阅模式参数
# 3.向logs交换机中插入数据:"Hello world"
message = 'usa.news---------'
channel.basic_publish(exchange='logs3',
routing_key='usa.news', # usa.news
body=message,
properties=pika.BasicProperties(
delivery_mode=2, # make message persistent
)
)
print(" [x] Sent 'Hello World!'")
运行结果:
消费者模式:
示例代码1:
import pika
# 1.连接rabbit
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.124.104'))
channel = connection.channel()
# 2.声明一个名为logs类型的topic的交换机
channel.exchange_declare(exchange='logs3', exchange_type='topic')
# 3.创建队列
result = channel.queue_declare("", exclusive=True) # 随机生成一个队列名
queue_name = result.method.queue
print(queue_name)
# 4.将指定队列绑定到交换机上
channel.queue_bind(exchange='logs3', queue=queue_name, routing_key='news.#')
# # 使用for循环将指定队列绑定到交换机上
# for key in ['info.#', 'waring.#', 'error.#']:
# channel.queue_bind(exchange='logs2', queue=queue_name, routing_key=key)
# 5.确定回调函数
def callback(ch, method, properties, body):
print(" [x] Received %r" % body)
ch.basic_ack(delivery_tag=method.delivery_tag)
# 6.确定监听队列参数
channel.basic_consume(queue=queue_name, # 指定队列
auto_ack=False, # 手动应答方式
on_message_callback=callback)
print(' [*] Waiting for messages. To exit press CTRL+C')
# 7.正式监听
channel.start_consuming()
运行结果:
示例代码2:
import pika
# 1.连接rabbit
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.124.104'))
channel = connection.channel()
# 2.声明一个名为logs类型的topic的交换机
channel.exchange_declare(exchange='logs3', exchange_type='topic')
# 3.创建队列
result = channel.queue_declare("", exclusive=True) # 随机生成一个队列名
queue_name = result.method.queue
print(queue_name)
# 4.将指定队列绑定到交换机上
channel.queue_bind(exchange='logs3', queue=queue_name, routing_key='#.news')
# # 使用for循环将指定队列绑定到交换机上
# for key in ['info.#', 'waring.#', 'error.#']:
# channel.queue_bind(exchange='logs2', queue=queue_name, routing_key=key)
# 5.确定回调函数
def callback(ch, method, properties, body):
print(" [x] Received %r" % body)
ch.basic_ack(delivery_tag=method.delivery_tag)
# 6.确定监听队列参数
channel.basic_consume(queue=queue_name, # 指定队列
auto_ack=False, # 手动应答方式
on_message_callback=callback)
print(' [*] Waiting for messages. To exit press CTRL+C')
# 7.正式监听
channel.start_consuming()
运行结果:
示例代码3:
import pika
# 1.连接rabbit
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.124.104'))
channel = connection.channel()
# 2.声明一个名为logs类型的topic的交换机
channel.exchange_declare(exchange='logs3', exchange_type='topic')
# 3.创建队列
result = channel.queue_declare("", exclusive=True) # 随机生成一个队列名
queue_name = result.method.queue
print(queue_name)
# 4.将指定队列绑定到交换机上
channel.queue_bind(exchange='logs3', queue=queue_name, routing_key='#.weather')
# # 使用for循环将指定队列绑定到交换机上
# for key in ['info.#', 'waring.#', 'error.#']:
# channel.queue_bind(exchange='logs2', queue=queue_name, routing_key=key)
# 5.确定回调函数
def callback(ch, method, properties, body):
print(" [x] Received %r" % body)
ch.basic_ack(delivery_tag=method.delivery_tag)
# 6.确定监听队列参数
channel.basic_consume(queue=queue_name, # 指定队列
auto_ack=False, # 手动应答方式
on_message_callback=callback)
print(' [*] Waiting for messages. To exit press CTRL+C')
# 7.正式监听
channel.start_consuming()
运行结果:
来源:https://blog.csdn.net/weixin_44799217/article/details/126533663