文档章节

使用python开发RabbitMQ应用

沉淀岁月
 沉淀岁月
发布于 2016/09/07 08:25
字数 1780
阅读 48
收藏 0

使用python开发RabbitMQ应用

(参考了RabbitMQ网站上提供的英文版本入门指南: http://www.rabbitmq.com/getstarted.html

 

测试环境:CentOS 6.2

1,测试环境准备

安装python(一般系统都自带了python)

安装RabbitMQ server可以参考前面的文章。

安装pika

使用pip安装的时候可能会报错:

importerror no module named pkg_resources

请用下面命令解决这个问题:

$ curl https://bitbucket.org/pypa/setuptools/raw/bootstrap/ez_setup.py | python

然后还可能出现:

pkg_resources.distributionnotfound pip==1.4.1

这时候先把pip卸载掉, sudo yum remove python-pip

然后去下载最新的get-pip.py文件,python get-pip.py安装

在/etc/profile里面将/usr/local/python27/bin加入PATH最前面

 

把rabbitmq server启动一下和准备好测试目录rabbitmq_app:

$ /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程序

$ cd tut1

$ vim send.py (代码如下)

$ vim receive.py (代码如下)

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

  1. #!/usr/bin/env python  

  2. # -*- coding: utf-8 -*-  

  3. import sys  

  4. import pika  

  5. connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))  

  6. channel = connection.channel()  

  7. channel.queue_declare(queue = 'hello')  

  8. if len (sys.argv) < 2 :  

  9.      print 'message is empty!'  

  10.      sys.exit(0)  

  11. message = sys.argv[1]  

  12. channel.basic_publish(exchange = '', routing_key='hello', body = message)  

  13. print "[x] sent: '" + message + "'\n"  

  14. connection.close()  

 

跑一下send.py发送一个消息

$ python send.py 'Hello World!'

$ python send.py '你好刀哥'

$ /usr/local/rabbitmq/sbin/rabbitmqctl list_queues

Listing queues ...

hello   2

... done .

如果你也看到hello队列里面有一个消息的话,就证明可以发消息了。

然后写一个接收消息脚本:receive.py

  1. #!/usr/bin/env python  

  2. # -*- coding: utf-8 -*-  

  3. import pika  

  4. connection = pika.BlockingConnection(pika.ConnectionParameters( 'localhost' ))  

  5. channel = connection.channel()  

  6. channel.queue_declare(queue = 'hello' )  

  7. print '[*] Waiting for messages. To exit press CTRL+C'  

  8.    

  9. def callback(ch, method, properties, body):  

  10.      print body  

  11.    

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

  13. channel.start_consuming()  

 

其中第12行的 no_ack=True 表示消费完了这个消息以后不主动把完成状态通知rabbitmq。

然后开另外一个shell,执行一下receive.py

$ python receive.py

[*] Waiting for messages. To exit press CTRL+C

Hello World!

你好刀哥

 

3,实例二:工作队列(work queue / task queue)

一般应用于把比较耗时的任务从主线任务分离出来。比如一个http页面请求,里面需要发送带大附件的邮件、或者是要处理一张头像图片等。这类型工作队列的 处理端一般有多个worker进程,分担队列里面的任务。这就有点负载均衡的策略在里面了。尽量做到每个进程的工作量比较平均,而且是完成了一个任务才接 第二个任务。看看我们的实现吧。

$ cd tut2

$ vim manager.py (代码如下)

$ vim worker.py (代码如下)

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

  1. #!/usr/bin/env python  

  2. # -*- coding: utf-8 -*-  

  3. import pika  

  4. import sys  

  5. parameters = pika.ConnectionParameters(host = 'localhost' )  

  6. connection = pika.BlockingConnection(parameters)  

  7. channel = connection.channel()  

  8. channel.queue_declare(queue = 'task_queue' , durable = True )  

  9. message = ' ' .join(sys.argv[ 1 :]) or "Hello World!"  

  10. channel.basic_publish(exchange = '',  

  11.                        routing_key = 'task_queue' ,  

  12.                        body = message,  

  13.                        properties = pika.BasicProperties(  

  14.                           delivery_mode = 2 , # make message persistent  

  15.                        ))  

  16. print " [x] Sent %r" % (message,)  

  17. connection.close()  

 

其中第8行的 durable=True 声明了队列需要持久化,第14行的 delivery_mode = 2 声明了队列的消息需要持久化。

 

