本文共 6230 字，大约阅读时间需要 20 分钟。

简介

RabbitMQ 是一个由 Erlang 语言开发的 AMQP 的开源实现。

rabbitMQ是一款基于AMQP协议的消息中间件，它能够在应用之间提供可靠的消息传输。在易用性，扩展性，高可用性上表现优秀。使用消息中间件利于应用之间的解耦，生产者（客户端）无需知道消费者（服务端）的存在。而且两端可以使用不同的语言编写，大大提供了灵活性。 应用场景：异步，消峰，解耦。

AMQP协议简介

AMQP，即Advanced Message Queuing Protocol,一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准,为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件同产品，不同的开发语言等条件的限制。

Queue

Rabbit工作模式

1.简单模式

### 生产者import pikaconnection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))channel = connection.channel()channel.queue_declare(queue='hello')channel.basic_publish(exchange='',                      routing_key='hello',                      body='Hello World!')print(" [x] Sent 'Hello World!'")### 消费者import pikaconnection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))channel = connection.channel()channel.queue_declare(queue='hello')def callback(ch, method, properties, body):    print(" [x] Received %r" % body)channel.basic_consume(queue='hello',                      auto_ack=True,                      on_message_callback=callback)print(' [*] Waiting for messages. To exit press CTRL+C')channel.start_consuming()

参数

应答参数

auto_ack=Falsech.basic_ack(delivery_tag=method.delivery_tag)

持久化参数

#声明queuechannel.queue_declare(queue='hello2', durable=True)  # 若声明过，则换一个名字 channel.basic_publish(exchange='',                      routing_key='hello2',                      body='Hello World!',                      properties=pika.BasicProperties(                          delivery_mode=2,  # make message persistent                          )                      )

分发参数

有两个消费者同时监听一个的队列。其中一个线程sleep2秒，另一个消费者线程sleep1秒，但是处理的消息是一样多。这种方式叫轮询分发（round-robin）不管谁忙，都不会多给消息，总是你一个我一个。想要做到公平分发（fair dispatch），必须关闭自动应答ack，改成手动应答。使用basicQos(perfetch=1)限制每次只发送不超过1条消息到同一个消费者，消费者必须手动反馈告知队列，才会发送下一个。

channel.basic_qos(prefetch_count=1)

交换机模式

发布订阅

发布订阅和简单的消息队列区别在于，发布订阅会将消息发送给所有的订阅者，而消息队列中的数据被消费一次便消失。所以，RabbitMQ实现发布和订阅时，会为每一个订阅者创建一个队列，而发布者发布消息时，会将消息放置在所有相关队列中。

#生产者import pikaconnection = pika.BlockingConnection(pika.ConnectionParameters(        host='localhost'))channel = connection.channel()channel.exchange_declare(exchange='logs',                         exchange_type='fanout')message = "info: Hello World!"channel.basic_publish(exchange='logs',                      routing_key='',                      body=message)print(" [x] Sent %r" % message)connection.close() #消费者import pikaconnection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))channel = connection.channel()channel.exchange_declare(exchange='logs',                         exchange_type='fanout')result = channel.queue_declare("",exclusive=True)queue_name = result.method.queuechannel.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(queue=queue_name,                      auto_ack=True,                      on_message_callback=callback)channel.start_consuming()

关键字

# 生产者import pikaconnection = pika.BlockingConnection(pika.ConnectionParameters(        host='localhost'))channel = connection.channel()channel.exchange_declare(exchange='logs2',                         exchange_type='direct')message = "info: Hello Yuan!"channel.basic_publish(exchange='logs2',                      routing_key='info',                      body=message)print(" [x] Sent %r" % message)connection.close()# 消费者import pikaimport sysconnection = pika.BlockingConnection(pika.ConnectionParameters(        host='localhost'))channel = connection.channel()channel.exchange_declare(exchange='logs2',                         exchange_type='direct')result = channel.queue_declare("",exclusive=True)queue_name = result.method.queueseverities = sys.argv[1:]if not severities:    sys.stderr.write("Usage: %s [info] [warning] [error]\n" % sys.argv[0])    sys.exit(1)for severity in severities:    channel.queue_bind(exchange='logs2',                       queue=queue_name,                       routing_key=severity)print(' [*] Waiting for logs. To exit press CTRL+C')def callback(ch, method, properties, body):    print(" [x] %r" % body)channel.basic_consume(queue=queue_name,                      auto_ack=True,                      on_message_callback=callback)channel.start_consuming()

交换机之通配符

通配符交换机”与之前的路由模式相比，它将信息的传输类型的key更加细化，以“key1.key2.keyN…”的模式来指定信息传输的key的大类型和大类型下面的小类型，让消费者可以更加精细的确认自己想要获取的信息类型。而在消费者一段，不用精确的指定具体到哪一个大类型下的小类型的key，而是可以使用类似正则表达式(但与正则表达式规则完全不同)的通配符在指定一定范围或符合某一个字符串匹配规则的key，来获取想要的信息。

“通配符交换机”（Topic Exchange）将路由键和某模式进行匹配。此时队列需要绑定在一个模式上。符号“#”匹配一个或多个词，符号“”仅匹配一个词。因此“audit.#”能够匹配到“audit.irs.corporate”，但是“audit.”只会匹配到“audit.irs”。（这里与我们一般的正则表达式的“*”和“#”刚好相反，这里我们需要注意一下。）

下面是一个解释通配符模式交换机工作的一个样例

# 生产者import pikaconnection = pika.BlockingConnection(pika.ConnectionParameters(        host='localhost'))channel = connection.channel()channel.exchange_declare(exchange='logs3',                         exchange_type='topic')message = "info: Hello ERU!"channel.basic_publish(exchange='logs3',                      routing_key='europe.weather',                      body=message)print(" [x] Sent %r" % message)connection.close()# 消费者import pikaimport sysconnection = pika.BlockingConnection(pika.ConnectionParameters(        host='localhost'))channel = connection.channel()channel.exchange_declare(exchange='logs3',                         exchange_type='topic')result = channel.queue_declare("",exclusive=True)queue_name = result.method.queuechannel.queue_bind(exchange='logs3',                   queue=queue_name,                   routing_key="#.news")print(' [*] Waiting for logs. To exit press CTRL+C')def callback(ch, method, properties, body):    print(" [x] %r" % body)channel.basic_consume(queue=queue_name,                      auto_ack=True,                      on_message_callback=callback)channel.start_consuming()

几种MQ对比

RocketMQ：

开源分布式消息中间件，不遵循JMS规范，吞吐能力强，最新发版4.7，社区活跃，技术比较成熟；（吞吐量十万级以上）

Kafka:

分布式消息中间件，不遵循JMS规范，吞吐能力强，更倾向于数据流处理，主要应用于大数据流处理和日志处理，为处理日志而生；（吞吐量十万级以上） （专门做数据流处理的服务，严格意义上不算消息中间件） kafka topic随着增多 吞吐能力急剧下降，kafka topic可以很大 但不可以很多

ActiveMQ:

分布式消息中间件，高性能，开源，吞吐能力一般（万级）遵循JMS规范，技术成熟 支持很多种协议，资料多； （吞吐量万级）

RabbitMQ:

Erlang语言开发，遵循JMS规范，高性能，开源，吞吐能力一般（万级），用在分布式系统中存储转发消息、具有高可用；（吞吐量万级，处理数据的速度是微秒级别，其他MQ都是毫秒级））