python操做RabbiMQ

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

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

puthon操做rabbitMQ的模塊名爲：pika

https://github.com/pika/pika

在操做rabbitMQ以前，先經過Queue理解生產者消費者模型
生產者消費者模型

1 #!/usr/bin/env python
 2 # _*_coding:utf-8 _*_
 3 
 4 import  Queue
 5 import  threading
 6 
 7 message = Queue.Queue(10)
 8 
 9 
10 def producter(i):
11 
12     while True:
13         message.put(i)
14 
15 def consumer(i):
16     while True:
17         message.get()
18 
19 
20 
21 for i in range(5):
22     w = threading.Thread(target=producter,args=(i,))
23     w.start()
24 
25 for i in range(2):
26     w = threading.Thread(target=consumer,args=(i,))
27     w.start()

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

生產者

1 #!/usr/bin/env python
 2 # _*_coding:utf-8 _*_
 3 
 4 import  pika
 5 
 6 connection = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.1.108')) #鏈接一臺rabbitMQ
 7 
 8 channel = connection.channel()  #建立一個頻道
 9 
10 channel.queue_declare("chenchao")  #聲明一個消息隊列
11 
12 
13 channel.basic_publish(exchange='',routing_key="chenchao",body="Fucking!")   #發送消息
14 
15 print "sent Fucking world!!!!"
16 
17 connection.close()  #關閉鏈接

消費者

1 #!/usr/bin/env python
 2 # _*_coding:utf-8 _*_
 3 
 4 # #############################消費者########################
 5 
 6 import  pika
 7 
 8 connection = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.1.108'))
 9 
10 channel = connection.channel()
11 
12 channel.queue_declare("chenchao")
13 
14 def callback(ch, method, properties, body):  #固定格式 必須有4個參數
15     print body
16 
17 
18 channel.basic_consume(callback,queue="chenchao",no_ack=True)  #將從隊列裏取出的數據回調給callback方法
19 
20 channel.start_consuming()   #開始取值

一、acknowledgment 消息不丟失

no-ack ＝ False，若是在傳遞消息的過程當中消費者遇到狀況(its channel is closed, connection is closed, or TCP connection is lost)掛掉了，那麼，RabbitMQ會從新將該任務消息添加到隊列中。

消費者

1 #!/usr/bin/env python
 2 # _*_coding:utf-8 _*_
 3 
 4 # #############################消費者########################
 5 
 6 import  pika
 7 
 8 connection = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.1.108'))
 9 
10 channel = connection.channel()
11 
12 channel.queue_declare("chenchao")
13 
14 def callback(ch, method, properties, body):  #固定格式 必須有4個參數
15     print body
16     import time
17     time.sleep(8)
18     print "sleep over"
19     ch.basic_ack(delivery_tag = method.delivery_tag )   #向生產者發送應答
20 
21 
22 channel.basic_consume(callback,queue="chenchao",no_ack=False)  #將從隊列裏取出的數據回調給callback方法
23 
24 channel.start_consuming()   #開始取值

ch.basic_ack(delivery_tag = method.delivery_tag ) #向生產者發送應答，表示已經接收到了數據
no_ack=False
二、durable 消息不丟失

若是以前的隊列爲非持久化的，那麼以後就不能將其修改成持久化的，必須從新建立一個新的隊列，並聲明爲持久化的隊列，而且在發送消息時也要註明是持久化消息才行。

生產者

1 #!/usr/bin/env python
 2 # _*_coding:utf-8 _*_
 3 
 4 # #######################生產者######################
 5 
 6 import  pika
 7 
 8 connection = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.1.108')) #鏈接一臺rabbitMQ
 9 
10 channel = connection.channel()  #建立一個頻道
11 
12 channel.queue_declare("chenchao2",durable=True)  #聲明一個消息隊列爲持久化的隊列
13 
14 
15 channel.basic_publish(exchange='',
16                       routing_key="chenchao2",
17                       body="Message NO.1",
18                       properties=pika.BasicProperties(delivery_mode=2,))  #發送消息爲持久化的數據
19 
20 print "sent Message OK!"
21 
22 connection.close()  #關閉鏈接

durable=True
properties=pika.BasicProperties(delivery_mode=2,)
消費者

1 #!/usr/bin/env python
 2 # _*_coding:utf-8 _*_
 3 
 4 # #############################消費者########################
 5 
 6 import  pika
 7 
 8 connection = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.1.108'))
 9 
10 channel = connection.channel()
11 
12 channel.queue_declare("chenchao2",durable=True)  #生命一個可持續化的隊列（若是隊列已經存在，這句無關緊要）
13 
14 def callback(ch, method, properties, body):  #固定格式 必須有4個參數
15     print body
16     import time
17     time.sleep(8)
18     print "sleep over"
19     ch.basic_ack(delivery_tag = method.delivery_tag )   #向生產者發送應答
20 
21 
22 channel.basic_consume(callback,queue="chenchao2",no_ack=False)  #將從隊列裏取出的數據回調給callback方法
23 
24 channel.start_consuming()   #開始取值

三、消息獲取順序

