介紹

RabbitMQ是一個在AMQP基礎上完整的，可復用的企業消息系統。他遵循Mozilla Public License開源協議。

MQ全稱為Message Queue,

消息隊列(MQ)是一種應用程序對應用程序的通信方法。應用程序通過讀寫出入隊列的消息(針對應用程序的數據)來通信，而無需專用連接來鏈接它們。消

息傳遞指的是程序之間通過在消息中發送數據進行通信，而不是通過直接調用彼此來通信，直接調用通常是用于諸如遠程過程調用的技術。排隊指的是應用程序通過

隊列來通信。隊列的使用除去了接收和發送應用程序同時執行的要求。

應用場景：

RabbitMQ無疑是目前最流行的消息隊列之一，對各種語言環境的支持也很豐富，作為一個.NET developer有必要學習和了解這一工具。消息隊列的使用場景大概有3種：

1、系統集成，分布式系統的設計。各種子系統通過消息來對接，這種解決方案也逐步發展成一種架構風格，即“通過消息傳遞的架構”。

2、當系統中的同步處理方式嚴重影響了吞吐量，比如日志記錄。假如需要記錄系統中所有的用戶行為日志，如果通過同步的方式記錄日志勢必會影響系統的響應速度，當我們將日志消息發送到消息隊列，記錄日志的子系統就會通過異步的方式去消費日志消息。

3、系統的高可用性，比如電商的秒殺場景。當某一時刻應用服務器或數據庫服務器收到大量請求，將會出現系統宕機。如果能夠將請求轉發到消息隊列，再由服務器去消費這些消息將會使得請求變得平穩，提高系統的可用性。

安裝

安裝RabbitMQ

基礎環境：

內核3.10.0-327.el7.x86_64

系統版本

CentOS Linux release7.2.1511(Core)

安裝配置epel源#rpm -ivh http://mirrors.neusoft.edu.cn/epel/7/x86_64/e/epel-release-7-7.noarch.rpm

安裝erlang#yum install erlang

下載RabbitMQ 3.6.1

#wget http://www.rabbitmq.com/releases/rabbitmq-server/v3.6.1/rabbitmq-server-3.6.1-1.noarch.rpm

安裝rabbitmq-server#rpm -ivh rabbitmq-server-3.6.1-1.noarch.rpm

生成配置文件#cp /usr/share/doc/rabbitmq-server-3.6.1/rabbitmq.config.example /etc/rabbitmq/rabbitmq.config

啟動RabbitMQ#rabbitmq-server start

安裝Python API

#pip3 install pika

or

#easy_install pika

Python 操作RabbitMQ

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

1.生產者代碼

#!/usr/bin/env python#-*- coding: utf-8 -*-#auth : pangguoping

importpika########################## 生產者 #########################

credentials = pika.PlainCredentials('admin', 'admin')#鏈接rabbit服務器(localhost是本機，如果是其他服務器請修改為ip地址)

connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.1.103',5672,'/',credentials))#創建頻道

channel =connection.channel()#聲明消息隊列，消息將在這個隊列中進行傳遞。如果將消息發送到不存在的隊列，rabbitmq將會自動清除這些消息。如果隊列不存在，則創建

channel.queue_declare(queue='hello')#exchange -- 它使我們能夠確切地指定消息應該到哪個隊列去。#向隊列插入數值 routing_key是隊列名 body是要插入的內容

channel.basic_publish(exchange='',

routing_key='hello',

body='Hello World!')print("開始隊列")#緩沖區已經flush而且消息已經確認發送到了RabbitMQ中，關閉鏈接

connection.close()

2.消費者代碼

#!/usr/bin/env python#-*- coding: utf-8 -*-#auth : pangguoping

importpika########################### 消費者 ##########################

credentials = pika.PlainCredentials('admin', 'admin')#連接到rabbitmq服務器

connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.1.103',5672,'/',credentials))

channel=connection.channel()#聲明消息隊列，消息將在這個隊列中進行傳遞。如果隊列不存在，則創建

channel.queue_declare(queue='wzg')#定義一個回調函數來處理，這邊的回調函數就是將信息打印出來。

defcallback(ch, method, properties, body):print("[x] Received %r" %body)#告訴rabbitmq使用callback來接收信息

channel.basic_consume(callback,

queue='hello',

no_ack=True)#no_ack=True表示在回調函數中不需要發送確認標識

