消息队列两个用处：服务间解耦，缓解压力（削峰平谷），以前用过ZMQ、狼厂内部的NMQ，现在接触了java开源的kafka和RabbitMQ。目前先不求甚解，有个大概的认识。

RabbitMQ的安装和入门例子见http://www.rabbitmq.com/，挺全的。安装前需要安装erlang，启动方便。

RabbitMQ是一个强壮的消息队列，安装使用都很容易，支持常用的发布/订阅、消息分发功能。

架构

exchange是消息队列的核心，producer只向exchange发送消息，exchange根据routing key向不同的绑定队列发送消息，consumer从各自的队列接收消息。

使用实例(python)：

producer

import pika

import sys

#producer/consumer通用，与rabbitmq建立连接

connection = pika.BlockingConnection(pika.ConnectionParameters(

        host='localhost'))

channel = connection.channel()

channel.exchange_declare(exchange='logs', type='fanout')#声明创建一个名为logs的exchange，类型为fanout，type有fanout/direct/topic/headers

message = ' '.join(sys.argv[1:]) or "info: Hello World!"

channel.basic_publish(exchange='logs', routing_key='', body=message) #p向某个exchange发送消息，并指出路由key

print " [x] Sent %r" % (message,)

connection.close()

consumer

#同上，建立连接、声明exchange

result = channel.queue_declare(exclusive=True)#声明一个队列，名字由系统产生，exclusive指连接断开后删除队列，queue_declare(queue="xx"）可以自己命名队列

queue_name = result.method.queue

channel.queue_bind(exchange='logs', queue=queue_name) #将队列与名为logs的exchange绑定，不指定exchange则与默认的exchange绑定

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) #consumer接收消息不需向rabbit发送确认

channel.start_consuming()

消息的负载均衡

#解决办法，consumer加如下代码

channel.basic_qos(prefetch_count=1)#每次向consumer发一条消息，直到收到consumer的确认再发下一条

#consumer每次处理完信息后向rabbitmq发送确认。在callback函数结尾处加上ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_consume(callback,queue=queue_name)#默认no_ack=false，rabbitmq会一直等待直到收到确认消息

通信协议AMQP(Advanced Message Queuing Protocol)

AMQP模型中，消息在producer中产生，发送到MQ的exchange上，exchange根据配置的路由方式发到相应的Queue上，Queue又将消息发送给consumer，消息从queue到consumer有push和pull两种方式。 消息队列的使用过程大概如下：

• 客户端连接到消息队列服务器，打开一个channel。
• 客户端声明一个exchange，并设置相关属性。
• 客户端声明一个queue，并设置相关属性。
• 客户端使用routing key，在exchange和queue之间建立好绑定关系。
• 客户端投递消息到exchange。

exchange接收到消息后，就根据消息的key和已经设置的binding，进行消息路由，将消息投递到一个或多个队列里。

exchange支持的分发类型

#如上文

#p:

channel.basic_public(exchange='logs',routing_key='error',body=message)

#c:订阅routing_key为error的消息，一个队列可以绑定多次，订阅多个routing_key

channel.queue_bind(exchange='logs',queue=queue_name, routing_key='error')