python使用pika庫(kù)調(diào)用rabbitmq交換機(jī)模式詳解

更新時(shí)間：2022年08月31日 16:12:14 作者：IT之一小佬

這篇文章主要介紹了python使用pika庫(kù)調(diào)用rabbitmq交換機(jī)模式詳解，文章圍繞主題展開(kāi)詳細(xì)的內(nèi)容介紹，具有一定的參考價(jià)值，感興趣的小伙伴可以參考一下

1、交換機(jī)之發(fā)布訂閱

發(fā)布訂閱和簡(jiǎn)單的消息隊(duì)列區(qū)別在于，發(fā)布訂閱會(huì)將消息發(fā)送給所有的訂閱者，而消息隊(duì)列中的數(shù)據(jù)被消費(fèi)一次便消失。所以，RabbitMQ實(shí)現(xiàn)發(fā)布和訂閱時(shí)，會(huì)為每一個(gè)訂閱者創(chuàng)建一個(gè)隊(duì)列，而發(fā)布者發(fā)布消息時(shí)，會(huì)將消息放置在所有相關(guān)隊(duì)列中。

生產(chǎn)者模式：

示例代碼：

import pika
# 1.連接rabbit
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.124.104'))
channel = connection.channel()
 
# 2.聲明一個(gè)名為logs類(lèi)型的fanout的交換機(jī)
channel.exchange_declare(exchange='logs', exchange_type='fanout')  # 發(fā)布訂閱模式參數(shù)
 
# 3.向logs交換機(jī)中插入數(shù)據(jù)："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!'")

運(yùn)行結(jié)果：

消費(fèi)者模式：

示例代碼：

import pika
 
# 1.連接rabbit
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.124.104'))
channel = connection.channel()
 
# 2.聲明一個(gè)名為logs類(lèi)型的fanout的交換機(jī)
channel.exchange_declare(exchange='logs', exchange_type='fanout')
 
# 3.創(chuàng)建隊(duì)列
result = channel.queue_declare("", exclusive=True)  # 隨機(jī)生成一個(gè)隊(duì)列名
queue_name = result.method.queue
print(queue_name)
 
# 4.將指定隊(duì)列綁定到交換機(jī)上
channel.queue_bind(exchange='logs', queue=queue_name)
 
# 5.確定回調(diào)函數(shù)
def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)
    ch.basic_ack(delivery_tag=method.delivery_tag)

# 6.確定監(jiān)聽(tīng)隊(duì)列參數(shù)
channel.basic_consume(queue=queue_name,  # 指定隊(duì)列
                      auto_ack=False,  # 手動(dòng)應(yīng)答方式
                      on_message_callback=callback)
print(' [*] Waiting for messages. To exit press CTRL+C')
# 7.正式監(jiān)聽(tīng)
channel.start_consuming()

運(yùn)行結(jié)果：【將程序重復(fù)執(zhí)行三次，三個(gè)消費(fèi)者都收到了同樣的消息】

2、交換機(jī)之關(guān)鍵字

生產(chǎn)者模式：

示例代碼： 【將info分別改為warning、error運(yùn)行】

import pika
 
# 1.連接rabbit
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.124.104'))
channel = connection.channel()
 
# 2.聲明一個(gè)名為logs類(lèi)型的direct的交換機(jī)
channel.exchange_declare(exchange='logs2', exchange_type='direct')  # 發(fā)布訂閱模式參數(shù)
 
# 3.向logs交換機(jī)中插入數(shù)據(jù)："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!'")

運(yùn)行結(jié)果：

消費(fèi)者模式：

示例代碼1：

import pika
 
# 1.連接rabbit
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.124.104'))
channel = connection.channel()
 
# 2.聲明一個(gè)名為logs類(lèi)型的direct的交換機(jī)
channel.exchange_declare(exchange='logs2', exchange_type='direct')
 
# 3.創(chuàng)建隊(duì)列
result = channel.queue_declare("", exclusive=True)  # 隨機(jī)生成一個(gè)隊(duì)列名
queue_name = result.method.queue
print(queue_name)
 
# 4.將指定隊(duì)列綁定到交換機(jī)上
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循環(huán)將指定隊(duì)列綁定到交換機(jī)上
# for key in ['info', 'waring', 'error']:
#     channel.queue_bind(exchange='logs2', queue=queue_name, routing_key=key)
 
# 5.確定回調(diào)函數(shù)
def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)
    ch.basic_ack(delivery_tag=method.delivery_tag)
 
# 6.確定監(jiān)聽(tīng)隊(duì)列參數(shù)
channel.basic_consume(queue=queue_name,  # 指定隊(duì)列
                      auto_ack=False,  # 手動(dòng)應(yīng)答方式
                      on_message_callback=callback)
print(' [*] Waiting for messages. To exit press CTRL+C')
# 7.正式監(jiān)聽(tīng)
channel.start_consuming()

