伍哥原创之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教程