RabbitMQ是AMQP（网络的消息传输协议）的一种实现。可复用的企业消息系统。
RabbitMQ安装：
for Linux： 安装配置epel源 $ rpm -ivh http://dl.fedoraproject.org/pub/epel/6/i386/epel-release-6-8.noarch.rpm 安装erlang $ yum -y install erlang 安装RabbitMQ $ yum -y install rabbitmq-server 注意：service rabbitmq-server start/stop
安装API：
pip install pika or easy_install pika or 源码 https://pypi.python.org/pypi/pika
使用API操作RabbitMQ
#!/usr/bin/env python # -*- coding:utf-8 -*- import pika # ######################### 生产者 (创造数据)######################### connection = pika.BlockingConnection(pika.ConnectionParameters( host='localhost'))#封装socket逻辑部分 channel = connection.channel()#拿到句柄,建立连接,操作后面部分 channel.queue_declare(queue='hello', durable=True)#创建队列,hello为队列名 durable 消息不丢失 channel.basic_publish(exchange='', routing_key='hello',#把body数据放到队列里面 body='Hello World!', propertiee = pika.BaseucProperties( delivery_mode = 2,# make message persistent 使消息持久 ))#数据 print(" [x] Sent 'Hello World!'") connection.close() #!/usr/bin/env python # -*- coding:utf-8 -*- import pika # ########################## 消费者(取数据) ########################## connection = pika.BlockingConnection(pika.ConnectionParameters( host='localhost')) channel = connection.channel() channel.queue_declare(queue='hello')#创建队列,hello为队列名,不知道生产者,消费者谁先连接所以需要两个都创建 def callback(ch, method, properties, body):# 回调函数, print(" [x] Received %r" % body) import time time.sleep(10) print('ok') ch.basic_ack(delivery_tag=method.delivery_tag) #调整为有应答 channel.basic_qos(prefetch_count=1)# 默认消息队列里的数据是按照顺序被取走的，表示谁来谁取，不再按照奇偶数排列 channel.basic_consume(callback,#函数,取到函数后执行上面的回调函数 queue='hello',# 必须要要有队列名, 取到的数据放到body里面 no_ack=False)#no-ack ＝ False如果消费者遇到情况挂掉了，那么，RabbitMQ会重新将该任务添加到队列中 print(' [*] Waiting for messages. To exit press CTRL+C') channel.start_consuming()
exchange工作模型（fanout，direct，topic）
发布订阅：
exchange type = fanout
发布订阅和简单的消息队列区别在于，发布订阅会将消息发送给所有的订阅者，而消息队列中的数据被消费一次便消失。所以，RabbitMQ实现发布和订阅时，会为每一个订阅者创建一个队列，而发布者发布消息时，会将消息放置在所有相关队列中。
##################发布者############################ 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() ######################订阅者############################## 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()
关键字发送：
exchange type = direct
之前事例，发送消息时明确指定某个队列并向其中发送消息，RabbitMQ还支持根据关键字发送，即：队列绑定关键字，发送者将数据根据关键字发送到消息exchange，exchange根据 关键字 判定应该将数据发送至指定队列。
#########################消费者######################## 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: sys.stderr.write("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):#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() #####################################生产者########################### 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 type = topic
在topic类型下，可以让队列绑定几个模糊的关键字，之后发送者将数据发送到exchange，exchange将传入”路由值“和 ”关键字“进行匹配，匹配成功，则将数据发送到指定队列。
- # 表示可以匹配 0 个 或 多个 单词
- * 表示只能匹配 一个 单词
发送者路由值 队列中 old.boy.python old.* -- 不匹配 old.boy.python old.# -- 匹配
################################消费者########################## import pika import sys connection = pika.BlockingConnection(pika.ConnectionParameters( host='localhost')) channel = connection.channel() channel.exchange_declare(exchange='topic_logs', type='topic') result = channel.queue_declare(exclusive=True) queue_name = result.method.queue binding_keys = sys.argv[1:] if not binding_keys: sys.stderr.write("Usage: %s [binding_key]... " % sys.argv[0]) sys.exit(1) for binding_key in binding_keys: channel.queue_bind(exchange='topic_logs', queue=queue_name, routing_key=binding_key) 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() ###############################################生产者################################## import pika import sys connection = pika.BlockingConnection(pika.ConnectionParameters( host='localhost')) channel = connection.channel() channel.exchange_declare(exchange='topic_logs', type='topic') routing_key = sys.argv[1] if len(sys.argv) > 1 else 'anonymous.info' message = ' '.join(sys.argv[2:]) or 'Hello World!' channel.basic_publish(exchange='topic_logs', routing_key=routing_key, body=message) print(" [x] Sent %r:%r" % (routing_key, message)) connection.close()