運(yùn)行結(jié)果：

示例代碼2：

import pika
 
# 1.連接rabbit
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.124.104'))
channel = connection.channel()
 
# 2.聲明一個(gè)名為logs類(lèi)型的direct的交換機(jī)
channel.exchange_declare(exchange='logs2', exchange_type='direct')
 
# 3.創(chuàng)建隊(duì)列
result = channel.queue_declare("", exclusive=True)  # 隨機(jī)生成一個(gè)隊(duì)列名
queue_name = result.method.queue
print(queue_name)
 
# 4.將指定隊(duì)列綁定到交換機(jī)上
channel.queue_bind(exchange='logs2', queue=queue_name, routing_key='info')
 
# # 使用for循環(huán)將指定隊(duì)列綁定到交換機(jī)上
# for key in ['info', 'waring', 'error']:
#     channel.queue_bind(exchange='logs2', queue=queue_name, routing_key=key)
# 5.確定回調(diào)函數(shù)
def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)
    ch.basic_ack(delivery_tag=method.delivery_tag)

# 6.確定監(jiān)聽(tīng)隊(duì)列參數(shù)
channel.basic_consume(queue=queue_name,  # 指定隊(duì)列
                      auto_ack=False,  # 手動(dòng)應(yīng)答方式
                      on_message_callback=callback)
print(' [*] Waiting for messages. To exit press CTRL+C')
# 7.正式監(jiān)聽(tīng)
channel.start_consuming()

運(yùn)行結(jié)果：

示例代碼3：

import pika
 
# 1.連接rabbit
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.124.104'))
channel = connection.channel()
 
# 2.聲明一個(gè)名為logs類(lèi)型的direct的交換機(jī)
channel.exchange_declare(exchange='logs2', exchange_type='direct')
 
# 3.創(chuàng)建隊(duì)列
result = channel.queue_declare("", exclusive=True)  # 隨機(jī)生成一個(gè)隊(duì)列名
queue_name = result.method.queue
print(queue_name)
 
# 4.將指定隊(duì)列綁定到交換機(jī)上
channel.queue_bind(exchange='logs2', queue=queue_name, routing_key='error')
 
# # 使用for循環(huán)將指定隊(duì)列綁定到交換機(jī)上
# for key in ['info', 'waring', 'error']:
#     channel.queue_bind(exchange='logs2', queue=queue_name, routing_key=key)
 
# 5.確定回調(diào)函數(shù)
def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)
    ch.basic_ack(delivery_tag=method.delivery_tag)
 
# 6.確定監(jiān)聽(tīng)隊(duì)列參數(shù)
channel.basic_consume(queue=queue_name,  # 指定隊(duì)列
                      auto_ack=False,  # 手動(dòng)應(yīng)答方式
                      on_message_callback=callback)
print(' [*] Waiting for messages. To exit press CTRL+C')
# 7.正式監(jiān)聽(tīng)
channel.start_consuming()

運(yùn)行結(jié)果：

3、交換機(jī)之通配符

通配符交換機(jī)”與之前的路由模式相比，它將信息的傳輸類(lèi)型的key更加細(xì)化，以“key1.key2.keyN....”的模式來(lái)指定信息傳輸?shù)膋ey的大類(lèi)型和大類(lèi)型下面的小類(lèi)型，讓消費(fèi)者可以更加精細(xì)的確認(rèn)自己想要獲取的信息類(lèi)型。而在消費(fèi)者一段，不用精確的指定具體到哪一個(gè)大類(lèi)型下的小類(lèi)型的key，而是可以使用類(lèi)似正則表達(dá)式(但與正則表達(dá)式規(guī)則完全不同)的通配符在指定一定范圍或符合某一個(gè)字符串匹配規(guī)則的key，來(lái)獲取想要的信息。

“通配符交換機(jī)”（Topic Exchange）將路由鍵和某模式進(jìn)行匹配。此時(shí)隊(duì)列需要綁定在一個(gè)模式上。符號(hào)“#”匹配一個(gè)或多個(gè)詞，符號(hào)“*”僅匹配一個(gè)詞。因此“audit.#”能夠匹配到“audit.irs.corporate”，但是“audit.*”只會(huì)匹配到“audit.irs”。（這里與一般的正則表達(dá)式的“*”和“#”剛好相反，這里我們需要注意一下。）

生產(chǎn)者模式：

示例代碼： 【分別將routing_key改為usa.news、news.usa和usa.weather執(zhí)行一遍】

import pika
 
# 1.連接rabbit
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.124.104'))
channel = connection.channel()
 
# 2.聲明一個(gè)名為logs類(lèi)型的topic的交換機(jī)
channel.exchange_declare(exchange='logs3', exchange_type='topic')  # 發(fā)布訂閱模式參數(shù)
 
