2. 程式人生 > >python操作RabbitMQ

python操作RabbitMQ

阿新 • • 發佈：2017-10-12
相關 cred 開源 callback 進行 偶數 生產者 開源協議 erro

介紹

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

MQ全稱為Message Queue, 消息隊列（MQ）是一種應用程序對應用程序的通信方法。應用程序通過讀寫出入隊列的消息（針對應用程序的數據）來通信，而無需專用連接來鏈接它們。消 息傳遞指的是程序之間通過在消息中發送數據進行通信，而不是通過直接調用彼此來通信，直接調用通常是用於諸如遠程過程調用的技術。排隊指的是應用程序通過 隊列來通信。隊列的使用除去了接收和發送應用程序同時執行的要求。

安裝

RabbitMQ 下載地址：http://www.rabbitmq.com/install-standalone-mac.html

安裝配置epel源
   $ rpm -ivh http://dl.fedoraproject.org/pub/epel/6/i386/epel-release-6-8.noarch.rpm
 
安裝erlang
   $ yum -y install erlang
 
安裝RabbitMQ
   $ yum -y install rabbitmq-server

使用

官方使用文檔：http://www.rabbitmq.com/getstarted.html

1.實現最簡單的隊列通信

技術分享

#!/usr/bin/env python
#一個簡單的發數據send端，一對一模式(生產者)
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘localhost‘))	 #初始化連接，連接到RabbitMQ
channel = connection.channel()	 #生成一個連接管道

channel.queue_declare(queue=‘hello‘)	#聲明一個隊列名

 
#生產消息
channel.basic_publish(exchange=‘‘,		
                      routing_key=‘hello‘,	#要發送到那個隊列中
                      body=‘Hello World!‘	#發送的內容
					  )
print(" [x] Sent ‘Hello World!‘")
connection.close()	#關閉RabbitMQ的連接

#!/usr/bin/env python
#一個簡單的recvive端,一對一模式(消費者)
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘localhost‘))	#初始化連接，連接到RabbitMQ
channel = connection.channel()	#生成一個連接管道

channel.queue_declare(queue=‘hello‘)	#這裏也要聲明一下隊列的名稱，如果客戶端啟動，服務端沒有啟動，沒有聲明會報錯

def callback(ch, method, properties, body):	#定義一個接收到消息之後的回調函數
    print(" [x] Received %r" % body)

#消費消息
channel.basic_consume(callback,		#當接收到一個消息之後回調CALLBACk函數處理
                      queue=‘hello‘,	#定義那個隊列中去取，消息
                      no_ack=True	#這個意思是不返回確認，消息處理完成之後返回確認，RabbitMQ從隊列中刪除，一般不用
					  )

print(‘ [*] Waiting for messages. To exit press CTRL+C‘)

channel.start_consuming()	#啟動連接，如果沒有數據接收就會一直等待，不會停止

2. Work Queues

技術分享

在這種模式下，RabbitMQ會默認把p發的消息依次分發給各個消費者(c),跟負載均衡差不多

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會重新將該任務添加到隊列中。(消費者端)

2) 消息持久化

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

channel.queue_declare(queue=‘hello‘, durable=True) # 聲明隊列持久化

Ps: 但是這樣程序會執行錯誤，因為‘hello’這個隊列已經存在，並且是非持久化的，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) 消息公平分發

技術分享

如果Rabbit只管按順序把消息發到各個消費者身上，不考慮消費者負載的話，很可能出現，一個機器配置不高的消費者那裏堆積了很多消息處理不完，同時配置高的消費者卻一直很輕松。為解決此問題，可以在各個消費者端，配置perfetch=1,意思就是告訴RabbitMQ在我這個消費者當前消息還沒處理完的時候就不要再給

channel.basic_qos(prefetch_count=1) 表示誰來誰取，不再按照奇偶數排列

帶消息持久化+公平分發的完整代碼

生產者端：

#!/usr/bin/env python
import pika
import sys
 
connection = pika.BlockingConnection(pika.ConnectionParameters(
        host=‘localhost‘))
channel = connection.channel()
 
channel.queue_declare(queue=‘task_queue‘, durable=True)
 
