Python之操作RabbitMQ

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

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

RabbitMQ安裝

epel源安裝

現在服務器安裝epel源,為什么選擇epel源呢?強烈推薦大家使用epel源,epel是社區強烈打造的免費開源發行軟件包版本庫，系統包含大概有1萬多個軟件包，163和sohu的鏡像是沒有這么多軟件了.

首選確認你的版本號，然后才能選擇相應的epel，命令如下

[root@cobbler ~]# cat /etc/issue
CentOS release 6.5 (Final)

Centos6*源安裝:

rpm -ivh http://dl.fedoraproject.org/pub/epel/6/x86_64/epel-release-6-8.noarch.rpm

驗證是否安裝成功，執行如下命令:

[root@cobbler ~]# yum repolist  
Loaded plugins: fastestmirror, refresh-packagekit, security
Loading mirror speeds from cached hostfile
 * base: mirrors.yun-idc.com
 * epel: ftp.riken.jp
 * extras: mirrors.yun-idc.com
 * updates: mirrors.yun-idc.com
repo id                                        repo name                                                                              status
base                                           CentOS-6 - Base                                                                         6,696
epel                                           Extra Packages for Enterprise Linux 6 - x86_64                                         12,155
extras                                         CentOS-6 - Extras                                                                          62
updates                                        CentOS-6 - Updates                                                                        263
repolist: 19,176

看到epel，說明安裝成功了，可以看到epel有1萬2千155個包.有了他你不在需要tar、configure、make等等繁瑣的動作了。使用yum即可搞定一切.

還有一個好處,如果你用自動化運維,使用saltstack puppet ansilble 等等統一配置管理時,都一個重要的要求是統一標准化,我們用epel源,直接一條命令就能安裝了,不是很爽么?

安裝RabbitMQ

安裝erlang
   $ yum -y install erlang
 
安裝RabbitMQ
   $ yum -y install rabbitmq-server

Python中安裝API

pip install pika
or
easy_install pika

Python操作RabbitMQ

基本用法

發布者端:

import pika

connection=pika.BlockingConnection(pika.ConnectionParameters(host='192.168.4.193'))
channel=connection.channel()
channel.queue_declare(queue='cc')   #如果有cc的隊列,略過;如果沒有,創建cc的隊列

channel.basic_publish(exchange='',routing_key='cc',body='hello!world!!!')
print("[x] sent 'hello,world!'")
connection.close()

接收端:

import pika

#創建一個連接對象,對象中綁定了rabbitmq的IP
connection=pika.BlockingConnection(pika.ConnectionParameters(host='192.168.4.193'))
#創建一個頻道對象
channel=connection.channel()
#頻道中聲明指定queue,如果MQ中沒有指定queue就創建,如果有,則略過
channel.queue_declare(queue='cc')
#定義回調函數
def callback(ch,method,properties,body):
    print('[x] Recieved %r'%body)
    # channel.close()
#no_ack=Fales:表示消費完以后不主動把狀態通知rabbitmq,callback:回調函數,queue:指定隊列
channel.basic_consume(callback,queue='cc',no_ack=True)
# channel.basic_consume(callback,queue='cc')

print('[*] Waiting for msg')

channel.start_consuming()

acknowledgment 消息不丟失

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

消息接收端應該這么寫:

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.4.193'))
channel = connection.channel()
channel.queue_declare(queue='cc')

# 定義回調函數
def callback(ch, method, properties, body):
    print('[x] Recieved %r' % body)
    # channel.close()
    ch.basic_ack(delivery_tag=method.delivery_tag)


# no_ack=Fales:表示消費完以后不主動把狀態通知rabbitmq

channel.basic_consume(callback, queue='cc',
                      no_ack=False)  

print('[*] Waiting for msg')

channel.start_consuming()

durable 消息不丟失

消息生產者端發送消息時掛掉了,消費者接消息時掛掉了,以下方法會讓RabbitMQ重新將該消息添加到隊列中:

• 回調函數中的ch.basic_ack(delivery_tag=method.delivery_tag),消費端需要做的
• basic_comsume中的no_ack=False,消費端需要做的
• 發布消息端的basic_publish添加參數properties=pika.BasicProperties(delivery_mode=2),生產者端需要做的

消息生產者端:

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.4.193'))
channel = connection.channel()
channel.queue_declare(queue='cc')  # 如果有cc的隊列,略過;如果沒有,創建cc的隊列

channel.basic_publish(exchange='',
                      routing_key='cc',
                      body='hello!world!!!',
                      properties=pika.BasicProperties(delivery_mode=2)) #消息持久化
print("[x] sent 'hello,world!'")
connection.close()

消息消費者端:

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.4.193'))
channel = connection.channel()
channel.queue_declare(queue='cc')

