首页 > 资讯 > 后端开发 > Python >python使用pika库调用rabbitmq交换机模式详解

328

分享到

python使用pika库调用rabbitmq交换机模式详解

2024-04-02 19:04:59 328人浏览安东尼

Python 官方文档：入门教程 => 点击学习

摘要

目录1、交换机之发布订阅2、交换机之关键字3、交换机之通配符前言：交换机模式主要包括：交换机之发布订阅、交换机之关键字和交换机之通配符。 1、交换机之发布订阅发布订

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()

运行结果：

到此这篇关于python使用pika库调用rabbitMQ交换机模式详解的文章就介绍到这了,更多相关Python rabbitmq交换机模式内容请搜索编程网以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程网！

您可能感兴趣的文档:

--结束END--

本文标题: python使用pika库调用rabbitmq交换机模式详解

本文链接: https://lsjlt.com/news/120294.html(转载时请注明来源链接)

有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

回答

如何调试操作系统的错误？
操作系统

2023-11-15发布

回答

操作系统中的I/O系统是如何实现的？
操作系统

2023-11-15发布

回答

如何实现操作系统的内存管理？
操作系统

2023-11-15发布

回答

什么是虚拟内存，它对操作系统有什么影响？
操作系统

2023-11-15发布

回答

ASP中的MVC架构和WebForms架构有什么区别和使用场景？
ASP.NET

2023-11-15发布

回答

ASP中的数据验证和数据校验有什么不同？
ASP.NET

2023-11-15发布

回答

ASP中的ADO对象和DAO对象有什么区别和使用方法？
ASP.NET

2023-11-15发布

回答

Node.js中的包管理器NPM是什么？如何使用它进行依赖管理？
node.js

2023-11-15发布

回答

Vue.js中的动态组件是什么？如何使用它来动态渲染组件？
VUE

2023-11-15发布

回答

如何使用Vue.js实现懒加载和预加载？
VUE

2023-11-15发布

python使用pika库调用rabbitmq交换机模式详解

目录

1、交换机之发布订阅

2、交换机之关键字

3、交换机之通配符

python使用pika库调用rabbitmq交换机模式详解

python使用pika库调用rabbitmq参数使用详情

怎么使用Python交互模式

RabbitMQ交换机使用场景和消息可靠性总结分析

Python使用Matplotlib库创建3D图形和交互式图形详解

docker启动rabbitmq以及使用方式详解

Python标准库time使用方式详解

Java开发JUC交换器Exchanger使用详解

详解利用python-highcharts库绘制交互式可视化图表

Python标准库之Math,Random模块使用详解

详解Python里使用正则表达式的ASCII模式

Python机器学习库scikit-learn使用详解

使用python调用zxing库生成二维码图片详解

laravel中repository模式使用详解

Python神器之Pampy模式匹配库的用法详解

Pytho的HTTP交互httpx包模块使用详解

Android WebView使用方法详解附js交互调用方法

python正则表达式(re模块)的使用详解

python使用requests库提交multipart/form-data请求的方法详解

python urllib库的使用详解

python分析数据的方法是什么

如何使用Python实现抽奖小程序

python copy函数的作用是什么

python ffmpeg模块怎么安装和使用

python进程池创建队列的方法是什么

python无法运行文件的原因有哪些

python can't open file报错怎么解决

python keyerror错误怎么解决

python字符串处理与应用的方法有哪些

python全局变量如何定义