使用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
#!/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发送一个消息
$ python send.py 'Hello World!'
$ python send.py '你好刀哥'
$ /usr/local/rabbitmq/sbin/rabbitmqctl list_queues
Listing queues ...
hello 2
... done .
如果你也看到hello队列里面有一个消息的话,就证明可以发消息了。
然后写一个接收消息脚本:receive.py
#!/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
$ 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
#!/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
#!/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:
$ 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
#!/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
#!/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,路由模式 (选择接收信息)
$ cd tut4
$ vim emitlog.py (代码如下)
$ vim recelog.py (代码如下)
emitlog.py
#!/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
#!/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等等,大家可以根据产品或项目的实际情况选择。关键是原理必须搞懂。
其他资源:
中文入门篇: http://adamlu.net/dev/2011/09/rabbitmq-get-started/
博客新地址:http://yidao620c.github.io
相关推荐
本文详细介绍了如何使用Python与RabbitMQ进行消息队列的搭建与应用实践。从基本的概念出发,逐步深入到具体的实现细节和技术要点,旨在为读者提供一个全面的理解框架。希望读者能够在实际项目中灵活运用所学知识,...
这意味着Python开发的应用程序可以利用RabbitMQ强大的消息队列功能来提高系统的可扩展性和可靠性。 在Python中使用RabbitMQ主要涉及几个方面:首先需要安装RabbitMQ的Python客户端库(例如pika),然后在Python代码...
Python操作RabbitMQ是将Python编程语言与RabbitMQ消息中间件相结合,以实现应用程序间的高效、可靠的消息通信。RabbitMQ是一个基于AMQP(Advanced Message Queuing Protocol)的开源消息代理,它由Erlang语言开发,...
MQTT(Message Queuing Telemetry Transport)和RabbitMQ都是消息队列系统,它们在物联网(IoT)和分布式系统中广泛使用。MQTT是轻量级的发布/订阅协议,适用于低带宽、高延迟或不可靠的网络环境,而RabbitMQ是一个...
标题 "celery+ rabbitMq + python windows环境配置" 涉及到的是在Windows操作系统上集成使用Celery和RabbitMQ的流程。Celery是一个分布式任务队列,主要用于处理异步任务,而RabbitMQ则是一个开源的消息代理,常被...
### RabbitMQ基本概念与应用 #### 一、引言 RabbitMQ作为一款高效且可靠的开源消息代理系统,被广泛应用于各种应用场景之中,比如...通过对RabbitMQ的理解和掌握,可以帮助开发人员构建出更加稳定可靠的应用系统。
本文将详细介绍使用 Python 语言结合 pika 库实现的六种常见 RabbitMQ 消息模型,并提供示例代码。 #### 二、RabbitMQ 与 Redis 的对比 尽管 Redis 也可以用作消息中间件,但在生产环境中更多地推荐使用 RabbitMQ...
总之,`python_logging_rabbitmq`库是Python开发人员与RabbitMQ结合进行日志管理的一个强大工具,它简化了分布式环境中日志数据的收集和处理流程,提升了系统的可维护性和可扩展性。通过合理配置和使用,可以极大地...
**标题与描述解析** 标题和描述中提到的“amqpstorm”是一个专为Python设计的RabbitMQ客户端和管理库,具有线程安全的特性。...在实际开发中,了解并掌握amqpstorm的使用方法,将有助于提高代码质量和项目的稳定性。
RabbitMQ是一种流行的开源消息代理软件,它实现了高级消息队列协议(AMQP),主要用于在...通过这些内容,开发人员可以更好地理解如何在Python项目中集成RabbitMQ作为消息中间件,从而构建高效率、高可靠性的应用系统。
作为Python库,`miros_rabbitmq`适用于各种使用Python作为主要开发语言的项目。Python以其简洁的语法和丰富的库生态系统,成为了许多开发者的选择,尤其是在数据处理、Web开发和自动化脚本等领域。 总结,`miros_...
1. **多语言支持**:RabbitMQ提供了丰富的客户端库,如Java、Python、Ruby、.NET、Go等,方便不同语言的开发者使用。 2. **连接与认证**:客户端需要连接到RabbitMQ服务器,通常使用用户名和密码进行身份验证。 3. *...
`stock_learning_rabbitmq-0.0.6-py3-none-any.whl`是一个特定于Python的库,它专注于使用RabbitMQ进行分布式消息传递。这个库的版本号是0.0.6,表明它是该软件的早期版本,可能还处于不断更新和完善的过程中。 **...
标签“python 开发语言 后端 Python库”表明该库是用于Python开发的后端工具。这意味着它可能用于构建服务器端应用,处理数据传输或者服务间通信。 根据这些信息,我们可以推测`psql2rabbitmq`是一个Python库,它的...
3. 启动与管理:可以通过命令行启动`rabbitmq-server`服务,或者使用RabbitMQ提供的管理控制台(默认端口55672)进行监控和管理。 4. 插件管理:RabbitMQ有许多内置插件,如管理界面(rabbitmq_management)、镜像...
标题 "celery+ rabbitMq + python linux 下 例子 crontab例子" 涉及的是在Linux环境下使用Python的Celery分布式任务队列和RabbitMQ消息中间件结合使用,以及如何配合crontab进行定时任务的设置。下面将详细讲解这些...
RabbitMQ是一个开源的消息队列系统,它基于AMQP...通过运行和调试这个示例,你可以更好地掌握如何创建生产者和消费者,如何定义交换机和队列,以及如何设置路由规则,这些都是使用RabbitMQ进行分布式系统开发的基础。
1. **RabbitMQ概述**:RabbitMQ是一个用Erlang编程语言开发的消息中间件,它的核心功能是作为消息代理,接收并转发应用程序之间的消息。它支持多种消息协议,但最常用的是AMQP,该协议定义了消息的格式和交换方式。 ...