# 定義回調函數
def callback(ch, method, properties, body):
    print('[x] Recieved %r' % body)
    # channel.close()
    ch.basic_ack(delivery_tag=method.delivery_tag)


# no_ack=Fales:表示消費完以后不主動把狀態通知rabbitmq

channel.basic_consume(callback, queue='cc',
                      no_ack=True)

print('[*] Waiting for msg')

channel.start_consuming()

消息獲取順序

默認消息隊列里的數據是按照順序被消費者拿走，例如：消費者1去隊列中獲取 奇數 序列的任務，消費者2去隊列中獲取 偶數 序列的任務。但有大部分情況下,消息隊列后端的消費者服務器的處理能力是不相同的,這就會出現有的服務器閑置時間較長,資源浪費的情況,那么,我們就需要改變默認的消息隊列獲取順序!

channel.basic_qos(prefetch_count=1) 表示誰來誰取，不再按照奇偶數排列,這是消費者端需要做的

消費者端如下:

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.4.193'))
channel = connection.channel()
channel.queue_declare(queue='cc')

# 定義回調函數
def callback(ch, method, properties, body):
    print('[x] Recieved %r' % body)
    # channel.close()
    ch.basic_ack(delivery_tag=method.delivery_tag)


channel.basic_qos(prefetch_count=1)     #改變默認獲取順序,誰來誰取

# no_ack=Fales:表示消費完以后不主動把狀態通知rabbitmq
channel.basic_consume(callback, queue='cc',
                      no_ack=True)

print('[*] Waiting for msg')

channel.start_consuming()

發布和訂閱

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

關鍵字:exchange type = fanout

消息生產者:

import pika

connection=pika.BlockingConnection(pika.ConnectionParameters(host='192.168.11.131'))
channel = connection.channel()
channel.exchange_declare(exchange='logs_fanout',type='fanout')

msg='456'
channel.basic_publish(exchange='logs_fanout',routing_key='',body=msg)

print('開始發送:%s'%msg)
connection.close()

消息消費者:

import pika

connection=pika.BlockingConnection(pika.ConnectionParameters(host='192.168.11.131'))
channel = connection.channel()
channel.exchange_declare(exchange='logs_fanout',type='fanout')

#隨機創建隊列
result=channel.queue_declare(exclusive=True)
queue_name=result.method.queue

#綁定相關隊列名稱
channel.queue_bind(exchange='logs_fanout',queue=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()

關鍵字發送

之前事例，發送消息時明確指定某個隊列並向其中發送消息，RabbitMQ還支持根據關鍵字發送，即：隊列綁定關鍵字，發送者將數據根據關鍵字發送到消息exchange，exchange根據 關鍵字 判定應該將數據發送至指定隊列。

關鍵字:exchange type = direct,默認模式也為此模式.

消息生產者端:

import pika

connection=pika.BlockingConnection(pika.ConnectionParameters(host='192.168.11.131'))
channel = connection.channel()
channel.exchange_declare(exchange='logs_direct_test1',type='direct')


serverity='error'
msg='123'

channel.basic_publish(exchange='logs_direct_test1',routing_key=serverity,body=msg)

print('開始發送:%r:%r'%(serverity,msg))
connection.close()

消息消費者1:

import pika

connection=pika.BlockingConnection(pika.ConnectionParameters(host='192.168.11.131'))
channel = connection.channel()
channel.exchange_declare(exchange='logs_direct_test1',type='direct')

#隨機創建隊列
result=channel.queue_declare(exclusive=True)
queue_name=result.method.queue

serverities=['error','info','warning',]
for serverity in serverities:
    channel.queue_bind(exchange='logs_direct_test1',queue=queue_name,routing_key=serverity)

print('[***] 開始接受消息!')

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

消息消費者2:

import pika

connection=pika.BlockingConnection(pika.ConnectionParameters(host='192.168.11.131'))
channel = connection.channel()
channel.exchange_declare(exchange='logs_direct_test1',type='direct')

#隨機創建隊列
result=channel.queue_declare(exclusive=True)
queue_name=result.method.queue

serverities=['error',]
for serverity in serverities:
    channel.queue_bind(exchange='logs_direct_test1',queue=queue_name,routing_key=serverity)

print('[***] 開始接受消息!')

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

模糊匹配

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

關鍵字:exchange type = topic

• • 表示只能匹配 一個 單詞
• # 表示可以匹配0個或多個單詞

發送者路由值              隊列中
old.boy.python          old.*  -- 不匹配
old.boy.python          old.#  -- 匹配

消息生產者:

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

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

channel.exchange_declare(exchange='topic_logs',
                         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()

消息消費者:

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

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

channel.exchange_declare(exchange='topic_logs',
                         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()