python操作RabbitMq的三種工作模式

更新時間：2022年04月12日 11:54:38 作者：Jeff的技術棧

這篇文章主要為大家介紹了python操作RabbitMq的三種工作模式，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步早日升職加薪

一、簡介：

RabbitMq 是實現(xiàn)了高級消息隊列協(xié)議（AMQP）的開源消息代理中間件。消息隊列是一種應用程序?qū)贸绦虻耐ㄐ蟹绞?，應用程序通過寫消息，將消息傳遞于隊列，由另一應用程序讀取完成通信。而作為中間件的 RabbitMq 無疑是目前最流行的消息隊列之一。

RabbitMq 應用場景廣泛：

系統(tǒng)的高可用：日常生活當中各種商城秒殺，高流量，高并發(fā)的場景。當服務器接收到如此大量請求處理業(yè)務時，有宕機的風險。某些業(yè)務可能極其復雜，但這部分不是高時效性，不需要立即反饋給用戶，我們可以將這部分處理請求拋給隊列，讓程序后置去處理，減輕服務器在高并發(fā)場景下的壓力。
分布式系統(tǒng)，集成系統(tǒng)，子系統(tǒng)之間的對接，以及架構(gòu)設計中常常需要考慮消息隊列的應用。

二、RabbitMq 生產(chǎn)和消費

生產(chǎn)者（producter）：隊列消息的產(chǎn)生者，負責生產(chǎn)消息，并將消息傳入隊列

import pika
import json
credentials = pika.PlainCredentials('shampoo', '123456')  # mq用戶名和密碼
# 虛擬隊列需要指定參數(shù) virtual_host，如果是默認的可以不填。
connection = pika.BlockingConnection(pika.ConnectionParameters(host = '10.1.62.170',port = 5672,virtual_host = '/',credentials = credentials))
channel=connection.channel()
# 聲明消息隊列，消息將在這個隊列傳遞，如不存在，則創(chuàng)建
result = channel.queue_declare(queue = 'python-test')
for i in range(10):
    message=json.dumps({'OrderId':"1000%s"%i})
# 向隊列插入數(shù)值 routing_key是隊列名
    channel.basic_publish(exchange = '',routing_key = 'python-test',body = message)
    print(message)
connection.close()

消費者（consumer）：隊列消息的接收者，負責接收并處理消息隊列中的消息

import pika
credentials = pika.PlainCredentials('shampoo', '123456')
connection = pika.BlockingConnection(pika.ConnectionParameters(host = '10.1.62.170',port = 5672,virtual_host = '/',credentials = credentials))
channel = connection.channel()
# 申明消息隊列，消息在這個隊列傳遞，如果不存在，則創(chuàng)建隊列
channel.queue_declare(queue = 'python-test', durable = False)
# 定義一個回調(diào)函數(shù)來處理消息隊列中的消息，這里是打印出來
def callback(ch, method, properties, body):
    ch.basic_ack(delivery_tag = method.delivery_tag)
    print(body.decode())
# 告訴rabbitmq，用callback來接收消息
channel.basic_consume('python-test',callback)
# 開始接收信息，并進入阻塞狀態(tài)，隊列里有信息才會調(diào)用callback進行處理
channel.start_consuming()

三、RabbitMq 持久化

MQ默認建立的是臨時 queue 和 exchange，如果不聲明持久化，一旦 rabbitmq 掛掉，queue、exchange 將會全部丟失。所以我們一般在創(chuàng)建 queue 或者 exchange 的時候會聲明持久化。

1.queue 聲明持久化

# 聲明消息隊列，消息將在這個隊列傳遞，如不存在，則創(chuàng)建。durable = True 代表消息隊列持久化存儲，F(xiàn)alse 非持久化存儲
result = channel.queue_declare(queue = 'python-test',durable = True)

2.exchange 聲明持久化

# 聲明exchange，由exchange指定消息在哪個隊列傳遞，如不存在，則創(chuàng)建.durable = True 代表exchange持久化存儲，F(xiàn)alse 非持久化存儲
channel.exchange_declare(exchange = 'python-test', durable = True)

注意：如果已存在一個非持久化的 queue 或 exchange ，執(zhí)行上述代碼會報錯，因為當前狀態(tài)不能更改 queue 或 exchange 存儲屬性，需要刪除重建。如果 queue 和 exchange 中一個聲明了持久化，另一個沒有聲明持久化，則不允許綁定。

3.消息持久化

雖然 exchange 和 queue 都申明了持久化，但如果消息只存在內(nèi)存里，rabbitmq 重啟后，內(nèi)存里的東西還是會丟失。所以必須聲明消息也是持久化，從內(nèi)存轉(zhuǎn)存到硬盤。

# 向隊列插入數(shù)值 routing_key是隊列名。delivery_mode = 2 聲明消息在隊列中持久化，delivery_mod = 1 消息非持久化
    channel.basic_publish(exchange = '',routing_key = 'python-test',body = message,
                          properties=pika.BasicProperties(delivery_mode = 2))

4.acknowledgement 消息不丟失

消費者（consumer）調(diào)用callback函數(shù)時，會存在處理消息失敗的風險，如果處理失敗，則消息丟失。但是也可以選擇消費者處理失敗時，將消息回退給 rabbitmq ，重新再被消費者消費，這個時候需要設置確認標識。

channel.basic_consume(callback,queue = 'python-test',
# no_ack 設置成 False，在調(diào)用callback函數(shù)時，未收到確認標識，消息會重回隊列。True，無論調(diào)用callback成功與否，消息都被消費掉
                      no_ack = False)

四、RabbitMq 發(fā)布與訂閱

rabbitmq 的發(fā)布與訂閱要借助交換機（Exchange）的原理實現(xiàn)：

Exchange 一共有三種工作模式：fanout, direct, topicd

模式一：fanout

這種模式下，傳遞到 exchange 的消息將會轉(zhuǎn)發(fā)到所有與其綁定的 queue 上。

不需要指定 routing_key ，即使指定了也是無效。
需要提前將 exchange 和 queue 綁定，一個 exchange 可以綁定多個 queue，一個queue可以綁定多個exchange。
需要先啟動訂閱者，此模式下的隊列是 consumer 隨機生成的，發(fā)布者僅僅發(fā)布消息到 exchange ，由 exchange 轉(zhuǎn)發(fā)消息至 queue。

發(fā)布者：

import pika
import json
credentials = pika.PlainCredentials('shampoo', '123456')  # mq用戶名和密碼
# 虛擬隊列需要指定參數(shù) virtual_host，如果是默認的可以不填。
connection = pika.BlockingConnection(pika.ConnectionParameters(host = '10.1.62.170',port = 5672,virtual_host = '/',credentials = credentials))
channel=connection.channel()
# 聲明exchange，由exchange指定消息在哪個隊列傳遞，如不存在，則創(chuàng)建。durable = True 代表exchange持久化存儲，F(xiàn)alse 非持久化存儲
channel.exchange_declare(exchange = 'python-test',durable = True, exchange_type='fanout')
for i in range(10):
    message=json.dumps({'OrderId':"1000%s"%i})
# 向隊列插入數(shù)值 routing_key是隊列名。delivery_mode = 2 聲明消息在隊列中持久化，delivery_mod = 1 消息非持久化。routing_key 不需要配置
    channel.basic_publish(exchange = 'python-test',routing_key = '',body = message,
                          properties=pika.BasicProperties(delivery_mode = 2))
    print(message)
connection.close()

訂閱者：

import pika
credentials = pika.PlainCredentials('shampoo', '123456')
connection = pika.BlockingConnection(pika.ConnectionParameters(host = '10.1.62.170',port = 5672,virtual_host = '/',credentials = credentials))
channel = connection.channel()
# 創(chuàng)建臨時隊列,隊列名傳空字符，consumer關閉后，隊列自動刪除
result = channel.queue_declare('',exclusive=True)
# 聲明exchange，由exchange指定消息在哪個隊列傳遞，如不存在，則創(chuàng)建。durable = True 代表exchange持久化存儲，F(xiàn)alse 非持久化存儲
channel.exchange_declare(exchange = 'python-test',durable = True, exchange_type='fanout')
# 綁定exchange和隊列  exchange 使我們能夠確切地指定消息應該到哪個隊列去
channel.queue_bind(exchange = 'python-test',queue = result.method.queue)
# 定義一個回調(diào)函數(shù)來處理消息隊列中的消息，這里是打印出來
def callback(ch, method, properties, body):
    ch.basic_ack(delivery_tag = method.delivery_tag)
    print(body.decode())
channel.basic_consume(result.method.queue,callback,# 設置成 False，在調(diào)用callback函數(shù)時，未收到確認標識，消息會重回隊列。True，無論調(diào)用callback成功與否，消息都被消費掉
                      auto_ack = False)
channel.start_consuming()

模式二：direct

這種工作模式的原理是消息發(fā)送至 exchange，exchange 根據(jù) 路由鍵（routing_key）轉(zhuǎn)發(fā)到相對應的 queue 上。

可以使用默認 exchange =' ' ，也可以自定義 exchange
這種模式下不需要將 exchange 和任何進行綁定，當然綁定也是可以的。可以將 exchange 和 queue ，routing_key 和 queue 進行綁定
傳遞或接受消息時需要指定 routing_key
需要先啟動訂閱者，此模式下的隊列是 consumer 隨機生成的，發(fā)布者僅僅發(fā)布消息到 exchange ，由 exchange 轉(zhuǎn)發(fā)消息至 queue。

發(fā)布者：

import pika
import json
credentials = pika.PlainCredentials('shampoo', '123456')  # mq用戶名和密碼
# 虛擬隊列需要指定參數(shù) virtual_host，如果是默認的可以不填。
connection = pika.BlockingConnection(pika.ConnectionParameters(host = '10.1.62.170',port = 5672,virtual_host = '/',credentials = credentials))
channel=connection.channel()
# 聲明exchange，由exchange指定消息在哪個隊列傳遞，如不存在，則創(chuàng)建。durable = True 代表exchange持久化存儲，F(xiàn)alse 非持久化存儲
channel.exchange_declare(exchange = 'python-test',durable = True, exchange_type='direct')
for i in range(10):
    message=json.dumps({'OrderId':"1000%s"%i})
# 指定 routing_key。delivery_mode = 2 聲明消息在隊列中持久化，delivery_mod = 1 消息非持久化
    channel.basic_publish(exchange = 'python-test',routing_key = 'OrderId',body = message,
                          properties=pika.BasicProperties(delivery_mode = 2))
    print(message)
connection.close()

消費者：

import pika
credentials = pika.PlainCredentials('shampoo', '123456')
connection = pika.BlockingConnection(pika.ConnectionParameters(host = '10.1.62.170',port = 5672,virtual_host = '/',credentials = credentials))
channel = connection.channel()
# 創(chuàng)建臨時隊列，隊列名傳空字符，consumer關閉后，隊列自動刪除
result = channel.queue_declare('',exclusive=True)
# 聲明exchange，由exchange指定消息在哪個隊列傳遞，如不存在，則創(chuàng)建。durable = True 代表exchange持久化存儲，F(xiàn)alse 非持久化存儲
channel.exchange_declare(exchange = 'python-test',durable = True, exchange_type='direct')
# 綁定exchange和隊列  exchange 使我們能夠確切地指定消息應該到哪個隊列去
channel.queue_bind(exchange = 'python-test',queue = result.method.queue,routing_key='OrderId')
# 定義一個回調(diào)函數(shù)來處理消息隊列中的消息，這里是打印出來
def callback(ch, method, properties, body):
    ch.basic_ack(delivery_tag = method.delivery_tag)
    print(body.decode())
