RabbitMQ簡介

　　rabbitmq中文翻譯的話，主要還是mq字母上：Message Queue，即消息隊列的意思。rabbitmq服務類似于mysql、apache服務，只是提供的功能不一樣。rabbimq是用來提供發送消息的服務，可以用在不同的應用程序之間進行通信。

Python操作RabbitMQ

操作準備：安裝API

pip install pika

簡單的發送與接收案例

　　對于RabbitMQ來說，生產和消費不再針對內存里的一個Queue對象，而是某臺服務器上的RabbitMQ Server實現的消息隊列。

* > 生產者代碼

import pika# ######################### 生產者 ######################### # 連接到rabbitmq服務器 connection = pika.BlockingConnection(pika.ConnectionParameters(host='172.25.50.13')) channel = connection.channel() # 聲明消息隊列，消息將在這個隊列中進行傳遞。如果將消息發送到不存在的隊列，rabbitmq將會自動清除這些消息。如果隊列不存在，則創建 channel.queue_declare(queue='wzg') # 發送消息到上面聲明的'wzg'隊列，其中exchange表示交換器，能精確指定消息應該發送到哪個隊列(有指定的隊列可以不進行配置)，routing_key設置為隊列的名稱，body就是發送的內容 channel.basic_publish(exchange='',routing_key='wzg',body='Hello World!') print(" [x] Sent 'Hello World!'") # 關閉連接 connection.close()

* > 消費者代碼

import pika # ########################## 消費者 ########################## # 連接到rabbitmq服務器 connection = pika.BlockingConnection(pika.ConnectionParameters(host='172.25.50.13')) channel = connection.channel()# 聲明消息隊列，消息將在這個隊列中進行傳遞。如果隊列不存在，則創建 channel.queue_declare(queue='wzg')# 定義一個回調函數來處理，這邊的回調函數就是將信息打印出來。 def callback(ch, method, properties, body):print(" [x] Received %r" % body)# 告訴rabbitmq使用callback來接收信息 channel.basic_consume(callback,queue='wzg',no_ack=True) # no_ack=True表示在回調函數中不需要發送確認標識print(' [*] Waiting for messages. To exit press CTRL+C')# 開始接收信息，并進入阻塞狀態，隊列里有信息才會調用callback進行處理。按ctrl+c退出。 channel.start_consuming()

上述代碼中的知識點1：消息確認（Message acknowledgment）

　　生效方法：channel.basic_consume(consumer_callback,?queue,?no_ack=False,?exclusive=False,?consumer_tag=None,?arguments=None)　　

　　即no_ack=False（默認為False，即必須有確認標識）,在回調函數consumer_callback中，未收到確認標識，那么，RabbitMQ會重新將該任務添加到隊列中。

import pika # ########################## 消費者 ########################## # 連接到rabbitmq服務器 connection = pika.BlockingConnection(pika.ConnectionParameters(host='172.25.50.13')) channel = connection.channel() # 聲明消息隊列，消息將在這個隊列中進行傳遞 channel.queue_declare(queue='wzg')# 定義一個回調函數來處理，這邊的回調函數就是將信息打印出來。 def callback(ch, method, properties, body):print(" [x] Received %r" % body)exit() # 在這里exit，就會中斷發送確認標識，可以用來測試消息確認功能ch.basic_ack(delivery_tag=method.delivery_tag)# 告訴rabbitmq使用callback來接收信息 channel.basic_consume(callback,queue='wzg',no_ack=False) # no_ack=False表示在回調函數中必須發送確認標識，否則認為任務異常，會被重新添加到隊列中print(' [*] Waiting for messages. To exit press CTRL+C')# 開始接收信息，并進入阻塞狀態，隊列里有信息才會調用callback進行處理。按ctrl+c退出。 channel.start_consuming()

上述代碼中的知識點2：消息持久化存儲（Message durability）

　　雖然有了消息反饋機制，但是如果rabbitmq自身掛掉的話，那么任務還是會丟失。所以需要將任務持久化存儲起來。聲明持久化存儲

channel.queue_declare(queue='wzg', durable=True) # 聲明持久化

　　　Ps:?但是這樣程序會執行錯誤，因為‘wzg’這個隊列已經存在，并且是非持久化的，rabbitmq不允許使用不同的參數來重新定義存在的隊列。因此需要重新定義一個隊列