# 3.向logs交換機(jī)中插入數(shù)據(jù)："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!'")

運(yùn)行結(jié)果：

消費(fèi)者模式：

示例代碼1：

import pika
 
# 1.連接rabbit
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.124.104'))
channel = connection.channel()
 
# 2.聲明一個(gè)名為logs類(lèi)型的topic的交換機(jī)
channel.exchange_declare(exchange='logs3', exchange_type='topic')
 
# 3.創(chuàng)建隊(duì)列
result = channel.queue_declare("", exclusive=True)  # 隨機(jī)生成一個(gè)隊(duì)列名
queue_name = result.method.queue
print(queue_name)
 
# 4.將指定隊(duì)列綁定到交換機(jī)上
channel.queue_bind(exchange='logs3', queue=queue_name, routing_key='news.#')
 
# # 使用for循環(huán)將指定隊(duì)列綁定到交換機(jī)上
# for key in ['info.#', 'waring.#', 'error.#']:
#     channel.queue_bind(exchange='logs2', queue=queue_name, routing_key=key)
# 5.確定回調(diào)函數(shù)
def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)
    ch.basic_ack(delivery_tag=method.delivery_tag)
 
 
# 6.確定監(jiān)聽(tīng)隊(duì)列參數(shù)
channel.basic_consume(queue=queue_name,  # 指定隊(duì)列
                      auto_ack=False,  # 手動(dòng)應(yīng)答方式
                      on_message_callback=callback)
 
print(' [*] Waiting for messages. To exit press CTRL+C')
# 7.正式監(jiān)聽(tīng)
channel.start_consuming()

運(yùn)行結(jié)果：

示例代碼2：

import pika
 
# 1.連接rabbit
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.124.104'))
channel = connection.channel()
 
# 2.聲明一個(gè)名為logs類(lèi)型的topic的交換機(jī)
channel.exchange_declare(exchange='logs3', exchange_type='topic')
 
# 3.創(chuàng)建隊(duì)列
result = channel.queue_declare("", exclusive=True)  # 隨機(jī)生成一個(gè)隊(duì)列名
queue_name = result.method.queue
print(queue_name)
 
# 4.將指定隊(duì)列綁定到交換機(jī)上
channel.queue_bind(exchange='logs3', queue=queue_name, routing_key='#.news')
 
# # 使用for循環(huán)將指定隊(duì)列綁定到交換機(jī)上
# for key in ['info.#', 'waring.#', 'error.#']:
#     channel.queue_bind(exchange='logs2', queue=queue_name, routing_key=key)

# 5.確定回調(diào)函數(shù)
def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)
    ch.basic_ack(delivery_tag=method.delivery_tag)
# 6.確定監(jiān)聽(tīng)隊(duì)列參數(shù)
channel.basic_consume(queue=queue_name,  # 指定隊(duì)列
                      auto_ack=False,  # 手動(dòng)應(yīng)答方式
                      on_message_callback=callback)
print(' [*] Waiting for messages. To exit press CTRL+C')
# 7.正式監(jiān)聽(tīng)
channel.start_consuming()

運(yùn)行結(jié)果：

示例代碼3：

import pika
 
# 1.連接rabbit
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.124.104'))
channel = connection.channel()
 
# 2.聲明一個(gè)名為logs類(lèi)型的topic的交換機(jī)
channel.exchange_declare(exchange='logs3', exchange_type='topic')
 
# 3.創(chuàng)建隊(duì)列
result = channel.queue_declare("", exclusive=True)  # 隨機(jī)生成一個(gè)隊(duì)列名
queue_name = result.method.queue
print(queue_name)
 
# 4.將指定隊(duì)列綁定到交換機(jī)上
channel.queue_bind(exchange='logs3', queue=queue_name, routing_key='#.weather')
 
# # 使用for循環(huán)將指定隊(duì)列綁定到交換機(jī)上
# for key in ['info.#', 'waring.#', 'error.#']:
#     channel.queue_bind(exchange='logs2', queue=queue_name, routing_key=key)

# 5.確定回調(diào)函數(shù)
def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)
    ch.basic_ack(delivery_tag=method.delivery_tag)
 
 
# 6.確定監(jiān)聽(tīng)隊(duì)列參數(shù)
channel.basic_consume(queue=queue_name,  # 指定隊(duì)列
                      auto_ack=False,  # 手動(dòng)應(yīng)答方式
                      on_message_callback=callback)
print(' [*] Waiting for messages. To exit press CTRL+C')
# 7.正式監(jiān)聽(tīng)
channel.start_consuming()

運(yùn)行結(jié)果：

到此這篇關(guān)于python使用pika庫(kù)調(diào)用rabbitmq交換機(jī)模式詳解的文章就介紹到這了,更多相關(guān)python rabbitmq交換機(jī)模式內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: