利用Python操作消息隊(duì)列RabbitMQ的方法教程
前言
RabbitMQ是一個(gè)在AMQP基礎(chǔ)上完整的,可復(fù)用的企業(yè)消息系統(tǒng)。他遵循Mozilla Public License開源協(xié)議。
MQ全稱為Message Queue, 消息隊(duì)列(MQ)是一種應(yīng)用程序?qū)?yīng)用程序的通信方法。應(yīng)用程序通過讀寫出入隊(duì)列的消息(針對應(yīng)用程序的數(shù)據(jù))來通信,而無需專用連接來鏈接它們。消 息傳遞指的是程序之間通過在消息中發(fā)送數(shù)據(jù)進(jìn)行通信,而不是通過直接調(diào)用彼此來通信,直接調(diào)用通常是用于諸如遠(yuǎn)程過程調(diào)用的技術(shù)。排隊(duì)指的是應(yīng)用程序通過 隊(duì)列來通信。隊(duì)列的使用除去了接收和發(fā)送應(yīng)用程序同時(shí)執(zhí)行的要求。
應(yīng)用場景:
RabbitMQ無疑是目前最流行的消息隊(duì)列之一,對各種語言環(huán)境的支持也很豐富,作為一個(gè).NET developer有必要學(xué)習(xí)和了解這一工具。消息隊(duì)列的使用場景大概有3種:
1、系統(tǒng)集成,分布式系統(tǒng)的設(shè)計(jì)。各種子系統(tǒng)通過消息來對接,這種解決方案也逐步發(fā)展成一種架構(gòu)風(fēng)格,即“通過消息傳遞的架構(gòu)”。
2、當(dāng)系統(tǒng)中的同步處理方式嚴(yán)重影響了吞吐量,比如日志記錄。假如需要記錄系統(tǒng)中所有的用戶行為日志,如果通過同步的方式記錄日志勢必會(huì)影響系統(tǒng)的響應(yīng)速度,當(dāng)我們將日志消息發(fā)送到消息隊(duì)列,記錄日志的子系統(tǒng)就會(huì)通過異步的方式去消費(fèi)日志消息。
3、系統(tǒng)的高可用性,比如電商的秒殺場景。當(dāng)某一時(shí)刻應(yīng)用服務(wù)器或數(shù)據(jù)庫服務(wù)器收到大量請求,將會(huì)出現(xiàn)系統(tǒng)宕機(jī)。如果能夠?qū)⒄埱筠D(zhuǎn)發(fā)到消息隊(duì)列,再由服務(wù)器去消費(fèi)這些消息將會(huì)使得請求變得平穩(wěn),提高系統(tǒng)的可用性。
一、安裝環(huán)境
首先是在 Linux 上安裝 rabbitmq
# 環(huán)境為CentOS 7 yum install rabbitmq-server # 安裝RabbitMQ systemctl start rabbitmq-server # 啟動(dòng) systemctl enable rabbitmq-server # 開機(jī)自啟 systemctl stop firewall-cmd # 臨時(shí)關(guān)閉防火墻
然后用 pip 安裝 Python3 的開發(fā)包
pip3 install pika
安裝好軟件之后可以訪問http://115.xx.xx.xx:15672/來訪問自帶的 web 頁面來查看和管理 RabbitMQ。默認(rèn)管理員的用戶密碼都是guest
二、簡單的向隊(duì)列中加入消息
#!/usr/bin/env python3 # coding=utf-8 # @Time : 2017/6/13 19:25 # @Author : Shawn # @Blog : https://blog.just666.cn # @Email : shawnbluce@gmail.com # @purpose : RabbitMQ_Producer import pika # 創(chuàng)建連接對象 connection = pika.BlockingConnection(pika.ConnectionParameters(host='115.xx.xx.xx')) # 創(chuàng)建頻道對象 channel = connection.channel() # 指定一個(gè)隊(duì)列,如果該隊(duì)列不存在則創(chuàng)建 channel.queue_declare(queue='test_queue') # 提交消息 for i in range(10): channel.basic_publish(exchange='', routing_key='test_queue', body='hello,world' + str(i)) print("sent...") # 關(guān)閉連接 connection.close()
三、簡單的從隊(duì)列中獲取消息
#!/usr/bin/env python3 # coding=utf-8 # @Time : 2017/6/13 19:40 # @Author : Shawn # @Blog : https://blog.just666.cn # @Email : shawnbluce@gmail.com # @purpose : RabbitMQ_Consumer import pika credentials = pika.PlainCredentials('guest', 'guest') # 連接到RabbitMQ服務(wù)器 connection = pika.BlockingConnection(pika.ConnectionParameters('115.xx.xx.xx', 5672, '/', credentials)) channel = connection.channel() # 指定一個(gè)隊(duì)列,如果該隊(duì)列不存在則創(chuàng)建 channel.queue_declare(queue='test_queue') # 定義一個(gè)回調(diào)函數(shù) def callback(ch, method, properties, body): print(body.decode('utf-8')) # 告訴RabbitMQ使用callback來接收信息 channel.basic_consume(callback, queue='test_queue', no_ack=False) print('waiting...') # 開始接收信息,并進(jìn)入阻塞狀態(tài),隊(duì)列里有信息才會(huì)調(diào)用callback進(jìn)行處理。按ctrl+c退出。 channel.start_consuming()
四、萬一消費(fèi)者掉線了
想象這樣一種情況:
消費(fèi)者從消息隊(duì)列中獲取了 n 條數(shù)據(jù),正要處理呢結(jié)果宕機(jī)了,那該怎么辦?在 RabbieMQ 中有一個(gè) ACK 可以用來確認(rèn)消費(fèi)者處理結(jié)束。就有點(diǎn)類似網(wǎng)絡(luò)中的 ACK,消費(fèi)者每次從隊(duì)列中獲取了數(shù)據(jù)之后隊(duì)列不會(huì)立刻將數(shù)據(jù)移除,而是等待對應(yīng)的 ACK。消費(fèi)者獲取到數(shù)據(jù)并處理完成之后會(huì)向隊(duì)列發(fā)送一個(gè) ACK 包,通知 RabbitMQ 這堆消息已經(jīng)處理妥當(dāng)了,可以刪除了,這時(shí)候 RabbitMQ 才會(huì)將數(shù)據(jù)從隊(duì)列中移除。所以這種情況下即使消費(fèi)者掉線也沒有什么問題,數(shù)據(jù)依舊會(huì)在隊(duì)列中存在,留給其他消費(fèi)者處理。
在 Python 中這樣實(shí)現(xiàn):
消費(fèi)者有這樣一行代碼channel.basic_consume(callback, queue='test_queue', no_ack=False)
,其中no_ack=False
表示不發(fā)送確認(rèn)包。將其修改為no_ack=True就會(huì)在每次處理完之后向 RabbitMQ 發(fā)送一個(gè)確認(rèn)包,以確認(rèn)消息處理完畢。
五、萬一 RabbitMQ 宕機(jī)了呢
雖然有了 ACK 包,但是萬一 RabbitMQ 掛了那數(shù)據(jù)還是會(huì)損失。所以我們可以給 RabbitMQ 設(shè)置一個(gè)數(shù)據(jù)持久化存儲。RabbitMQ 會(huì)將數(shù)據(jù)持久化存儲到磁盤上,保證下次再啟動(dòng)的時(shí)候隊(duì)列還在。
在 Python 中這樣實(shí)現(xiàn):
我們聲明一個(gè)隊(duì)列是這樣的channel.queue_declare(queue='test_queue')
,如果需要持久化一個(gè)隊(duì)列可以這樣聲明channel.queue_declare(queue='test_queue', durable=True)
。不過這行直接放在代碼中是不能執(zhí)行的,因?yàn)橐郧耙呀?jīng)有了一個(gè)名為test_queue的隊(duì)列,RabbitMQ 不允許用不同的方式聲明同一個(gè)隊(duì)列,所以可以換一個(gè)隊(duì)列名新建來指定數(shù)據(jù)持久化存儲。不過如果只是這樣聲明的話,在 RabbitMQ 宕機(jī)重啟后確實(shí)隊(duì)列還在,不過隊(duì)列里的數(shù)據(jù)就沒有了。除非我們這樣來聲明隊(duì)列channel.basic_publish(exchange='', routing_key="test_queue", body=message, properties=pika.BasicProperties(delivery_mode = 2,))
。
六、最簡單的發(fā)布訂閱
最簡單的發(fā)布訂閱在 RabbitMQ 中稱之為Fanout模式。也就是說訂閱者訂閱某個(gè)頻道,然后發(fā)布者向這個(gè)頻道中發(fā)布消息,所有訂閱者就都能接收到這條消息。不過因?yàn)榘l(fā)布者需要使用訂閱者創(chuàng)建的隨機(jī)隊(duì)列所以需要先啟動(dòng)訂閱者才能啟動(dòng)發(fā)布者。
發(fā)布者代碼:
#!/usr/bin/env python3 # coding=utf-8 # @Time : 2017/6/13 20:21 # @Author : Shawn # @Blog : https://blog.just666.cn # @Email : shawnbluce@gmail.com # @purpose : RabbitMQ_Publisher import pika # 創(chuàng)建連接對象 connection = pika.BlockingConnection(pika.ConnectionParameters(host='115.xx.xx.xx')) # 創(chuàng)建頻道對象 channel = connection.channel() # 定義交換機(jī),exchange表示交換機(jī)名稱,type表示類型 channel.exchange_declare(exchange='my_fanout', type='fanout') message = 'Hello Python' # 將消息發(fā)送到交換機(jī) channel.basic_publish(exchange='my_fanout', # 指定exchange routing_key='', # fanout下不需要配置,配置了也不會(huì)生效 body=message) connection.close()
訂閱者代碼:
#!/usr/bin/env python3 # coding=utf-8 # @Time : 2017/6/13 20:20 # @Author : Shawn # @Blog : https://blog.just666.cn # @Email : shawnbluce@gmail.com # @purpose : RabbitMQ_Subscriber import pika credentials = pika.PlainCredentials('guest', 'guest') # 連接到RabbitMQ connection = pika.BlockingConnection(pika.ConnectionParameters('115.xx.xx.xx', 5672, '/', credentials)) channel = connection.channel() # 定義交換機(jī),進(jìn)行exchange聲明,exchange表示交換機(jī)名稱,type表示類型 channel.exchange_declare(exchange='my_fanout', type='fanout') # 隨機(jī)創(chuàng)建隊(duì)列 result = channel.queue_declare(exclusive=True) # exclusive=True表示建立臨時(shí)隊(duì)列,當(dāng)consumer關(guān)閉后,該隊(duì)列就會(huì)被刪除 queue_name = result.method.queue # 將隊(duì)列與exchange進(jìn)行綁定 channel.queue_bind(exchange='my_fanout', queue=queue_name) # 定義回調(diào)方法 def callback(ch, method, properties, body): print(body.decode('utf-8')) # 從隊(duì)列獲取信息 channel.basic_consume(callback, queue=queue_name, no_ack=True) channel.start_consuming()
總結(jié)
以上就是這篇文章的全部內(nèi)容,希望本文的內(nèi)容對大家的學(xué)習(xí)或者工作能帶來一定的幫助,如果有疑問大家可以留言交流,謝謝大家對腳本之家的支持。
相關(guān)文章
Python自動(dòng)化辦公之定時(shí)發(fā)送郵件的實(shí)現(xiàn)
python中的schedule模塊可以使我們方便簡單的使用定時(shí)任務(wù),即在特定的時(shí)間自動(dòng)的執(zhí)行一些任務(wù)的功能,本文將用這一模塊實(shí)現(xiàn)郵件自動(dòng)發(fā)送,需要的可以參考一下2022-05-05Python&Matlab實(shí)現(xiàn)伏羲八卦圖的繪制
八卦最先由伏羲根據(jù)燧人氏造設(shè)的兩幅星圖歷法《河圖洛書》創(chuàng)設(shè)。本文將通過Python和Matlab分別繪制伏羲八卦圖,感興趣的小伙伴可以了解一下2022-03-0310個(gè)使用Python必須知道的內(nèi)置函數(shù)
這篇文章小編主要向大家介紹的是10個(gè)使用Python必須知道的內(nèi)置函數(shù)reduce()、split()、map()等,更多后置函數(shù)請看下文2021-09-09Python實(shí)現(xiàn)身份證號碼驗(yàn)證的示例代碼
本文主要介紹了Python實(shí)現(xiàn)身份證號碼驗(yàn)證的示例代碼,當(dāng)用戶輸入身份證號,按下檢查按鈕,即可判斷身份證號是否正確,具有一定的參考價(jià)值,感興趣的可以了解一下2024-02-02