伍哥原創之python開發RabbitMQ應用

時間  2019-11-05
標籤 原創 python 開發 rabbitmq 應用 欄目 Python 简体版
原文   原文鏈接

【伍哥原創】html

使用python開發RabbitMQ應用
（參考了RabbitMQ網站上提供的英文版本入門指南: http://www.rabbitmq.com/getstarted.html）java

測試環境：CentOS 6.2python

1，測試環境準備git

安裝RabbitMQ server，python（通常系統都自帶了python）和pika 0.9.5。github

安裝RabbitMQ server能夠參考伍哥前面的文章。shell

安裝pika通常有兩種方式：
能夠經過pip或者easy_install來進行，不過pip是python的包管理器，須要單獨安裝，而CentOS已經準備好easy_install了（其實就是一個python腳本）。
官方文檔能夠參考這裏：http://pika.github.com/vim

1
2
3
4
5
6
7
8
9
10
11
$ easy_install pika
Reading https: //tonyg .github.com /pika/
Reading http: //pika .github.com/
Best match: pika 0.9.5
Downloading http: //pypi .python.org /packages/source/p/pika/pika-0 .9.5. tar .gz
Processing pika-0.9.5. tar .gz
Running pika-0.9.5 /setup .py -q bdist_egg --dist- dir /tmp/easy_install-1C9Vbo/pika-0 .9.5 /egg-dist-tmp-k1W5aK
Adding pika 0.9.5 to easy- install .pth file
Installed /usr/lib/python2 .6 /site-packages/pika-0 .9.5-py2.6.egg
Processing dependencies for pika
Finished processing dependencies for pika


你應該會看到: pika被安裝在/usr/lib/python2.6/site-packages/pika-0.9.5-py2.6.egg （蟒蛇蛋！嘿嘿）bash

而後把rabbitmq server啓動一下和準備好測試目錄rabbitmq_app：app

1
2
3
4
5
6
$ /usr/local/rabbitmq/sbin/rabbitmq-server -detached
 
$ cd ~
$ mkdir -p test /rabbitmq_app
$ cd test /rabbitmq_app
$ mkdir tut1 tut2 tut3 tut4 tut5 tut6

2，實例一：來個hello world程序
負載均衡

1
2
3
$ cd tut1
$ vim send.py (代碼以下)
$ vim receive.py (代碼以下)

首先是消息發送程序: send.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import sys
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters( 'localhost' ))
channel = connection.channel()
channel.queue_declare(queue = 'hello' )
if len (sys.argv) < 2 :
     print 'message is empty!'
     sys.exit( 0 )
message = sys.argv[ 1 ]
channel.basic_publish(exchange = ' ', routing_key=' hello', body = message)
print " [x] sent: '" + message + "' \n"
connection.close()

跑一下send.py發送一個消息

1
2
3
4
5
6
$ python send.py 'Hello World!'
$ python send.py '你好伍哥'
$ /usr/local/rabbitmq/sbin/rabbitmqctl list_queues
Listing queues ...
hello   2
... done .

若是你也看到hello隊列裏面有一個消息的話，就證實能夠發消息了。
而後寫一個接收消息腳本：receive.py

1
2
3
4
5
6
7
8
9
10
11
12
13
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters( 'localhost' ))
channel = connection.channel()
channel.queue_declare(queue = 'hello' )
print '[*] Waiting for messages. To exit press CTRL+C'
 
def callback(ch, method, properties, body):
     print body
 
channel.basic_consume(callback, queue = 'hello' , no_ack = True )
channel.start_consuming()

其中第12行的 no_ack=True 表示消費完了這個消息之後不主動把完成狀態通知rabbitmq。
而後開另一個shell，執行一下receive.py

1
2
3
4
$ python receive.py
[*] Waiting for messages. To exit press CTRL+C
Hello World!
你好伍哥

3，實例二：工做隊列（work queue / task queue）


通常應用於把比較耗時的任務從主線任務分離出來。好比一個http頁面請求，裏面須要發送帶大附件的郵件、或者是要處理一張頭像圖片等。這類型工做隊列的 處理端通常有多個worker進程，分擔隊列裏面的任務。這就有點負載均衡的策略在裏面了。儘可能作到每一個進程的工做量比較平均，並且是完成了一個任務才接 第二個任務。看看咱們的實現吧。

1
2
3
$ cd tut2
$ vim manager.py (代碼以下)
$ vim worker.py (代碼以下)

首先是消息發送程序: manager.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import pika
import sys
parameters = pika.ConnectionParameters(host = 'localhost' )
connection = pika.BlockingConnection(parameters)
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()

其中第8行的 durable=True 聲明瞭隊列須要持久化，第14行的 delivery_mode = 2 聲明瞭隊列的消息須要持久化。

