`
flyingdutchman
  • 浏览: 358086 次
  • 性别: Icon_minigender_1
  • 来自: 上海
社区版块
存档分类
最新评论

RabbitMQ:使用python发布/订阅消息

阅读更多
        在上一章节中,我们学习了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

       
分享到:
评论
1 楼 fbwfbi 2014-03-14  
发现使用pika-0.9.13的版本依然出错:
Traceback (most recent call last):
  File "hello_world_producer_with_comfirms.py", line 34, in <module>
    channel.confirm_delivery(callback = confirm_handler)
TypeError: confirm_delivery() got an unexpected keyword argument 'callback'

将confirm_delivery(callback = confirm_handler)去掉里面的callback参数即可

相关推荐

    RabbitMQ实战:高效部署分布式消息队列 pdf版

    - **管理控制台**:使用Web管理界面监控和管理RabbitMQ实例。 - **环境配置**:配置RabbitMQ的节点、用户权限、虚拟主机等。 **3. 生产者与消费者** - **创建生产者**:编写代码发送消息到RabbitMQ,支持多种编程...

    rabbitmq安装包.zip

    消费者则订阅队列,接收来自RabbitMQ的消息。 **RabbitMQ高级特性** 1. 工作队列:通过多个消费者竞争消费消息,实现负载均衡,提高处理效率。 2. 消息确认:消费者可以通过ack机制确认消息已被处理,RabbitMQ会...

    Qpaca:使用falcon(REST),Docker和RabbitMQ发布-订阅包装器的实现

    您现在可以使用使用凭证:guest / guest访问RabbitMQ管理 拉图: docker pull csarcom/qpaca:latest 起始发布者: docker run -d -p 8000:8000 --link my_rabbitmq:my_rabbitmq -v /$(pwd):/app csarcom/qpaca ...

    RabbitMQDemo_RabbitMQ发布与订阅Demo_rabbitmq_JSON_中间件_

    总结来说,"RabbitMQDemo_RabbitMQ发布与订阅Demo_rabbitmq_JSON_中间件_"是一个演示项目,其核心在于利用RabbitMQ作为中间件,通过发布者发送JSON格式的消息到队列,由订阅者从队列中接收并处理这些消息。...

    rabbitMQ发布和订阅共7页.pdf.zip

    1. **发布**: 生产者发布消息到RabbitMQ时,会指定一个特定的主题(topic)。这个主题可以是任意字符串,例如"news.europe"或"weather.us”。 2. **交换器类型**: 在发布/订阅模式中,通常使用`topic`类型的交换器...

    RabbitMQ 发布者订阅者

    在RabbitMQ的实际操作中,开发者可以使用其官方提供的客户端库,如Python的pika库,Java的RabbitMQ Java Client等,编写发布者和订阅者的代码。同时,管理界面RabbitMQ Management提供了图形化工具,方便进行交换机...

    RabbitMQ 3.7安装及Python使用

    连接建立后,可以创建消息通道,定义消息的发布和订阅,从而实现消息的生产和消费。 值得注意的是,RabbitMQ支持多种协议与消息格式,包括AMQP 0-9-1, AMQP 1.0,以及其他消息协议,如STOMP、MQTT等。因此,Python...

    python rabbitmq

    RabbitMQ支持多种消息传递模式,包括发布/订阅(Publish/Subscribe)、工作队列(Work Queues)、路由(Routing)等,并且具有良好的扩展性和可靠性。 **2.2 关键术语解释** - **生产者(Producer)**: 发送消息的程序或...

    stomp协议客户端调试助手工具

    详细介绍参见博客:《Python实现基于WebSocket的stomp协议的小巧调试助手工具》,文章地址:https://blog.csdn.net/yyz_1987/article/details/143111710,可以用来调试stomp协议的通信和订阅、发布消息。 stomp协议...

    rabbitmq-server-generic-unix-2.8.2.tar.gz

    通过创建生产者和消费者,你可以发送和接收消息,实现异步处理、任务队列、发布/订阅等多种应用场景。 总结来说,RabbitMQ是一个强大的消息中间件,提供可靠的消息传输和队列管理功能。2.8.2版本的RabbitMQ适用于...

    封装好的mqtt和rabbitmq可以直接使用

    MQTT是轻量级的发布/订阅协议,适用于低带宽、高延迟或不可靠的网络环境,而RabbitMQ是一个开源的消息代理,支持多种消息协议,包括MQTT。 在给定的场景中,设备通过RabbitMQ来推送消息到MQTT服务器。RabbitMQ作为...

    Sqlite3+RabbitMQ+Celery Python从零开始搭建一个生产者消费者服务模型配套安装文件

    在Python中,我们可以使用pika或者amqpstorm等库与RabbitMQ进行交互。 Celery是一个异步任务队列,它使用消息传递通过分布式系统执行任务。Celery支持多种消息代理,如RabbitMQ、Redis等,它简化了在Python应用中...

    RabbitMQ实战-随书源码

    随书源码“RabbitMQ实战-随书源码”和“RabbitMQ in action 和 RabbitMQ实战-高效部署分布式消息队列”的配套代码,提供了丰富的示例,帮助读者深入理解RabbitMQ的使用和实践。 1. **RabbitMQ基础** - **AMQP协议*...

    RabbitMQ实战 高效部署分布式消息队列完整版带书签

    AMQP的主要特征是面向消息、队列、路由(包括点对点和发布/订阅)、可靠性、安全。 RabbitMQ是一个开源的AMQP实现,服务器端用Erlang语言编写,支持多种客户端,如:Python、Ruby、.NET、Java、JMS、C、PHP、...

    rabbitmq-server3.7.8.zip

    2. **消息模型**:RabbitMQ采用发布/订阅、路由、队列和工作队列等消息模式,以满足不同类型的通信需求。发布者将消息发送到交换机,交换机根据预设的路由规则将消息分发到相应的队列,消费者从队列中获取并处理消息...

    RabbitMQ消息中间件技术精讲

    RabbitMQ支持多种消息传递模式,包括简单模式、工作队列模式、发布/订阅模式、路由模式、主题模式等,适用于各种复杂的应用场景。 #### 二、RabbitMQ的工作原理 RabbitMQ作为一款消息中间件,其核心组件主要包括...

    rabbitmq-server-3.8.9.exe

    - **发布/订阅模型**:生产者发布消息到主题,所有订阅该主题的消费者都能收到消息。 - **直接交换**:消息根据精确匹配的路由键被发送到指定队列。 - **头部交换**:基于消息的headers属性进行路由。 - **扇出...

    rabbitMQ代码案例 简单入门

    3. **消息发布/订阅模式**:理解生产者如何发布消息到交换器,以及消费者如何订阅队列来获取消息。 4. **工作队列模式**:学习如何使用RabbitMQ实现任务调度,通过多个消费者并行处理任务以提高效率。 5. **路由与...

    TongLINK/Q消息中间件(说明)

    - **发布/订阅(Pub/Sub)**:消息发布者将消息发送到特定主题,订阅者可以选择感兴趣的主题来接收消息。 - **路由**:消息可以通过路由机制被分发到多个目标。 - **集群**:通过集群技术提高系统的可用性和扩展...

    rabbitmq安装包

    - **发布/订阅(Publish/Subscribe)**:生产者发布消息到一个主题交换器,消费者订阅特定的路由键。所有匹配的订阅者都会收到消息。 - **工作队列(Work Queues)**:也称为FIFO(先进先出)队列,多个消费者从同一个...

Global site tag (gtag.js) - Google Analytics