探索 OpenStack 之（14）：OpenStack 中 RabbitMQ 的使用

5.1一个简单的使用py-amqplib的Consumer实现

#创建Connection和Channel连接到RabbitMQ服务器
conn=amqp.Connection(host="localhost:5672",userid="guest",password="1111",virtual_host="/",insist=False)
chan=conn.channel()

#创建queue
result=chan.queue_declare(queue="debug",durable=True,exclusive=False,auto_delete=False)

#创建exchange
result=chan.exchange_declare(exchange="sorting_room2",type="topic",durable=True,auto_delete=False,)

#创建binding
result=chan.queue_bind(queue="debug",exchange="sorting_room2",routing_key="*.debug")

#回调函数，当有message到达queue后，该函数会被调用
defrecv_callback(msg):
print'Received:'+msg.body+'fromchannel#'+str(msg.channel.channel_id)
#lChannel.basic_ack(msg.delivery_tag)#如果no_ack=False的话，可以需要发回一个确认

#启动一个consumer，consumer_tag是该consumer的一个唯一标识符
#no_ack=True表示该consumer不会发回确认
chan.basic_consume(queue='debug',no_ack=True,callback=recv_callback,consumer_tag="debugtag")

#等待有消息发到queue
whileTrue:
chan.wait()

#终止该consumer
chan.basic_cancel("testtag")

#关闭connection和channel
chan.close()
conn.close()

5.2一个简单的使用py-amqplib的Producer实现代码

fromamqplibimportclient_0_8asamqp
importsys

#创建connection和channel
conn=amqp.Connection(host="localhost:5672",userid="guest",password="1111",virtual_host="/",insist=False)
chan=conn.channel()

#创建message
msg=amqp.Message(sys.argv[1])
msg.properties["delivery_mode"]=2

#发送message
chan.basic_publish(msg,exchange="sorting_room2",routing_key=(sys.argv[2]))

#关闭connection和channel
chan.close()
conn.close()

5.3使用pika

5.3.1安装pika

wgethttps://pypi.python.org/packages/source/p/pika/pika-0.9.14.tar.gz#md5=b99aad4b88961d3c7e4876b8327fc97c
tarzxvfpika-0.9.14.tar.gz
cdpika-0.9.14
pythonsetup.pyinstall

5.3.2使用pika编程（来源）

#!/usr/bin/envpython
'''
rabbitmqtracescripts.
require(rabbitmq_tracing):
$sudorabbitmq-pluginsenablerabbitmq_tracing
usage:
$sudorabbitmqctltrace_on
$./rabbitmqtrace.py
<<output>>
'''
importsys
importtime
fromoptparseimportOptionParser
importpika

__AUTHOR__='smallfish'
__VERSION__='0.0.1'

def_out(args):
printtime.strftime('%Y-%m-%d%H:%M:%S'),args

def_run(host,port,vhost,user,password):
conn=pika.BlockingConnection(pika.ConnectionParameters(host=host,port=port,virtual_host=vhost,
credentials=pika.PlainCredentials(user,password)))
chan=conn.channel()
def_on_message(ch,method,properties,body):
ret={}
ret['routing_key']=method.routing_key
ret['headers']=properties.headers
ret['body']=body
_out(ret)
_out('startsubscribeamq.rabbitmq.trace')
ret=chan.queue_declare(exclusive=False,auto_delete=True)
queue=ret.method.queue
chan.queue_bind(exchange='amq.rabbitmq.trace',queue=queue,routing_key='#')
chan.queue_bind(exchange='amq.rabbitmq.log',queue=queue,routing_key='#')
chan.basic_consume(_on_message,queue=queue,no_ack=True)
chan.start_consuming()

defmain():
parser=OptionParser('usage:%prog')
parser.add_option('','--host',metavar='host',default='localhost',help='rabbitmqhostaddress,default:%default')
parser.add_option('','--port',metavar='port',default=5672,type='int',help='rabbitmqport,default:%default')
parser.add_option('','--vhost',metavar='vhost',default='/',help='rabbitmqvhost,default:%default')
parser.add_option('','--user',metavar='user',default='guest',help='rabbitmquser,default:%default')
parser.add_option('','--password',metavar='password',default='guest',help='rabbitmqpassword,default:%default')
(options,args)=parser.parse_args()
_run(options.host,options.port,options.vhost,options.user,options.password)

if__name__=='__main__':
main()

6插件和消息追踪