而後寫一個接收消息腳本：worker.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
#!/usr/bin/env python
# -*- coding: utf-8 -*-
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( '.' ) )
     print " [x] Done"
     ch.basic_ack(delivery_tag = method.delivery_tag)
 
channel.basic_qos(prefetch_count = 1 )
channel.basic_consume(callback,
                       queue = 'task_queue' )
channel.start_consuming()

其中第15行的 basic_ack 是執行完任務通知rabbitmq，第17行的basic_qos是告訴rabbitmq只有當worker完成了任務之後才分派1條新的消息，實現公平分派。
測試方法，開3個bash，2個跑worker，1個跑manager：

1
2
3
4
$ python manager.py task1.
$ python manager.py task2..
$ python manager.py task3...
$ python manager.py task4....

點號數量決定worker工做的時間( 實際上是睡覺時間，呵呵 time.sleep(body.count('.')) )。
而在worker那邊，能夠看到每一個worker都處理了兩個任務。
這種分配機制就是所謂的循環調度（Round-robin dispatching）

4，實例三：發佈和訂閱

發佈訂閱模式，簡單來講就像是廣播，一個消息發佈出來之後，全部訂閱者都能聽到，至於接收到這個信息之後你們作什麼就看具體我的了。

啊！怎麼突然冒出個X，是什麼玩意！這個X就是所謂的exchange，簡單來講就是消息的管家，由他決定接收到的信息是放特定的隊列，仍是全部隊列，仍是直接丟棄。
其實在前兩個實例裏面，已經用到了exchange （channel.basic_publish(exchange='',...），這個exchange的名字爲空，外號無名（人若無名，即可專心練劍~）。他會把你的消息都轉達給routing_key指明的隊列。
當咱們聲明瞭exchange之後，咱們須要爲queue和exchange創建聯繫，這時候，就要用到綁定（binding）了。

1
2
3
$ cd tut3
$ vim emitlog.py (代碼以下)
$ vim recelog.py (代碼以下)

emitlog.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
#!/usr/bin/env python
import pika
import sys
 
connection = pika.BlockingConnection(pika.ConnectionParameters(
         host = 'localhost' ))
channel = connection.channel()
 
channel.exchange_declare(exchange = 'logs' ,
                          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()

recelog.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
#!/usr/bin/env python
import pika
 
connection = pika.BlockingConnection(pika.ConnectionParameters(
         host = 'localhost' ))
channel = connection.channel()
channel.exchange_declare(exchange = 'logs' ,
                          type = 'fanout' )
result = channel.queue_declare(exclusive = True )
queue_name = result.method.queue
channel.queue_bind(exchange = 'logs' ,
                    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()

測試：
和前一個實例差很少。開3個bash，2個跑recelog，1個跑emitlog。查看recelog是否都收到emitlog發送的消息。代碼裏面用 了一個fanout(意思是成扇形展開)類型的exchange，只要和exchange綁定的queue都能收到一份消息的 copy，routing_key會被忽略掉。

5，路由模式 （選擇接收信息）

1
2
3
$ cd tut4
$ vim emitlog.py (代碼以下)
$ vim recelog.py (代碼以下)

emitlog.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
#!/usr/bin/env python
import pika
import sys
 
connection = pika.BlockingConnection(pika.ConnectionParameters(
         host = 'localhost' ))
channel = connection.channel()
channel.exchange_declare(exchange = 'direct_logs' ,
                          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()

這裏聲明exchange時類型定義爲direct（直接匹配），就是說只有當一個信息的routing_key和隊列的binding_key一 致時，信息纔會被放入到這個隊列。消息發佈給exchange時必須帶上routing_key。其實在消息生產端，隊列這個概念是透明的。

recelog.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
#!/usr/bin/env python
import pika
import sys
 
connection = pika.BlockingConnection(pika.ConnectionParameters(
         host = 'localhost' ))
channel = connection.channel()
 
channel.exchange_declare(exchange = 'direct_logs' ,
                          type = 'direct' )
 
result = channel.queue_declare(exclusive = True )
queue_name = result.method.queue
 
severities = sys.argv[ 1 :]
if not severities:
     print >> sys.stderr, "Usage: %s [info] [warning] [error]" % \
                          (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()

這裏首先定義exchange，和消息發送端是同樣的。而後定義隊列，隊列是自動命名，而且只要進程終止，隊列就會終止。而後把隊列和 exchange綁定，綁定時的routing_key是用戶輸入的，若是輸入多個key，就作屢次的綁定。注意這裏的隊列仍是一個。若是你須要創建兩個 隊列，就得跑兩次這個python腳本。

6，topic和rpc

官方tutorial還有兩個高級一點的實例，topic和rpc，這裏就不做說明了，留着你們學學英文吧 :) RabbitMQ提供了不少消息隊列客戶端代碼，好比python，java，c等等，你們能夠根據產品或項目的實際狀況選擇。關鍵是原理必須搞懂。

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程