channel.queue_declare(queue='test_queue', durable=True) # 聲明持久化

注意：如果僅僅是設置了隊列的持久化，僅隊列本身可以在rabbit-server宕機后保留，隊列中的信息依然會丟失，如果想讓隊列中的信息或者任務保留，還需要做以下設置：

channel.basic_publish(exchange='',routing_key="test_queue",body=message,properties=pika.BasicProperties(delivery_mode = 2, # 使消息或任務也持久化存儲))

消息隊列持久化包括3個部分：
　　（1）exchange持久化，在聲明時指定durable => 1
　　（2）queue持久化，在聲明時指定durable => 1
　　（3）消息持久化，在投遞時指定delivery_mode=> 2（1是非持久化）

如果exchange和queue都是持久化的，那么它們之間的binding也是持久化的。如果exchange和queue兩者之間有一個持久化，一個非持久化，就不允許建立綁定。

上述代碼中的知識點3：公平調度（Fair dispatch）

　　上面實例中，雖然每個工作者是依次分配到任務，但是每個任務不一定一樣。可能有的任務比較重，執行時間比較久；有的任務比較輕，執行時間比較短。如果能公平調度就最好了，使用basic_qos設置prefetch_count=1，使得rabbitmq不會在同一時間給工作者分配多個任務，即只有工作者完成任務之后，才會再次接收到任務。

channel.basic_qos(prefetch_count=1) # 在消費者中進行設置

發布與訂閱

　　RabbitMQ的發布與訂閱，借助于交換機(Exchange)來實現。

　　交換機的工作原理：消息發送端先將消息發送給交換機，交換機再將消息發送到綁定的消息隊列，而后每個接收端(consumer)都能從各自的消息隊列里接收到信息。

　　　　　　　　　　　　

　　Exchange有三種工作模式，分別為：Fanout,?Direct,?Topic

模式1：Fanout

　　任何發送到Fanout Exchange的消息都會被轉發到與該Exchange綁定(Binding)的所有Queue上
　　1.可以理解為路由表的模式
　　2.這種模式不需要routing_key(即使指定，也是無效的)
　　3.這種模式需要提前將Exchange與Queue進行綁定，一個Exchange可以綁定多個Queue，一個Queue可以同多個Exchange進行綁定。
　　4.如果接受到消息的Exchange沒有與任何Queue綁定，則消息會被拋棄。

　　注意：這個時候必須先啟動消費者，即訂閱者。因為隨機隊列是在consumer啟動的時候隨機生成的，并且進行綁定的。producer僅僅是發送至exchange，并不直接與隨機隊列進行通信。

* > 生產者代碼

# rabbitmq 發布者 import pika # 創建連接 connection = pika.BlockingConnection(pika.ConnectionParameters(host='172.25.50.13')) channel = connection.channel() # 定義交換機，exchange表示交換機名稱，type表示類型 channel.exchange_declare(exchange='logs_fanout',type='fanout')message = 'Hello Python' # 將消息發送到交換機 channel.basic_publish(exchange='logs_fanout', # 指定exchangerouting_key='', # fanout下不需要配置，配置了也不會生效body=message) print(" [x] Sent %r" % message) connection.close()

* > 消費者代碼

# rabbitmq 訂閱者 import pikaconnection = pika.BlockingConnection(pika.ConnectionParameters(host='172.25.50.13')) channel = connection.channel()# 定義交換機，進行exchange聲明，exchange表示交換機名稱，type表示類型 channel.exchange_declare(exchange='logs_fanout',type='fanout')# 隨機創建隊列 result = channel.queue_declare(exclusive=True) # exclusive=True表示建立臨時隊列，當consumer關閉后，該隊列就會被刪除 queue_name = result.method.queue # 將隊列與exchange進行綁定 channel.queue_bind(exchange='logs_fanout',queue=queue_name)print(' [*] Waiting for logs. To exit press CTRL+C')def callback(ch, method, properties, body):print(" [x] %r" % body) # 從隊列獲取信息 channel.basic_consume(callback,queue=queue_name,no_ack=True)channel.start_consuming()

　　

?模式2：Direct

　　　　　　　　　　　

　　路由鍵的工作原理：每個接收端的消息隊列在綁定交換機的時候，可以設定相應的路由鍵。發送端通過交換機發送信息時，可以指明路由鍵 ，交換機會根據路由鍵把消息發送到相應的消息隊列，這樣接收端就能接收到消息了。　　

　　任何發送到Direct Exchange的消息都會被轉發到routing_key中指定的Queue：
　　1.一般情況可以使用rabbitMQ自帶的Exchange：”” ?(該Exchange的名字為空字符串)， 也可以自定義Exchange ??
　　2.這種模式下不需要將Exchange進行任何綁定(bind)操作。當然也可以進行綁定。可以將不同的routing_key與不同的queue進行綁定，不同的queue與不同exchange進行綁定
　　3.消息傳遞時需要一個“routing_key”
　　4.如果消息中中不存在routing_key中綁定的隊列名，則該消息會被拋棄。
　　如果一個exchange 聲明為direct，并且bind中指定了routing_key,那么發送消息時需要同時指明該exchange和routing_key.

?* > 消費者代碼

# 消費者 import pika # 建立連接 connection = pika.BlockingConnection(pika.ConnectionParameters(host='172.25.50.13')) channel = connection.channel() # 定義exchange和類型 channel.exchange_declare(exchange='direct_test',type='direct')# 生成隨機隊列 result = channel.queue_declare(exclusive=True) queue_name = result.method.queue severities = ['error',] # 將隨機隊列與routing_key關鍵字以及exchange進行綁定 for severity in severities:channel.queue_bind(exchange='direct_test',queue=queue_name,routing_key=severity) print(' [*] Waiting for logs. To exit press CTRL+C') def callback(ch, method, properties, body):print(" [x] %r:%r" % (method.routing_key, body)) # 接收消息 channel.basic_consume(callback,queue=queue_name,no_ack=True) channel.start_consuming()

* > 生產者代碼

# 發布者 import pika # 建立連接 connection = pika.BlockingConnection(pika.ConnectionParameters(host='172.25.50.13')) channel = connection.channel() # 定義交換機名稱及類型 channel.exchange_declare(exchange='direct_test',type='direct')severity = 'info' message = '123' # 發布消息至交換機direct_test，且發布的消息攜帶的關鍵字routing_key是info channel.basic_publish(exchange='direct_test',routing_key=severity,body=message) print(" [x] Sent %r:%r" % (severity, message)) connection.close()

　　當接收端正在運行時，可以使用rabbitmqctl list_bindings來查看綁定情況。

?模式3：Topic

　　　　　　　　　　　　　　?　　

?

?　　路由鍵模糊匹配，其實是路由鍵(routing_key)的擴展，就是可以使用正則表達式，和常用的正則表示式不同，這里的話“#”表示所有、全部的意思；“*”只匹配到一個詞。

　　任何發送到Topic Exchange的消息都會被轉發到所有關心routing_key中指定話題的Queue上
　　1.這種模式較為復雜，簡單來說，就是每個隊列都有其關心的主題，所有的消息都帶有一個“標題”(routing_key)，Exchange會將消息轉發到所有關注主題能與　　routing_key模糊匹配的隊列。
　　2.這種模式需要routing_key，也許要提前綁定Exchange與Queue。
　　3.在進行綁定時，要提供一個該隊列關心的主題，如“#.log.#”表示該隊列關心所有涉及log的消息(一個routing_key為”MQ.log.error”的消息會被轉發到該隊列)。
　　4.“#”表示0個或若干個關鍵字，“*”表示一個關鍵字。如“log.*”能與“log.warn”匹配，無法與“log.warn.timeout”匹配；但是“log.#”能與上述兩者匹配。
　　5.同樣，如果Exchange沒有發現能夠與routing_key匹配的Queue，則會拋棄此消息。

　　具體代碼這里不在多余寫，參照第二種模式的就可以，唯一變動的地方就是exchange type的聲明，以及進行綁定和發送的時候routing_key使用正則模式即可。

?

　　至此，利用Python操作RabbitMQ，以及一些簡單用法，在這里介紹完畢。如有錯誤，請指正！

?

轉載于:https://www.cnblogs.com/jishuweiwang/p/5713783.html

總結

以上是生活随笔為你收集整理的Python菜鸟之路：Python基础-Python操作RabbitMQ的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。