然后写一个接收消息脚本:worker.py

  1. #!/usr/bin/env python  

  2. # -*- coding: utf-8 -*-  

  3. import pika  

  4. import time  

  5. connection = pika.BlockingConnection(pika.ConnectionParameters(  

  6.          host = 'localhost' ))  

  7. channel = connection.channel()  

  8. channel.queue_declare(queue = 'task_queue' , durable = True )  

  9. print ' [*] Waiting for messages. To exit press CTRL+C'  

  10.    

  11. def callback(ch, method, properties, body):  

  12.      print " [x] Received %r" % (body,)  

  13.      time.sleep( body.count( '.' ) )  

  14.      print " [x] Done"  

  15.      ch.basic_ack(delivery_tag = method.delivery_tag)  

  16.    

  17. channel.basic_qos(prefetch_count = 1 )  

  18. channel.basic_consume(callback,  

  19.                        queue = 'task_queue' )  

  20. channel.start_consuming()  

 

其中第15行的 basic_ack 是执行完任务通知rabbitmq,第17行的basic_qos是告诉rabbitmq只有当worker完成了任务以后才分派1条新的消息,实现公平分派。

测试方法,开3个bash,2个跑worker,1个跑manager:

$ 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)了。

$ cd tut3

$ vim emitlog.py (代码如下)

$ vim recelog.py (代码如下)

emitlog.py

  1. #!/usr/bin/env python  

  2. import pika  

  3. import sys  

  4.    

  5. connection = pika.BlockingConnection(pika.ConnectionParameters(  

  6.          host = 'localhost' ))  

  7. channel = connection.channel()  

  8.    

  9. channel.exchange_declare(exchange = 'logs' ,  

  10.                           type = 'fanout' )  

  11.    

  12. message = ' ' .join(sys.argv[ 1 :]) or "info: Hello World!"  

  13. channel.basic_publish(exchange = 'logs' ,  

  14.                        routing_key = '',  

  15.                        body = message)  

  16. print " [x] Sent %r" % (message,)  

  17. connection.close()  

 

recelog.py

 

  1. #!/usr/bin/env python  

  2. import pika  

  3.    

  4. connection = pika.BlockingConnection(pika.ConnectionParameters(  

  5.          host = 'localhost' ))  

  6. channel = connection.channel()  

  7. channel.exchange_declare(exchange = 'logs' ,  

  8.                           type = 'fanout' )  

  9. result = channel.queue_declare(exclusive = True )  

  10. queue_name = result.method.queue  

  11. channel.queue_bind(exchange = 'logs' ,  

  12.                     queue = queue_name)  

  13. print ' [*] Waiting for logs. To exit press CTRL+C'  

  14.    

  15. def callback(ch, method, properties, body):  

  16.      print " [x] %r" % (body,)  

  17.    

  18. channel.basic_consume(callback,  

  19.                        queue = queue_name,  

  20.                        no_ack = True )  

  21. channel.start_consuming()  

 

测试:

和前一个实例差不多。开3个bash,2个跑recelog,1个跑emitlog。查看recelog是否都收到emitlog发送的消息。代码里面用 了一个fanout(意思是成扇形展开)类型的exchange,只要和exchange绑定的queue都能收到一份消息的 copy,routing_key会被忽略掉。

 

5,路由模式 (选择接收信息)

$ cd tut4

$ vim emitlog.py (代码如下)

$ vim recelog.py (代码如下)

emitlog.py

  1. #!/usr/bin/env python  

  2. import pika  

  3. import sys  

  4.    

  5. connection = pika.BlockingConnection(pika.ConnectionParameters(  

  6.          host = 'localhost' ))  

  7. channel = connection.channel()  

  8. channel.exchange_declare(exchange = 'direct_logs' ,  

  9.                           type = 'direct' )  

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

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

  12. channel.basic_publish(exchange = 'direct_logs' ,  

  13.                        routing_key = severity,  

  14.                        body = message)  

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

  16. connection.close()  

 

这里声明exchange时类型定义为direct(直接匹配),就是说只有当一个信息的routing_key和队列的binding_key一 致时,信息才会被放入到这个队列。消息发布给exchange时必须带上routing_key。其实在消息生产端,队列这个概念是透明的。

 