#channel.basic_qos(prefetch_count=1)
# 告訴rabbitmq，用callback來接受消息
channel.basic_consume(result.method.queue,callback,
# 設置成 False，在調(diào)用callback函數(shù)時，未收到確認標識，消息會重回隊列。True，無論調(diào)用callback成功與否，消息都被消費掉
                      auto_ack = False)
channel.start_consuming()

模式三：topicd

這種模式和第二種模式差不多，exchange 也是通過路由鍵 routing_key 來轉(zhuǎn)發(fā)消息到指定的 queue 。不同點是 routing_key 使用正則表達式支持模糊匹配，但匹配規(guī)則又與常規(guī)的正則表達式不同，比如“#”是匹配全部，“*”是匹配一個詞。

舉例：routing_key =“#orderid#”，意思是將消息轉(zhuǎn)發(fā)至所有 routing_key 包含 “orderid” 字符的隊列中。代碼和模式二類似，就不貼出來了。

以上就是python操作RabbitMq的三種工作模式的詳細內(nèi)容，更多關于python操作RabbitMq工作模式的資料請關注腳本之家其它相關文章！

您可能感興趣的文章:

Python實現(xiàn)將通信達.day文件讀取為DataFrame
今天小編就為大家分享一篇Python實現(xiàn)將通信達.day文件讀取為DataFrame，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧
2018-12-12
OpenCV實現(xiàn)對象跟蹤的方法
OpenCV 是一個很好的處理圖像和視頻的工具，本文主要介紹了OpenCV 進行對象跟蹤，文中通過示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下
2021-10-10
深入理解python中sort()與sorted()的區(qū)別
Python list內(nèi)置sort()方法用來排序，也可以用python內(nèi)置的全局sorted()方法來對可迭代的序列排序生成新的序列。這篇文章主要介紹了python中sort()與sorted()的區(qū)別,需要的朋友可以參考下
2018-08-08
使用Python將PDF文件轉(zhuǎn)存為圖片的代碼示例
因工作中的某些奇葩要求,需要將PDF文件的每頁內(nèi)容轉(zhuǎn)存成按順序編號的圖片,用第三方軟件或者在線轉(zhuǎn)換也可以,但批量操作還是Python方便,所以本文給大家介紹了使用Python將PDF文件轉(zhuǎn)存為圖片的方法,需要的朋友可以參考下
2023-09-09
利用python讀取YUV文件轉(zhuǎn)RGB 8bit/10bit通用
今天小編就為大家分享一篇利用python讀取YUV文件轉(zhuǎn)RGB 8bit/10bit通用，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧
2019-12-12
舉例講解Python中is和id的用法
這篇文章主要介紹了舉例講解Python中is和id的用法,是Python學習當中的基礎知識,作者舉例進行了簡單說明,需要的朋友可以參考下
2015-04-04
python?memory_profiler庫生成器和迭代器內(nèi)存占用的時間分析
這篇文章主要介紹了python?memory_profiler庫生成器和迭代器內(nèi)存占用的時間分析，文章圍繞主題展開詳細的內(nèi)容介紹，感興趣的小伙伴可以參考一下
2022-06-06
Python實現(xiàn)微信自動回復信息的功能(根據(jù)不同信息回復對應的信息)
這篇文章主要介紹了Python實現(xiàn)微信自動回復信息的功能(根據(jù)不同信息回復對應的信息),我們使用的第三方包是UIAutomation,結(jié)合示例代碼給大家介紹的非常詳細,需要的朋友可以參考下
2023-09-09
python中的iterator和"lazy?iterator"區(qū)別介紹
這篇文章主要介紹了python中的iterator和?“l(fā)azy?iterator“之間有什么區(qū)別,本文通過實例代碼給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下
2023-04-04
使用python實現(xiàn)將excel數(shù)據(jù)導入word并設置字體樣式的代碼示例
在日常辦公和數(shù)據(jù)處理中,我們經(jīng)常需要將Excel中的數(shù)據(jù)導入到Word文檔中,手動完成這個過程可能既費時又容易出錯,本文將介紹如何使用Python自動化這一任務,需要的朋友可以參考下
2024-02-02