默認消息隊列裏的數據是按照順序被消費者拿走，例如：消費者1 去隊列中獲取 奇數 序列的任務，消費者1去隊列中獲取 偶數 序列的任務。

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

消費者

1 #!/usr/bin/env python
 2 # _*_coding:utf-8 _*_
 3 
 4 # #############################消費者########################
 5 
 6 import  pika
 7 
 8 connection = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.1.108'))
 9 
10 channel = connection.channel()
11 
12 channel.queue_declare("chenchao2",durable=True)  #生命一個可持續化的隊列（若是隊列已經存在，這句無關緊要）
13 
14 def callback(ch, method, properties, body):  #固定格式 必須有4個參數
15     print body
16     import time
17     time.sleep(8)
18     print "sleep over"
19     ch.basic_ack(delivery_tag = method.delivery_tag )   #向生產者發送應答
20 
21 channel.basic_qos(prefetch_count=1)   #獲取消息不在按奇偶規則獲取
22 
23 channel.basic_consume(callback,queue="chenchao2",no_ack=False)  #將從隊列裏取出的數據回調給callback方法
24 
25 channel.start_consuming()   #開始取值

四、發佈訂閱

發佈訂閱和簡單的消息隊列區別在於，發佈訂閱會將消息發送給全部的訂閱者，而消息隊列中的數據被消費一次便消失。因此，RabbitMQ實現發佈和訂閱時，會爲每個訂閱者建立一個隊列，而發佈者發佈消息時，會將消息放置在全部相關隊列中。

發佈者

1 #!/usr/bin/env python
 2 # _*_coding:utf-8 _*_
 3 
 4 import pika
 5 import sys
 6 
 7 connection = pika.BlockingConnection(pika.ConnectionParameters(
 8         host='192.168.1.108'))
 9 channel = connection.channel()
10 
11 
12 channel.exchange_declare(exchange='chenchao',
13                          type='fanout')     #聲明一個名稱爲chenchao的exchange  類型爲fanout
14 
15 
16 message = ' '.join(sys.argv[1:]) or "info: Hello every consumer2"
17 channel.basic_publish(exchange='chenchao',
18                       routing_key='',
19                       body=message)      #將消息發送到exchange裏，經過exchange發送到全部隊列
20 print(" [x] Sent %r" % message)
21 connection.close()

channel.exchange_declare(exchange='chenchao',

type='fanout')     #聲明一個名稱爲chenchao的exchange  類型爲fanout

訂閱者

1 #!/usr/bin/env python
 2 # _*_coding:utf-8 _*_
 3 
 4 import pika
 5 
 6 connection = pika.BlockingConnection(pika.ConnectionParameters(
 7         host='192.168.1.108'))
 8 channel = connection.channel()
 9 
10 
11 channel.exchange_declare(exchange='chenchao',
12                          type='fanout')       #聲明類型爲fanou名稱爲chenchao的exchange
13 
14 result = channel.queue_declare(exclusive=True)
15 queue_name = result.method.queue     #生成一個隨機名的隊列
16 
17 channel.queue_bind(exchange='chenchao',
18                    queue=queue_name)    #將隊列與exchange綁定
19 
20 print(' [*] Waiting for logs. To exit press CTRL+C')
21 
22 def callback(ch, method, properties, body):
23     print(" [x] %r" % body)
24 
25 channel.basic_consume(callback,
26                       queue=queue_name,
27                       no_ack=True)
28 
29 channel.start_consuming()

channel.queue_bind(exchange='chenchao',

queue=queue_name)    #將隊列與exchange綁定

五、關鍵字發送

exchange type = direct

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

生產者

1 #!/usr/bin/env python
 2 # _*_coding:utf-8 _*_
 3 
 4 import pika
 5 import sys
 6 
 7 connection = pika.BlockingConnection(pika.ConnectionParameters(
 8         host='192.168.1.108'))
 9 channel = connection.channel()
10 
11 
12 channel.exchange_declare(exchange='import',
13                          type='direct')     #聲明一個名稱爲import的exchange  類型爲direct
14 
15 
16 message = ' '.join(sys.argv[1:]) or "info: Hello are you Q1???"
17 channel.basic_publish(exchange='import',
18                       routing_key='Nice',
19                       body=message)      #將消息與關鍵字發送到exchange裏，經過關鍵字發送到綁定的隊列
20 print(" [x] Sent %r" % message)
21 connection.close()

type='direct'
routing_key='Nice'
訂閱者

1 #!/usr/bin/env python
 2 # _*_coding:utf-8 _*_
 3 
 4 import pika
 5 
 6 connection = pika.BlockingConnection(pika.ConnectionParameters(
 7         host='192.168.1.108'))
 8 channel = connection.channel()
 9 