message = ‘ ‘.join(sys.argv[1:]) or "Hello World!"
channel.basic_publish(exchange=‘‘,
                      routing_key=‘task_queue‘,
                      body=message,
                      properties=pika.BasicProperties(
                         delivery_mode = 2, # make message persistent
                      ))
print(" [x] Sent %r" % message)
connection.close()

消費者端：

#!/usr/bin/env python
import pika
import time
 
connection = pika.BlockingConnection(pika.ConnectionParameters(
        host=‘localhost‘))
channel = connection.channel()
 
channel.queue_declare(queue=‘task_queue‘, durable=True)
print(‘ [*] Waiting for messages. To exit press CTRL+C‘)
 
def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)
    time.sleep(body.count(b‘.‘))	#模擬處理操作
    print(" [x] Done")
    ch.basic_ack(delivery_tag = method.delivery_tag)
 
channel.basic_qos(prefetch_count=1)	#消息公平分發，配置perfetch=1,意思就是告訴RabbitMQ在我這個消費者當前消息還沒處理完的時候就不要再給我發新消息了
channel.basic_consume(callback,
                      queue=‘task_queue‘)
 
channel.start_consuming()

3. 發布訂閱

發布訂閱和簡單的消息隊列區別在於，發布訂閱會將消息發送給所有的訂閱者，而消息隊列中的數據被消費一次便消失。所以，RabbitMQ實現發布和訂閱時，會為每一個訂閱者創建一個隊列，而發布者發布消息時，會將消息放置在所有相關隊列中。

exchange_type = fanout

技術分享

發布者：

#!/usr/bin/env python
import pika
import sys

connection = pika.BlockingConnection(pika.ConnectionParameters(
        host=‘localhost‘))
channel = connection.channel()

channel.exchange_declare(exchange=‘logs‘,
                         exchange_type=‘fanout‘)

message = ‘ ‘.join(sys.argv[1:]) or "info: Hello World!"
channel.basic_publish(exchange=‘logs‘,
                      routing_key=‘‘,
                      body=message)
print(" [x] Sent %r" % message)
connection.close()

訂閱者：

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘localhost‘))

channel = connection.channel()

channel.exchange_declare(exchange=‘logs‘,
                         exchange_type=‘fanout‘)

result = channel.queue_declare(exclusive = True)    #不指定queue名字,rabbit會隨機分配一個名字,exclusive=True會在使用此queue的消費者斷開後,自動將queue刪除

queue_name = result.method.queue

channel.queue_bind(exchange = ‘logs‘,       #綁定
                   queue = queue_name)

print(‘ [*] Waiting for logs. To exit press CTRL+C‘)

print(‘randome queuename:‘,queue_name)

def callback(ch, method, properties, body):
    print(" [x] %r" % body)

channel.basic_consume(callback,
                      queue = queue_name,
                      no_ack = True
                      )

channel.start_consuming()

4. 有選擇的接受消息（關鍵字發送）

RabbitMQ還支持根據關鍵字發送，即：隊列綁定關鍵字，發送者將數據根據關鍵字發送到消息exchange，exchange根據 關鍵字 判定應該將數據發送至指定隊列。

　　 exchange_type = direct

技術分享

publisher：

import pika
import sys

connection = pika.BlockingConnection(pika.ConnectionParameters(
                                                    host = ‘localhost‘))

channel = connection.channel()

channel.exchange_declare(exchange = ‘direct_logs‘,
                         exchange_type = ‘direct‘)

severity = sys.argv[1] if len(sys.argv) > 1 else ‘info‘

message = ‘ ‘.join(sys.argv[2:]) or ‘Hello World!‘

channel.basic_publish(exchange = ‘direct_logs‘,
                      routing_key = severity,
                      body = message)

print(" [x] Sent %r:%r" % (severity, message))

connection.close()

subscriber：

import pika
import sys

connection = pika.BlockingConnection(pika.ConnectionParameters(
                                                        host = ‘localhost‘))

channel = connection.channel()

channel.exchange_declare(exchange = ‘direct_logs‘,
                         exchange_type = ‘direct‘)

result = channel.queue_declare(exclusive = True)

queue_name = result.method.queue

severities = sys.argv[1:]

if not severities:
    sys.stderr.write("Usage: %s [info] [warning] [error]\n" % sys.argv[0])
    sys.exit(1)

