python - 操作RabbitMQ

阿新 • • 發佈：2018-11-02

https://www.cnblogs.com/pangguoping/p/5720134.html

介紹

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 release 7.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
import pika
# ######################### 生產者 #########################
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

import pika

# ########################## 消費者 ##########################
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')


# 定義一個回撥函式來處理，這邊的回撥函式就是將資訊打印出來。
def callback(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
import pika
# ######################### 生產者 #########################
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()

消費者程式碼：

import pika
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')


def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)
    import time
    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 釋出者
import pika

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 訂閱者
import pika

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')


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.

消費者程式碼

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

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 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()

生產者

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# auth : pangguoping
# 釋出者
import pika

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

介紹

安裝

安裝RabbitMQ

Python 操作RabbitMQ

RabbitMQ持久化

1、acknowledgment 訊息不丟失的方法

2、訊息持久化儲存（Message durability）

釋出與訂閱

模式1 Fanout

模式2 Direct

模式3 Topic

python 操作RabbitMQ

Python操作rabbitmq系列（一）

Python操作rabbitmq系列（三）:多個接收端消費消息

python操作RabbitMQ

Python操作rabbitmq系列（六）:進行RPC調用

Python操作rabbitmq系列（五）:根據主題分配消息

python操作rabbitmq實現廣播效果

python - 操作RabbitMQ

Python操作RabbitMQ初體驗

python 操作mongoDB數據庫

Python操作遠程數據庫

#python# 操作文件和目錄

python 操作excel表格

python小工具：用python操作HP的Quality Center

使用python操作word

python操作excel

Python操作MySQL

python操作數據庫-數據表

Python 操作 Azure Blob Storage

Python操作SQLAchemy

python - 操作RabbitMQ

介紹

安裝

安裝RabbitMQ

Python 操作RabbitMQ

RabbitMQ持久化

1、acknowledgment 訊息不丟失的方法

2、訊息持久化儲存（Message durability）

釋出與訂閱

模式1 Fanout

模式2 Direct

模式3 Topic

相關推薦