10 
11 channel.exchange_declare(exchange='import',
12                          type='direct')       #聲明類型爲direct名稱爲import的exchange
13 
14 result = channel.queue_declare(exclusive=True)
15 queue_name = result.method.queue     #生成一個隨機名的隊列
16 
17 channel.queue_bind(exchange='import',
18                    queue=queue_name,
19                    routing_key="Queue1")    #將隊列與exchange綁定,並聲明關鍵字
20 
21 
22 channel.queue_bind(exchange='import',
23                    queue=queue_name,
24                    routing_key="Nice")    #將隊列與exchange綁定，並聲明關鍵字
25 
26 print(' [*] Waiting for logs. To exit press CTRL+C')
27 
28 def callback(ch, method, properties, body):
29     print(" [x] %r:%r" % (method.routing_key, body))
30 
31 channel.basic_consume(callback,
32                       queue=queue_name,
33                       no_ack=True)
34 
35 channel.start_consuming()

訂閱者2

1 #!/usr/bin/env python
 2 # _*_coding:utf-8 _*_
 3 
 4 import pika
 5 
 6 connection = pika.BlockingConnection(pika.ConnectionParameters(
 7         host='192.168.1.108'))
 8 channel = connection.channel()
 9 
10 
11 channel.exchange_declare(exchange='import',
12                          type='direct')       #聲明類型爲direct名稱爲import的exchange
13 
14 result = channel.queue_declare(exclusive=True)
15 queue_name = result.method.queue     #生成一個隨機名的隊列
16 
17 channel.queue_bind(exchange='import',
18                    queue=queue_name,
19                    routing_key="Queue2")    #將隊列與exchange綁定，並聲明關鍵字
20 
21 
22 channel.queue_bind(exchange='import',
23                    queue=queue_name,
24                    routing_key="Nice")    #將隊列與exchange綁定，並聲明關鍵字
25 
26 print(' [*] Waiting for logs. To exit press CTRL+C')
27 
28 def callback(ch, method, properties, body):
29     print(" [x] %r:%r" % (method.routing_key, body))
30 
31 channel.basic_consume(callback,
32                       queue=queue_name,
33                       no_ack=True)
34 
35 channel.start_consuming()

　　type='direct'

　　routing_key="Nice"

六、模糊匹配

exchange type = topic

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

表示能夠匹配 0 個 或 多個 單詞

• 表示只能匹配 一個 單詞

routing_key="Nice.*"
routing_key="Nice.#"

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

發佈者

1 #!/usr/bin/env python
 2 # _*_coding:utf-8 _*_
 3 
 4 import pika
 5 import sys
 6 
 7 connection = pika.BlockingConnection(pika.ConnectionParameters(
 8         host='192.168.1.108'))
 9 channel = connection.channel()
10 
11 
12 channel.exchange_declare(exchange='topic_logs',
13                          type='topic')     #聲明一個名稱爲import的exchange  類型爲direct
14 
15 
16 message = ' '.join(sys.argv[1:]) or "info: Are you choosed?"
17 channel.basic_publish(exchange='topic_logs',
18                       routing_key='Nice',
19                       body=message)      #將消息與關鍵字發送到exchange裏，經過關鍵字發送到綁定的隊列
20 print(" [x] Sent %r" % message)
21 connection.close()

type='topic'
訂閱者1

1 #!/usr/bin/env python
 2 # _*_coding:utf-8 _*_
 3 
 4 import pika
 5 
 6 connection = pika.BlockingConnection(pika.ConnectionParameters(
 7         host='192.168.1.108'))
 8 channel = connection.channel()
 9 
10 
11 channel.exchange_declare(exchange='topic_logs',
12                          type='topic')       #聲明類型爲topic的exchange
13 
14 result = channel.queue_declare(exclusive=True)
15 queue_name = result.method.queue     #生成一個隨機名的隊列
16 
17 
18 channel.queue_bind(exchange='topic_logs',
19                    queue=queue_name,
20                    routing_key="Nice.*")    #將隊列與exchange綁定，並聲明關鍵字 *表明只能匹配一個
21 
22 print(' [*] Waiting for logs. To exit press CTRL+C')
23 
24 def callback(ch, method, properties, body):
25     print(" [x] %r:%r" % (method.routing_key, body))
26 
27 channel.basic_consume(callback,
28                       queue=queue_name,
29                       no_ack=True)
30 
31 channel.start_consuming()

type='topic'
routing_key="Nice.*"
訂閱者2

1 #!/usr/bin/env python
 2 # _*_coding:utf-8 _*_
 3 
 4 import pika
 5 
 6 connection = pika.BlockingConnection(pika.ConnectionParameters(
 7         host='192.168.1.108'))
 8 channel = connection.channel()
 9 
10 
11 channel.exchange_declare(exchange='topic_logs',
12                          type='topic')       #聲明類型爲direct名稱爲import的exchange
13 
14 result = channel.queue_declare(exclusive=True)
15 queue_name = result.method.queue     #生成一個隨機名的隊列
16 
17 channel.queue_bind(exchange='topic_logs',
18                    queue=queue_name,
19                    routing_key="Nice.#")    #將隊列與exchange綁定，並聲明關鍵字 #表明只能匹配0個或者多個
20 
21 print(' [*] Waiting for logs. To exit press CTRL+C')
22 
23 def callback(ch, method, properties, body):
24     print(" [x] %r:%r" % (method.routing_key, body))
25 
26 channel.basic_consume(callback,
27                       queue=queue_name,
28                       no_ack=True)
29 
30 channel.start_consuming()

routing_key="Nice.#"