for severity in severities:
    channel.queue_bind(exchange = ‘direct_logs‘,
                       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()

5. 更細致的消息過濾（模糊匹配）

技術分享

exchange_type = topic

在topic類型下，可以讓隊列綁定幾個模糊的關鍵字，之後發送者將數據發送到exchange，exchange將傳入”路由值“和 ”關鍵字“進行匹配，匹配成功，則將數據發送到指定隊列。

  • # 表示可以匹配 0 個 或 多個 單詞
  • * 表示只能匹配 一個 單詞
發送者路由值              隊列中
error.info.warning    error.*    //不匹配
error.info.warning    error.#    //匹配

消費者：

import pika
import sys

connection = pika.BlockingConnection(pika.ConnectionParameters(
                                                        host = ‘localhost‘))
channel = connection.channel()

channel.exchange_declare(exchange = ‘topic_logs‘,
                         exchange_type = ‘topic‘)

result = channel.queue_declare(exclusive = True)

queue_name = result.method.queue

binding_keys = sys.argv[1:]

if not binding_keys:
    sys.stderr.write("Usage: %s [binding_key]...\n" % sys.argv[0])
    sys.exit(1)

for binding_key in binding_keys:
    channel.queue_bind(exchange = ‘topic_logs‘,
                       queue = queue_name,
                       routing_key = binding_key)

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
import sys

connection = pika.BlockingConnection(pika.ConnectionParameters(
                                                    host = ‘localhost‘))
channel = connection.channel()

channel.exchange_declare(exchange = ‘topic_logs‘,
                         exchange_type = ‘topic‘)

routing_key = sys.argv[1] if len(sys.argv) > 1 else ‘anonymous.info‘

message = ‘ ‘.join(sys.argv[2:]) or ‘Hello World!‘

channel.basic_publish(exchange = ‘topic_logs‘,
                      routing_key = routing_key,
                      body = message)

print(" [x] Sent %r:%r" % (routing_key, message))

connection.close()

其他

技術分享 
#添加賬戶
sudo rabbitmqctl add_user musker 123.com
# 設置用戶為administrator角色
sudo rabbitmqctl set_user_tags musker administrator
# 設置權限
sudo rabbitmqctl set_permissions -p "/" musker‘.‘‘.‘‘.‘

# 然後重啟rabbiMQ服務
sudo /etc/init.d/rabbitmq-server restart
 
# 可以使用的用戶遠程連接rabbitmq server了。

credentials = pika.PlainCredentials("musker","123.com")

connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.14.47‘,credentials=credentials))

#查看隊列和消息數量
rabbitmqctl list_queues
View Code

python操作RabbitMQ

相關推薦

python 操作RabbitMQ

ack 模糊匹配 host star cal top con 消失 nbsp pip install pika 使用API操作RabbitMQ 基於Queue實現生產者消費者模型 View Code 對於RabbitMQ來說，生產和消費不再針對內存裏的一個Queue

Python操作rabbitmq系列（一）

targe 紅色 入門 web 之間 cap ssa 隊列 技術 從本文開始，接下來的內容，我們將討論rabbitmq的相關功能。我的這些文章，最終是要實現一個項目（具體是什麽暫不透露）。前面每一篇，都是在為這個系統做準備。rabbitmq，是我們這個項目的關鍵部分之一。所

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

name 連接 logs http clas header 消費者 exclusive pub 接著上一章。這一章，我們要將同一個消息發給多個客戶端。這就是發布訂閱模式。直接看代碼： 發送端： import pikaimport sysconnection = pika.B

python操作RabbitMQ

相關 cred 開源 callback 進行 偶數 生產者 開源協議 erro 介紹 RabbitMQ是一個在AMQP基礎上完整的，可復用的企業消息系統。他遵循Mozilla Public License開源協議。 MQ全稱為Message Queue, 消息隊列（MQ

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

block 異常 遠程 轉換 調用 成了 mage chang 多少 此刻，我們已經進入第6章，是官方的最後一個環節，但是，並非本系列的最後一個環節。因為在實戰中還有一些經驗教訓，並沒體現出來。由於馬上要給同事沒培訓celery了。我也來不及寫太多。等後面，我們再慢慢補充。

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