print('[*] Waiting for messages. To exit press CTRL+C')#開始接收信息，并進入阻塞狀態，隊列里有信息才會調用callback進行處理。按ctrl+c退出。

channel.start_consuming()

RabbitMQ持久化

1、acknowledgment 消息不丟失的方法

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

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

生產者代碼

#!/usr/bin/env python#-*- coding: utf-8 -*-#auth : pangguoping

importpika########################## 生產者 #########################

credentials = pika.PlainCredentials('admin', 'admin')#鏈接rabbit服務器(localhost是本機，如果是其他服務器請修改為ip地址)

connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.1.103',5672,'/',credentials))#創建頻道

channel =connection.channel()#聲明消息隊列，消息將在這個隊列中進行傳遞。如果將消息發送到不存在的隊列，rabbitmq將會自動清除這些消息。如果隊列不存在，則創建

channel.queue_declare(queue='hello')#exchange -- 它使我們能夠確切地指定消息應該到哪個隊列去。#向隊列插入數值 routing_key是隊列名 body是要插入的內容

channel.basic_publish(exchange='',

routing_key='hello',

body='Hello World!')print("開始隊列")#緩沖區已經flush而且消息已經確認發送到了RabbitMQ中，關閉鏈接

connection.close()

消費者代碼：

importpika

credentials= pika.PlainCredentials('admin', 'admin')#鏈接rabbit

connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.1.103',5672,'/',credentials))#創建頻道

channel =connection.channel()#如果生產者沒有運行創建隊列，那么消費者創建隊列

channel.queue_declare(queue='hello')defcallback(ch, method, properties, body):print("[x] Received %r" %body)importtime

time.sleep(10)print

'ok'ch.basic_ack(delivery_tag=method.delivery_tag) #主要使用此代碼

channel.basic_consume(callback,

queue='hello',

no_ack=False)print('[*] Waiting for messages. To exit press 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兩者之間有一個持久化，一個非持久化，就不允許建立綁定。

發布與訂閱

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，并不直接與隨機隊列進行通信。

生產者代碼

#!/usr/bin/env python#-*- coding: utf-8 -*-#auth : pangguoping#rabbitmq 發布者

importpika

credentials= pika.PlainCredentials('admin', 'admin')#鏈接rabbit服務器(localhost是本機，如果是其他服務器請修改為ip地址)

connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.1.103',5672,'/',credentials))

channel=connection.channel()#定義交換機，exchange表示交換機名稱，type表示類型

channel.exchange_declare(exchange='logs_fanout',

type='fanout')

message= 'Hello Python'

#將消息發送到交換機

channel.basic_publish(exchange='logs_fanout', #指定exchange

routing_key='', #fanout下不需要配置，配置了也不會生效

body=message)print("[x] Sent %r" %message)

connection.close()

消費者代碼

#!/usr/bin/env python#-*- coding: utf-8 -*-#auth : pangguoping

#rabbitmq 訂閱者

importpika

credentials= pika.PlainCredentials('admin', 'admin')#鏈接rabbit服務器(localhost是本機，如果是其他服務器請修改為ip地址)

connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.1.103',5672,'/',credentials))

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')defcallback(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.

消費者代碼

#!/usr/bin/env python#-*- coding: utf-8 -*-#auth : pangguoping#消費者

importpika

credentials= pika.PlainCredentials('admin', 'admin')#鏈接rabbit服務器(localhost是本機，如果是其他服務器請修改為ip地址)

connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.1.103',5672,'/',credentials))

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 inseverities:

channel.queue_bind(exchange='direct_test',

queue=queue_name,

routing_key=severity)print('[*] Waiting for logs. To exit press CTRL+C')defcallback(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()

生產者

#!/usr/bin/env python#-*- coding: utf-8 -*-#auth : pangguoping#發布者

importpika

credentials= pika.PlainCredentials('admin', 'admin')#鏈接rabbit服務器(localhost是本機，如果是其他服務器請修改為ip地址)

connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.1.103',5672,'/',credentials))

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，以及一些簡單用法，在這里介紹完畢。

參考http://www.cnblogs.com/jishuweiwang

總結

以上是生活随笔為你收集整理的python总线 rabbitmq_python - 操作RabbitMQ的全部內容，希望文章能夠幫你解決所遇到的問題。

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