在上一章节中,我们学习了
RabbitMQ:基础,运行和管理,在本节中我们来学习一下如何发布和订阅rabbitmq消息,我们使用python来开发应用程序。
我们先来看一下发布/订阅rabbitmq消息的流程。先来看消息生产者Publisher如何发布消息流程:
引用
1、打开RabbitMQ连接;
2、创建Channel通道;
3、声名一个exchange交换机;
4、生成一条消息;
5、发布消息;
6、关闭Channel通道;
7、关闭RabbitMQ连接。
然后我们再来看一下Consumer订阅流程和消费消息:
引用
1、打开RabbitMQ连接;
2、创建Channel通道;
3、声名一个exchange交换机;
4、声名一个queue队列;
5、将queue队列绑定到exchange交换机;
6、消费消息;
7、关闭Channel通道;
8、关闭RabbitMQ连接。
在写代码之前,我们需要先安装一下使用python链接RabbitMQ的依赖库——pika。我们使用easy_install来安装(easy_install的具体安装过程请参看
Centos 6 上安装easy_instal):
引用
# ./easy_install pika
好了,准备工作已经好了,现在我们开始写代码:
引用
# mkdir -p /data/rabbitmq-pika
# cd /data/rabbitmq-pika
# mkdir c2
# cd c2
在本节中,我们主要来开发一个带Publisher confirms的消息发布和订阅程序。我们先来开发消息生产者hello_world_producer_with_comfirms.py:
引用
# touch hello_world_producer_with_comfirms.py
# chmod +x hello_world_producer_with_comfirms.py
# vi hello_world_producer_with_comfirms.py
hello_world_producer_with_comfirms.py代码如下:
#coding=utf-8
import pika,sys
from pika import spec
#在"/"虚拟主机vhost上通过用户guest建立channel通道
user_name = 'guest'
user_passwd = 'guest'
target_host = 'localhost'
vhost = '/'
cred = pika.PlainCredentials(user_name,user_passwd)
conn_params = pika.ConnectionParameters(target_host,
virtual_host = vhost,
credentials = cred)
conn_broker = pika.BlockingConnection(conn_params)
channel = conn_broker.channel()
#定义消息发布后publisher接受到的确认信息处理函数
def confirm_handler(frame):
if type(frame.method) == spec.Confirm.SelectOk:
"""生产者创建的channel处于‘publisher comfirms’模式"""
print 'Channel in "confirm" mode!'
elif type(frame.method) == spec.Basic.Nack:
"""生产者接受到消息发送失败并且消息丢失的消息"""
print 'Message lost!'
elif type(frame.method) == spec.Basic.ack:
if frame.method.delivery_tag in msg_ids:
"""生产者接受到成功发布的消息"""
print 'Confirm received!'
msg_ids.remove(frame.method.delivery_tag)
#将生产者创建的channel处于"publisher confirms"模式
channel.confirm_delivery(callback = confirm_handler)
#创建一个direct类型的、持久化的、没有consumer时队列是否自动删除的exchage交换机
channel.exchange_declare(exchange = 'hello-exch',
type = 'direct',
passive = False,
durable = True,
auto_delete = False)
#使用接收到的信息创建消息
#使用接收到的信息创建消息
msg = sys.argv[1]
msg_props = pika.BasicProperties()
msg_props.content_type = 'text/plain'
#持久化消息
msg_props.delivery_mode = 2
msg_ids = []
print 'ready to publish...'
#发布消息
channel.basic_publish(body = msg,
exchange = 'hello-exch',
properties = msg_props,
routing_key = 'hala')
print 'published!'
msg_ids.append(len(msg_ids) + 1)
print len(msg_ids)
channel.close()
conn_broker.close()
接下来,我们先来开发一个带消息确认信息的消费者hello_world_consumer_with_ack.py:
引用
# hello_world_consumer_with_ack.py
# chmod +x hello_world_consumer_with_ack.py
# vi hello_world_consumer_with_ack.py
hello_world_consumer_with_ack.py代码如下:
#coding=utf-8
import pika
#在"/"虚拟主机vhost上通过用户guest建立channel通道
user_name = 'guest'
user_passwd = 'guest'
target_host = 'localhost'
vhost = '/'
cred = pika.PlainCredentials(user_name,user_passwd)
conn_params = pika.ConnectionParameters(target_host,
virtual_host = vhost,
credentials = cred)
conn_broker = pika.BlockingConnection(conn_params)
conn_channel = conn_broker.channel()
#创建一个direct类型的、持久化的、没有consumer时,队列是否自动删除exchage交换机
conn_channel.exchange_declare(exchange = 'hello-exch',
type = 'direct',
passive = False,
durable = True,
auto_delete = False)
#创建一个持久化的、没有consumer时队列是否自动删除的名为“hell-queue”
conn_channel.queue_declare(queue = 'hello-queue',
durable = True,
auto_delete = False)
#将“hello-queue”队列通过routing_key绑定到“hello-exch”交换机
conn_channel.queue_bind(queue = 'hello-queue',
exchange = 'hello-exch',
routing_key = 'hala')
#定义一个消息确认函数,消费者成功处理完消息后会给队列发送一个确认信息,然后该消息会被删除
def ack_info_handler(channel,method,header,body):
"""ack_info_handler """
print 'ack_info_handler() called!'
if body == 'quit':
channel.basic_cancel(consumer_tag = 'hello-hala')
channel.stop_sonsuming()
else:
print body
channel.basic_ack(delivery_tag = method.delivery_tag)
conn_channel.basic_consume(ack_info_handler,
queue = 'hello-queue',
no_ack = False,
consumer_tag = 'hello-hala')
print 'ready to consume msg...'
conn_channel.start_consuming()
测试:
引用
# /opt/mq/rabbitmq/sbin/rabbitmq-server start &
# python ./hello_world_consumer_with_ack.py
# python ./hello_world_producer_with_comfirms.py 'hello-world'
对于上面的hello_world_producer_with_comfirms.py,本人调试了很长时间,期间总是在:
引用
#将生产者创建的channel处于"publisher confirms"模式
channel.confirm_delivery(callback = confirm_handler)
报错,最后在重新下载pika-0.9.13.tar.gz并安装后才成功执行上述代码:
引用
# wget https://pypi.python.org/packages/source/p/pika/pika-0.9.13.tar.gz --no-check-certificate
# chmod +x pika-0.9.13.tar.gz
# tar xzvf pika-0.9.13.tar.gz
# cd pika-0.9.13
# python setup.py install
分享到:
相关推荐
您现在可以使用使用凭证:guest / guest访问RabbitMQ管理 拉图: docker pull csarcom/qpaca:latest 起始发布者: docker run -d -p 8000:8000 --link my_rabbitmq:my_rabbitmq -v /$(pwd):/app csarcom/qpaca ...
AMQP的主要特征是面向消息、队列、路由(包括点对点和发布/订阅)、可靠性、安全。 RabbitMQ是一个开源的AMQP实现,服务器端用Erlang语言编写,支持多种客户端,如:Python、Ruby、.NET、Java、JMS、C、PHP、...
AMQP的主要特征是面向消息、队列、路由(包括点对点和发布/订阅)、可靠性、安全。 RabbitMQ是一个开源的AMQP实现,服务器端用Erlang语言编写,支持多种客户端,如:Python、Ruby、.NET、Java、JMS、C、PHP、...
AMQP的主要特征是面向消息、队列、路由(包括点对点和发布/订阅)、可靠性、安全。 RabbitMQ是一个开源的AMQP实现,服务器端用Erlang语言编写,支持多种客户端,如:Python、Ruby、.NET、Java、JMS、C、PHP、...
有关中型故事的详细解释,请参见:发布/订阅角色定义发布者:声明了一个单独的RabbitMQ队列生成器,并将其添加到Celery的默认producer_pool中,该默认生成器被拉出并用于在Celery任务中将新消息发布到该队列。...
AMQP的主要特征是面向消息、队列、路由(包括点对点和发布/订阅)、可靠性、安全。 Github:https://github.com/rabbitmq 官网地址:http://www.rabbitmq.com 安装RabbitMQ 安装RabbitMQ 可以参考之前的文章 ...
事件:可插入且可靠的发布订阅消息(例如,RabbitMQ支持) 流程管理 在readthedocs.org上有。 安装(作为依赖项) # py-monotime requires python headers, and gevent and cython require build-essential $ ...
AMQP的主要特征是面向消息、队列、路由(包括点对点和发布/订阅)、可靠性、安全。RabbitMQ是一个开源的AMQP实现,服务器端用Erlang语言编写,支持多种客户端,如:Python、Ruby、.NET、Java、JMS、C、PHP、...
rabbitmq默认有7个交换机,其中amq.rabbitmq.log为系统日志的交换机,这个日志为topic类型,会有三个等级的(routing_key)的日志发送到这个交换机上。 代码如下 #!/usr/bin/env python # -*- coding: utf-8 -*- ...
AMQP的主要特征是面向消息、队列、路由(包括点对点和发布/订阅)、可靠性、安全。RabbitMQ是一个开源的AMQP实现,服务器端用Erlang语言编写,支持多种客户端,如:Python、Ruby、.NET、Java、JMS、
python入门到高级全栈工程师培训视频学习资料;本资料仅用于学习,请查看后24小时之内删除。 【课程内容】 第1章 01 计算机发展史 02 计算机系统 03 小结 04 数据的概念 05 进制转换 06 原码补码反码 07 物理层和...
RabbitMQ监控平台 功能 提供web界面订阅队列,后台线程定时轮询队列堆积数量,通过邮件的方式告警。 环境依赖 python2 mongo flask 使用说明 配置conf/env.ini [global] ; 全局配置,对所有队列统一指定消息堆积阀值...
SimpleMQ基本上是Redis Pub / Sub(所有订阅者都接收到已发布的消息)和RabbitMQ(您可以拥有多个使用方,并且代理将订阅的使用方之间的使用交织在一起)的混合体。 因此,您可以将每个队列配置为不同的行为,可以...
Redis 发布订阅及本节作业 第12周 上节回顾 数据库介绍 mysql基本使用 mysql数据类型与创建表 mysql 增删改查 mysql 外键关联 mysql 连接查询 mysql 事务与索引 mysql python交互 sqlalchemy orm介绍 sqlalchemy ...
命令解析器Shell 2.3Hadoop⽣态体系 ⼤数据存储框架Hadoop 分布式服务管理框架zookeeper hadoop⾼可⽤HA 数据仓库hive 数据查询框架impala+kudu+clouderManager ⽇志采集传输框架flume/idea 分布式发布-订阅消息系统...
在生产中,我建议使用传统的消息代理/发布订阅服务(例如Kafka,RabbitMQ等),但是为了简单起见,我使用MongoDB的构建了该服务。 什么是发件箱模式? 发件箱模式用于保证对数据库的更新和事件的发送是原子发生的...
该软件包使用侦听RabbitMQ队列中的任务。 通过访问OpenAgua数据库 安装 从点子 该软件包可通过PyPI获得(请参阅): pip install openagua-engine 从来源 python setup.py install应该可以解决问题。 入门 以下...