method type 同時 elephant com .info err 現在 bin 接著上一章，使用exchange_type=‘direct‘進行消息傳遞。這樣消息會完全匹配後發送到對應的接收端。現在我們想幹這樣一件事： C1獲取消息中包含：orange內容的消息，

python操作rabbitmq實現廣播效果

連接 lba cti all rec alt aid tin back 生產方（Fanout_Publisher.py） 1 # __author__ = ‘STEVEN‘ 2 import pika 3 #開啟socket 4 connection = pik

python - 操作RabbitMQ

https://www.cnblogs.com/pangguoping/p/5720134.html   介紹 RabbitMQ是一個在AMQP基礎上完整的，可複用的企業訊息系統。他遵循Mozilla Public License開源協議。MQ全稱為Message Queue, 訊息佇列（MQ

Python操作RabbitMQ初體驗

/*學院顯示頁 左上*/ var cpro_id = "u2659916"; 由於想用Python實現一套分散式系統，來管理和監控CDN的內容與執行狀態，誤打誤撞認識了RabbitMQ，推薦的人很多，如餘鋒《我為什麼要選擇RabbitMQ》等等。 在MQ這個詞彙映入眼簾的時候，

python 操作mongoDB數據庫

查看 $set 一個 關於 pan date set 完整 mongodb 網上關於python 操作mongoDB的相關文章相對不是很多，並且質量也不是很高！下面給出一個完整的 增刪改查示例程序！ #!/usr/bin/python # -*- coding

Python操作遠程數據庫

his exec 通配符 .get toc 大量 while 避免 登入 我的項目要往數據庫中插入create_time和update_time，那就勢必要引用現在的系統時間，經過大量的查找，終於發現往python是沒有對應時間datetime的相關通配符的，那麽我們要怎麽

#python# 操作文件和目錄

級別 編寫一個程序 路徑 pre 找文件 練習 itext os.path 當前 總結於廖雪峰老師的python教程。 操作文件和目錄的函數一部分放在os模塊中，一部分放在os.path模塊中。 總結如下： 1 os.path.abspath(‘.‘)

python 操作excel表格

雷達圖 tle sum utf-8 範圍 red lod 圖標 data Python 操作excel 表格 #coding=utf-8 import xlsxwriter #1.創建excel 對象 work = xlsxwr

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

over cti 步驟 response headers 服務器 登錄 chm format 背景是這樣的：這個組的測試人員每跑一個case都要上傳測試結果附件到QC。每個待測功能模塊可能包含幾十上百的case。於是手工上傳測試結果變成了繁重的體力勞動。令人驚訝的是我們的工

使用python操作word

data close def pip 行數 pri path http 行數據 有兩種方式: 使用win32com 使用docx 1.使用win32com擴展包 只對windows平臺有效 代碼: # coding=utf-8 import win32com fr

python操作excel

int lena cells src tar 程序 win32 div cep python操作exce的方式: 使用win32com 使用xlrd(讀excel)、xlwt(寫excel) 1.使用win32com方式 代碼: # coding=utf-8 from

Python操作MySQL

port sdf cnblogs ctc hal tro 多行 執行sql lai 本篇對於Python操作MySQL主要使用兩種方式： 原生模塊 pymsql ORM框架 SQLAchemy pymsql pymsql是Python中操作MySQL的模塊，其使用方法和

python操作數據庫-數據表

title 獲取 arch python操作 -- pymysql exe span 轉義 數據表： 數據類型： 幫助的三種形式： 在cmd中輸入： help 要幫助的主題詞，或 ? 要幫助的主題詞 或 \h 要幫助的主題詞 。 數據表的創建：

Python 操作 Azure Blob Storage

edi 安裝 一點 是你 接下來 ges list settings containe 筆者在《Azure 基礎：Blob Storage》一文中介紹了 Azure Blob Storage 的基本概念，並通過 C# 代碼展示了如何進行基本的操作。最近筆者需要在 Linux

Python操作SQLAchemy

kref func ont == 更多 字典 drop ring constrain 實在不懂就看這個：http://www.cnblogs.com/jixuege-1/p/6272888.html 本篇對於Python操作MySQL主要使用兩種方式： 原生模塊 pyms