RabbitMQ是一个在AMQP基础上完整的，可复用的企业消息系统。他遵循Mozilla Public License开源协议。
MQ全称为Message Queue，消息队列是一种应用程序对应程序的通信方法。应用程序通过读写出入队列的消息（针对应用程序的数据）来通信，而无需专用连接来链接他们。消息传递指的是程序之间通过在消息中发送数据进行通信，而不是通过直接调用彼此来通信，直接调用通常是用于诸如远程过程调用的技术。排队指的是应用程序通过队列来通信。队列的使用除去了接收和发送应用程序同时执行的要求。
RabbitMQ的安装：
1、安装配置epel源 $ rpm - ivh http: / / dl.fedoraproject.org / pub / epel / 6 / i386 / epel - release - 6 - 8.noarch .rpm 2、安装erlang $ yum - y install erlang 3、安装RabbitMQ $ yum - y install rabbitmq - server 4、启动与停止：service rabbitmq-server start/stop
python安装API：
pip install pika 或者 easy_install pika 或者源码安装： 
生产者 消费者 消费者 2、durable（持久化） 消息不丢失：当RabbitMQ服务宕机，后也不用担心消息的丢失。 生产者 消费者 消费者 订阅者 发布者 5、关键字发送：发送消息时，明确指定某个队列并向其中发送消息，RabbitMQ还支持根据关键字发送，即：队列绑定关键字，发送者将数据根据关键字发送到消息exchange，exchange根据关键字判定应该将数据发送至指定队列。 exchange type=direct 消费者 生产者 消费者 生产者
https: / / pypi.python.org / pypi / pika 对于RabbitMQ来说，生产和消费不再针对内存里的一个queue对象，而是某台服务器上的RabbitMQ Server实现的消息队列。
import pika# 创建连接connection = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.254.129'))channel = connection.channel() # 创建频道channel.queue_declare(queue='hello') # 创建队列hello，若存在则忽略# 向队列hello中发消息， body为发的消息内容channel.basic_publish(exchange='', routing_key='hello', body='Hello World')print('saf')connection.close() import pikaconn = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.254.129'))chanel = conn.channel()chanel.queue_declare(queue='hello')def callback(ch, method, properties, body): print(body)# ch为channel， method为函数名字， property为属性， body为取到的消息chanel.basic_consume(callback, queue='hello', no_ack=True)# no_ack = False为，如果消费者遇到情况（its channel is closed，connection# is closed，or TCP connection is lost）挂掉了，那么，RabbitMQ会重新将该任务添加到队列中print('hehehehe')chanel.start_consuming() 1、acknowledgment 消息不丢失：no_ack = False，如果消费者遇到情况（its channel is closed,connection is closed,or TCP connection is lost）挂掉了，那么，RabbitMQ会重新将该任务添加到队列中。
import pika, timeconn = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.254.129'))chanel = conn.channel()chanel.queue_declare(queue='hello')def callback(ch, method, properties, body): print('消息：', body) time.sleep(10) print('ok') ch.basic_ack(delivery_tag=method.delivery_tag)chanel.basic_consume(callback, queue='hello', no_ack=False)print('我是消费者，等待消息')chanel.start_consuming() import pikaconn = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.254.129'))channel = conn.channel()channel.queue_declare(queue='hello1', durable=True)# delivery_mode=2意思是做持久化channel.basic_publish(exchange='', routing_key='hello1', body='hello world', properties=pika.BasicProperties( delivery_mode=2, ))print('发送消息')conn.close()
import pika, timeconn = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.254.129'))chanel = conn.channel()chanel.queue_declare(queue='hello', durable=True)def callback(ch, method, properties, body): print('消息是：', body) time.sleep(10) ch.basic_ack(delivery_tag=method.delivery_tag)chanel.basic_consume(callback, queue='hello1', no_ack=False)print('等待消息中。。。')chanel.start_consuming() 3、消息获取顺序：默认消息队列里的数据是按照吮吸被消费者拿走（默认消费者按照间隔）
有些时候不需要按照间隔去去任务更好一些。channel.basic(prefetch_count=1)表示谁来谁取，不再按照间隔 import pikaconn = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.254.129'))channel = conn.channel()channel.queue_declare(queue='hello2')def callback(ch, method, properties, body): print('消息是', body) print('ok') ch.basic_ack(delivery_tag=method.delivery_tag)channel.basic_qos(prefetch_count=1)channel.basic_consume(callback, queue='hello2', no_ack=False)print('等待消息中')channel.start_consuming() 4、发布和订阅
发布订阅和简单的消息队列区别在于，发布订阅会将消息发送给所有的订阅者，而消息队列中的数据被消费一次便消失。所以，RabbitMQ实现发布和订阅时，会为每一个订阅者创建一个队列，而发布者发布消息时，会将消息放置在所有相关队列中。exchange type=fanout
# 订阅者import pikaconn = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.254.129'))channel = conn.channel()channel.exchange_declare(exchange='logs', type='fanout')res = channel.queue_declare(exclusive=True)queue_name = res.method.queueprint(queue_name)channel.queue_bind(exchange='logs', queue=queue_name)print('等待消息logs中')def callback(ch, method, properties, body): print('消息是', body)channel.basic_consume(callback, queue=queue_name, no_ack=True)channel.start_consuming() # 发布者import pika, sysconn = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.254.129'))channel = conn.channel()channel.exchange_declare(exchange='logs', type='fanout')msg = ' '.join(sys.argv[1:]) or 'info: Hello World'channel.basic_publish(exchange='logs', routing_key='', body=msg)print(msg)conn.close()
import pika, sysconn = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.254.129'))channel = conn.channel()channel.exchange_declare(exchange='direct_logs', type='direct')res = channel.queue_declare(exclusive=True)queue_name = res.method.queueseverities = sys.argv[1:]if not severities: # sys.stderr.write(sys.argv[0]) # sys.exit(1) severi = input('>>:').strip().split() severities = severifor severity in severities: channel.queue_bind(exchange='direct_logs', queue=queue_name, routing_key=severity)print('等待消息')def callback(ch, method, properties, body): print('收到消息：', method.routing_key, body)channel.basic_consume(callback, queue=queue_name, no_ack=True)channel.start_consuming() import pika, sysconn = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.254.129'))channel = conn.channel()channel.exchange_declare(exchange='direct_logs', type='direct')severity = sys.argv[1] if len(sys.argv) > 1 else 'info'msg = ' '.join(sys.argv[2:]) or 'Hello the Sea.'channel.basic_publish(exchange='direct_logs', routing_key=severity, body=msg)print('发送消息：', msg)conn.close() 6、模糊匹配：在topic类型下，可以让队列绑定几个模糊的关键字，之后发送者将数据发送到exchange， exchange将传入‘路由值’和‘关键字’进行匹配，匹配成功，则将数据发送到指定队列。 exchange type=topic
# 表示可以匹配0个或多个单词* 表示只能匹配一个单词
import pika, sysconn = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.254.129'))channel = conn.channel()channel.exchange_declare(exchange='topic_logs', type='topic')res = channel.queue_declare(exclusive=True)queue_name = res.method.queuebinding_keys = sys.argv[1:]if not binding_keys: # sys.stderr.write(sys.argv[0]) # sys.exit(1) binding_keys = input('>>:').strip().split()for binding_key in binding_keys: channel.queue_bind(exchange='topic_logs', queue=queue_name, routing_key=binding_key)print('等待消息中。。。')def callback(ch, method, properties, body): print(method.routing_key, body)channel.basic_consume(callback, queue=queue_name, no_ack=True)channel.start_consuming() import pika, sysconn = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.254.129'))channel = conn.channel()channel.exchange_declare(exchange='topic_logs', type='topic')routing_key = sys.argv[1] if len(sys.argv) > 1 else 'info.boy'msg = ' '.join(sys.argv[2:]) or 'Hello the sky'channel.basic_publish(exchange='topic_logs', routing_key=routing_key, body=msg)print(routing_key, msg)conn.close()