RabbitMQ
消息隊列:
消息（Message）是指在應用間傳送的數據。消息可以非常簡單，比如只包含文本字符串，也可以更復雜，可能包含嵌入對象。
消息隊列（Message Queue）是一種應用間的通信方式，消息發送后可以立即返回，由消息系統來確保消息的可靠傳遞。消息發布者只管把消息發布到 MQ 中而不用管誰來取，消息使用者只管從 MQ 中取消息而不管是誰發布的。這樣發布者和使用者都不用知道對方的存在。
為何用消息隊列:
從上面的描述中可以看出消息隊列是一種應用間的異步協作機制，那什么時候需要使用 MQ 呢？
以常見的訂單系統為例，用戶點擊【下單】按鈕之后的業務邏輯可能包括：扣減庫存、生成相應單據、發紅包、發短信通知。在業務發展初期這些邏輯可能放在一起同步執行，隨着業務的發展訂單量增長，需要提升系統服務的性能，這時可以將一些不需要立即生效的操作拆分出來異步執行，比如發放紅包、發短信通知等。這種場景下就可以用 MQ ，在下單的主流程（比如扣減庫存、生成相應單據）完成之后發送一條消息到 MQ 讓主流程快速完結，而由另外的單獨線程拉取MQ的消息（或者由 MQ 推送消息），當發現 MQ 中有發紅包或發短信之類的消息時，執行相應的業務邏輯。
RabbitMQ
RabbitMQ 是一個由 Erlang 語言開發的 AMQP 的開源實現。
rabbitMQ是一款基於AMQP協議的消息中間件，它能夠在應用之間提供可靠的消息傳輸。在易用性，擴展性，高可用性上表現優秀。使用消息中間件利於應用之間的解耦，生產者（客戶端）無需知道消費者（服務端）的存在。而且兩端可以使用不同的語言編寫，大大提供了靈活性。
安裝教程請參考:
ERLANG和rabbitmq安裝教程:
https://blog.csdn.net/weixin_39735923/article/details/79288578?tdsourcetag=s_pctim_aiomsg
1.RabbitMQ 之簡單模式
1.1最簡單的示例:
customer:
import pika # 1 連接rabbitmq connection = pika.BlockingConnection(pika.ConnectionParameters('localhost')) channel = connection.channel() # 2 創建隊列 channel.queue_declare(queue='hello') # 3.定義一個毀掉函數 def callback(ch,method,properties,body): print('客戶端收到了%s'%body ) # 確定監聽事件 channel.basic_consume(callback, queue='hello', no_ack=True, #此處代表不接受返回值 ) # 開始監聽 channel.start_consuming()
producer:
import pika # 1.鏈接rabbitmq connection = pika.BlockingConnection(pika.ConnectionParameters('localhost')) channel= connection.channel() # 2.創建隊列 channel.queue_declare(queue='hello') # 3.填值 channel.basic_publish( exchange='',# exchange=''說明為簡單模式 routing_key='hello', #關聯的隊列名 body='hello,mixtea', #往隊列內填入的內容 ) # 4 關閉連接 connection.close()
1.2 在調用 basic_consume 時一些參數的使用方法.
（1）no-ack ＝ False(主要預防消費端(處理端)出現問題導致數據丟失)，如果消費者遇到情況(its channel is closed, connection is closed, or TCP connection is lost)掛掉了，那么，RabbitMQ會重新將該任務添加到隊列中。
- 回調函數中的
ch.basic_ack(delivery_tag=method.delivery_tag) - basic_comsume中的
no_ack=False
customer:
import pika # 1 連接rabbitmq connection = pika.BlockingConnection(pika.ConnectionParameters('localhost')) channel = connection.channel() # 2 創建隊列 channel.queue_declare(queue='hello') # 3.定義一個毀掉函數(第二回重啟時記得刪掉mixtea,代表修復bug,再看看消息是否有丟失) def callback(ch,method,properties,body): mixtea print('客戶端收到了%s'%body ) # 應答 ch.basic_ack(delivery_tag=method.delivery_tag) # 確定監聽事件 channel.basic_consume(callback, queue='hello', no_ack=False, #no_ack=False此處代表接受返回值 ) # 開始監聽 channel.start_consuming()
producer:和上面簡單例子代碼一致
(2) durable ：消息不丟失,數據持久化,rabbitmq會將部分數據保存到磁盤上(主要防止rabbitmq服務器出現宕機)
測試方法:先關閉customer,開啟producer發送數據,此時數據發送到rabbitmq中,關閉rabbitmq,造成宕機現象,而后重啟,並且運行customer端接收數據.
可在未做持久化操作前后各嘗試一次,看差異
producer:
import pika # 1.鏈接rabbitmq connection = pika.BlockingConnection(pika.ConnectionParameters('localhost')) channel= connection.channel() # 2.創建隊列 # channel.queue_declare(queue='hello') channel.queue_declare(queue='hello2', durable=True) # 3.填值 channel.basic_publish( exchange='',# exchange=''說明為簡單模式 routing_key='hello2', #關聯的隊列名 body='hello,mixtea', #往隊列內填入的內容 properties=pika.BasicProperties( delivery_mode=2, # make message persistent 持久化數據 ) ) # 4 關閉連接 connection.close()
customer:
import pika # 1 連接rabbitmq connection = pika.BlockingConnection(pika.ConnectionParameters('localhost')) channel = connection.channel() # 2 創建隊列 # channel.queue_declare(queue='hello') channel.queue_declare(queue='hello2', durable=True) #現有的隊列不能修改成可持續化隊列,需重新創建隊列 # 3.定義一個毀掉函數(第二回重啟時記得刪掉mixtea,代表修復bug,再看看消息是否有丟失) def callback(ch,method,properties,body): print('客戶端收到了%s'%body ) # 應答 ch.basic_ack(delivery_tag=method.delivery_tag) # 確定監聽事件 channel.basic_consume(callback, queue='hello2', no_ack=False, #no_ack=False此處代表接受返回值 ) # 開始監聽 channel.start_consuming()
(3) 消息獲取順序
默認消息隊列里的數據是按照順序被消費者拿走，例如：消費者1 去隊列中獲取 奇數 序列的任務，消費者1去隊列中獲取 偶數 序列的任務。
channel.basic_qos(prefetch_count=1) 表示誰來誰取，不再按照奇偶數排列,平均分配
僅需要在客戶端加一句代碼即可:
import pika # 1 連接rabbitmq connection = pika.BlockingConnection(pika.ConnectionParameters('localhost')) channel = connection.channel() # 2 創建隊列 # channel.queue_declare(queue='hello') channel.queue_declare(queue='hello2', durable=True) #現有的隊列不能修改成可持續化隊列,需重新創建隊列 # 3.定義一個毀掉函數(第二回重啟時記得刪掉mixtea,代表修復bug,再看看消息是否有丟失) def callback(ch,method,properties,body): print('客戶端收到了%s'%body ) # 應答 ch.basic_ack(delivery_tag=method.delivery_tag) # 確定監聽事件 channel.basic_consume(callback, queue='hello2', no_ack=False, #no_ack=False此處代表接受返回值 ) # 分配方式 channel.basic_qos(prefetch_count=1) # 開始監聽 channel.start_consuming()
2.RabbitMQ 之exchange模式
2.1 發布訂閱 exchange type = fanout
發布訂閱和簡單的消息隊列區別在於，發布訂閱會將消息發送給所有的訂閱者，而消息隊列中的數據被消費一次便消失。所以，RabbitMQ實現發布和訂閱時，會為每一個訂閱者創建一個隊列，而發布者發布消息時，會將消息放置在所有相關隊列中。
大致流程如圖:
代碼如下:
生產者:
import pika # 1.鏈接rabbitmq connection = pika.BlockingConnection(pika.ConnectionParameters('localhost')) channel= connection.channel() # 2.聲明一個交換機 channel.exchange_declare(exchange='logs',exchange_type="fanout") #exchange_type="fanout" 代表為訂閱模式 #exchange='logs' logs為交換機名字,是變量可自己設定 # 3.填值 message ="Hello World&Mixtea!" channel.basic_publish( exchange='logs',# exchange=''說明為簡單模式 routing_key='', #關聯的隊列名 body = message, #往隊列內填入的內容 ) # 4 關閉連接 connection.close()
消費者:
import pika # 1 連接rabbitmq connection = pika.BlockingConnection(pika.ConnectionParameters('localhost')) channel = connection.channel() # 2.聲明交換機 channel.exchange_declare(exchange='logs',exchange_type='fanout') #3.創建隨機隊列名 result = channel.queue_declare(exclusive=True) queue_name = result.method.queue # 4.為隊列綁定交換機 channel.queue_bind(exchange='logs', queue=queue_name) # 4.回調函數 def callback(ch,method,properties,body): print('客戶端收到了%s'%body ) # 確定監聽事件 channel.basic_consume(callback, queue=queue_name, #no_ack=False此處代表不接受返回值 no_ack=True) # 開始監聽 channel.start_consuming()
2.2 關鍵字發送
發送消息時明確指定某個隊列並向其中發送消息，RabbitMQ還支持根據關鍵字發送，即：隊列綁定關鍵字，發送者將數據根據關鍵字發送到消息exchange，exchange根據 關鍵字 判定應該將數據發送至指定隊列。
其過程如下圖
詳細代碼如下:
生產者:
import pika # 1.鏈接rabbitmq connection = pika.BlockingConnection(pika.ConnectionParameters('localhost')) channel= connection.channel() # 2.聲明一個交換機 channel.exchange_declare(exchange='direct_logs',exchange_type="direct") #exchange_type="direct" 代表為關鍵字模式 #exchange='direct_logs' logs為交換機名字,是變量可自己設定 # 3.填值 message ="Hello World&Mixtea!" channel.basic_publish( exchange='direct_logs', routing_key='info', #說明此次發送的數值類型 body = message, #往隊列內填入的內容 ) # 4 關閉連接 connection.close()
消費者:
import pika # 1 連接rabbitmq connection = pika.BlockingConnection(pika.ConnectionParameters('localhost')) channel = connection.channel() # 2.聲明交換機 channel.exchange_declare(exchange='direct_logs',exchange_type='direct') #3.創建隨機隊列名 result = channel.queue_declare(exclusive=True) queue_name = result.method.queue # 4.為隊列綁定交換機 channel.queue_bind(exchange='direct_logs', queue=queue_name, routing_key='info', #在綁定交換機的同時綁定獲取值的類型 ) # 4.回調函數 def callback(ch,method,properties,body): print('客戶端收到了%s'%body ) # 確定監聽事件 channel.basic_consume(callback, queue=queue_name, #no_ack=False此處代表不接受返回值 no_ack=True, ) # 開始監聽 channel.start_consuming()
2.3 模糊匹配
exchange type = topic 發送者路由值 隊列中 old.boy.python old.* -- 不匹配 old.boy.python old.# -- 匹配
在topic類型下，可以讓隊列綁定幾個模糊的關鍵字，之后發送者將數據發送到exchange，exchange將傳入”路由值“和 ”關鍵字“進行匹配，匹配成功，則將數據發送到指定隊列。
- # 表示可以匹配 0 個 或 多個 單詞
- * 表示只能匹配 一個 單詞
使用方法和關鍵字查詢幾乎一樣,只需要在
routing_key="*.apple.#" 中的值改為模糊查詢的樣式即可(不常用)