網站首頁 編程語言 正文
前言：
交換機模式主要包括：交換機之發布訂閱、交換機之關鍵字和交換機之通配符。
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
相關推薦
- 2022-11-06 python分析inkscape路徑數據方案簡單介紹_python
- 2022-01-19 正則——用戶名和密碼校驗、數字、大小寫字母、數字和字母
- 2022-05-27 詳解Python實現字典合并的四種方法_python
- 2022-05-11 如何使 React 中的 useEffect、useLayoutEffect 只調用一次
- 2022-07-07 一篇文章讀懂nginx的gzip功能_nginx
- 2022-07-10 css中border屬性設置
- 2022-06-16 Go基礎教程系列之defer、panic和recover詳解_Golang
- 2023-02-07 詳解C#如何實現窗體換膚_C#教程
- 最近更新
-
- window11 系統安裝 yarn
- 超詳細win安裝深度學習環境2025年最新版（
- Linux 中運行的top命令 怎么退出？
- MySQL 中decimal 的用法？ 存儲小
- get 、set 、toString 方法的使
- @Resource和 @Autowired注解
- Java基礎操作-- 運算符，流程控制 Flo
- 1. Int 和Integer 的區別，Jav
- spring @retryable不生效的一種
- Spring Security之認證信息的處理
- Spring Security之認證過濾器
- Spring Security概述快速入門
- Spring Security之配置體系
- 【SpringBoot】SpringCache
- Spring Security之基于方法配置權
- redisson分布式鎖中waittime的設
- maven:解決release錯誤：Artif
- restTemplate使用總結
- Spring Security之安全異常處理
- MybatisPlus優雅實現加密？
- Spring ioc容器與Bean的生命周期。
- 【探索SpringCloud】服務發現-Nac
- Spring Security之基于HttpR
- Redis 底層數據結構-簡單動態字符串（SD
- arthas操作spring被代理目標對象命令
- Spring中的單例模式應用詳解
- 聊聊消息隊列，發送消息的4種方式
- bootspring第三方資源配置管理
- GIT同步修改后的遠程分支
提供CDN加速