我们先来看一下发布/订阅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