recelog.py

  1. #!/usr/bin/env python  

  2. import pika  

  3. import sys  

  4.    

  5. connection = pika.BlockingConnection(pika.ConnectionParameters(  

  6.          host = 'localhost' ))  

  7. channel = connection.channel()  

  8.    

  9. channel.exchange_declare(exchange = 'direct_logs' ,  

  10.                           type = 'direct' )  

  11. result = channel.queue_declare(exclusive = True )  

  12. queue_name = result.method.queue  

  13. severities = sys.argv[ 1 :]  

  14. if not severities:  

  15.      print >> sys.stderr, "Usage: %s [info] [warning] [error]" % \  

  16.                           (sys.argv[ 0 ],)  

  17.      sys.exit( 1 )  

  18. for severity in severities:  

  19.      channel.queue_bind(exchange = 'direct_logs' ,  

  20.                         queue = queue_name,  

  21.                         routing_key = severity)  

  22. print ' [*] Waiting for logs. To exit press CTRL+C'  

  23. def callback(ch, method, properties, body):  

  24.      print " [x] %r:%r" % (method.routing_key, body,)  

  25. channel.basic_consume(callback,  

  26.                        queue = queue_name,  

  27.                        no_ack = True )  

  28. channel.start_consuming()  

 

这里首先定义exchange,和消息发送端是一样的。然后定义队列,队列是自动命名,并且只要进程终止,队列就会终止。然后把队列和 exchange绑定,绑定时的routing_key是用户输入的,如果输入多个key,就做多次的绑定。注意这里的队列还是一个。如果你需要建立两个 队列,就得跑两次这个python脚本。

 

6,topic和rpc

官方tutorial还有两个高级一点的实例,topic和rpc,这里就不作说明了,留着大家学学英文吧 :)

RabbitMQ提供了很多消息队列客户端代码,比如python,java,c等等,大家可以根据产品或项目的实际情况选择。关键是原理必须搞懂。

本文转载自:http://yidao620c.iteye.com/blog/1947338

沉淀岁月
粉丝 27
博文 257
码字总数 91615
作品 0
朝阳
高级程序员
私信 提问
在CentOS上安装rabbitmq

转自:http://flyingdutchman.iteye.com/blog/1887283 这文章写得很好,除了安装软件编译时间比较长之外,安装这个几乎没出现什么错误。现在去配置下rabbitmq,马上就可以使用了。 在本节中我...

mac_zhao
2014/09/28
0
0
CentOS 6.4下Rabbitmq安装

在 CentOS 6.4上安装python* 注意啊,自己手动安装python2.7.5,不要动系统上面其他的版本 yum install zlib yum install zlib-devel 下载python-2.7.6.tar.gz文件 tar -xzvf python-2.7.6.t...

mac_zhao
2014/09/26
0
0
Python线性优化 第二部分: 在云中构建一个可扩展的架构

概要 对于脚本语言,例如Python,生成可扩展的、可平行化的代码需要仔细的思考。这篇系列文章探索了使用Pyomo在Python中模仿优化程序的方法。Pyomo是一个开源工具,可以用于定义象征问题,生成...

renwofei423
2013/03/05
1K
0
python操作rabbitmq操作数据

一、RabbitMQ 消息队列介绍 RabbitMQ也是消息队列,那RabbitMQ和之前python的Queue有什么区别么? 1 2 3 4 5 6 7 8 2)Install RabbitMQ Server 1 2 3)use RabbitMQ Server 1 2 2、基本示例...

数据架构师
2018/08/27
0
0
配置RabbitMQ默认群集模式

RabbitMQ是什么? MQ(Msaaage Queue,消息队列)是一种应用程序对应用程序的通信方式。应用程序通过读写出入队列的消息(针对应用程序的数据)来通信,而无须专用链接来连接它们。消息传递指...

cchenyz
2018/07/31
0
0

没有更多内容

加载失败,请刷新页面

加载更多

《生活的难题》的读后感3600字范文

《生活的难题》的读后感3600字范文: 假如我们对丑的事物也能够像对待美的事物那般抱持一种开放、接纳的心态,拥有相同的感受力,那么我们便会发现它们都是充满意义的,而这种认识会使生活变...

原创小博客
20分钟前
1
0
Linux learn(四)

7. Linux磁盘与文件系统管理 文件系统通常会将数据放在不同的区块,权限与属性放置到inode中,至于实际数据则放到datab lock区块中,另外,还有一个超级区块(superblock)会记录整个文件系统...

lazy~
28分钟前
1
0
微信公众号开发(四)

微信公众号开发时常需要一个用户授权绑定的过程。关于微信公众号的用户绑定,一般有如下两种实现方式: (1)通过发送短信验证码的方式; (2)使用用户登录时,向后端传递openid的方式。 使...

织梦之魂
今天
4
0
设计模式-工厂模式

工厂模式 工厂模式(Factory Pattern)是 Java 中最常用的设计模式之一。这种类型的设计模式属于创建型模式,它提供了一种创建对象的最佳方式。 在工厂模式中,我们在创建对象时不会对客户端...

HOT_POT
今天
9
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部