关于rabbitmq的介绍

 原文转载：http://blog.csdn.net/whycold/article/details/41119807

保护原帖，尊重技术，致敬工匠！

1、简介

MQ全称为Message Queue, 消息队列（MQ）是一种应用程序对应用程序的通讯方法。应用程序经过读写出入队列的消息（针对应用程序的数据）来通讯，而无需专用链接来连接它们。消息传递指的是程序之间经过在消息中发送数据进行通讯，而不是经过直接调用彼此来通讯，直接调用一般是用于诸如远程过程调用的技术。排队指的是应用程序经过 队列来通讯。队列的使用除去了接收和发送应用程序同时执行的要求。其中较为成熟的MQ产品有IBM WEBSPHERE MQ等等...

2、使用场景

在项目中，将一些无需即时返回且耗时的操做提取出来，进行了异步处理，而这种异步处理的方式大大的节省了服务器的请求响应时间，从而提升了系统的吞吐量。

3、相关名称介绍

一、ConnectionFactory、Connection、Channel

ConnectionFactory、Connection、Channel都是RabbitMQ对外提供的API中最基本的对象。

Connection是RabbitMQ的socket连接，它封装了socket协议相关部分逻辑。

ConnectionFactory为Connection的制造工厂。

Channel是咱们与RabbitMQ打交道的最重要的一个接口，咱们大部分的业务操做是在Channel这个接口中完成的，包括定义Queue、定义Exchange、绑定Queue与Exchange、发布消息等。

二、Queue

Queue（队列）是RabbitMQ的内部对象，用于存储消息，用下图表示。

RabbitMQ中的消息都只能存储在Queue中，生产者（下图中的P）生产消息并最终投递到Queue中，消费者（下图中的C）能够从Queue中获取消息并消费。

多个消费者能够订阅同一个Queue，这时Queue中的消息会被平均分摊给多个消费者进行处理，而不是每一个消费者都收到全部的消息并处理。

三、Message acknowledgment

在实际应用中，可能会发生消费者收到Queue中的消息，但没有处理完成就宕机（或出现其余意外）的状况，这种状况下就可能会致使消息丢失。为了不这种状况发生，咱们能够要求消费者在消费完消息后发送一个回执给RabbitMQ，RabbitMQ收到消息回执（Message acknowledgment）后才将该消息从Queue中移除；若是RabbitMQ没有收到回执并检测到消费者的RabbitMQ链接断开，则RabbitMQ会将该消息发送给其余消费者（若是存在多个消费者）进行处理。这里不存在timeout概念，一个消费者处理消息时间再长也不会致使该消息被发送给其余消费者，除非它的RabbitMQ链接断开。

这里会产生另一个问题，若是咱们的开发人员在处理完业务逻辑后，忘记发送回执给RabbitMQ，这将会致使严重的bug——Queue中堆积的消息会愈来愈多；消费者重启后会重复消费这些消息并重复执行业务逻辑…

另外pub message是没有ack的。

四、Message durability

若是咱们但愿即便在RabbitMQ服务重启的状况下，也不会丢失消息，咱们能够将Queue与Message都设置为可持久化的（durable），这样能够保证绝大部分状况下咱们的RabbitMQ消息不会丢失。但依然解决不了小几率丢失事件的发生（好比RabbitMQ服务器已经接收到生产者的消息，但还没来得及持久化该消息时RabbitMQ服务器就断电了），若是咱们须要对这种小几率事件也要管理起来，那么咱们要用到事务。因为这里仅为RabbitMQ的简单介绍，因此这里将不讲解RabbitMQ相关的事务。

五、Prefetch count

前面咱们讲到若是有多个消费者同时订阅同一个Queue中的消息，Queue中的消息会被平摊给多个消费者。这时若是每一个消息的处理时间不一样，就有可能会致使某些消费者一直在忙，而另一些消费者很快就处理完手头工做并一直空闲的状况。咱们能够经过设置prefetchCount来限制Queue每次发送给每一个消费者的消息数，好比咱们设置prefetchCount=1，则Queue每次给每一个消费者发送一条消息；消费者处理完这条消息后Queue会再给该消费者发送一条消息。

六、Exchange

在上一节咱们看到生产者将消息投递到Queue中，实际上这在RabbitMQ中这种事情永远都不会发生。实际的状况是，生产者将消息发送到Exchange（交换器，下图中的X），由Exchange将消息路由到一个或多个Queue中（或者丢弃）。

Exchange是按照什么逻辑将消息路由到Queue的？这个将在下面的八、Binding中介绍。

RabbitMQ中的Exchange有四种类型，不一样的类型有着不一样的路由策略，这将在下面的十、Exchange Types中介绍。

七、routing key

生产者在将消息发送给Exchange的时候，通常会指定一个routing key，来指定这个消息的路由规则，而这个routing key须要与Exchange Type及binding key联合使用才能最终生效。

在Exchange Type与binding key固定的状况下（在正常使用时通常这些内容都是固定配置好的），咱们的生产者就能够在发送消息给Exchange时，经过指定routing key来决定消息流向哪里。RabbitMQ为routing key设定的长度限制为255 bytes。

八、Binding

RabbitMQ中经过Binding将Exchange与Queue关联起来，这样RabbitMQ就知道如何正确地将消息路由到指定的Queue了。

九、Binding key

在绑定（Binding）Exchange与Queue的同时，通常会指定一个binding key；消费者将消息发送给Exchange时，通常会指定一个routing key；当binding key与routing key相匹配时，消息将会被路由到对应的Queue中。这个将在Exchange Types章节会列举实际的例子加以说明。

在绑定多个Queue到同一个Exchange的时候，这些Binding容许使用相同的binding key。
binding key 并非在全部状况下都生效，它依赖于Exchange Type，好比fanout类型的Exchange就会无视binding key，而是将消息路由到全部绑定到该Exchange的Queue。

十、Exchange Types

RabbitMQ经常使用的Exchange Type有fanout、direct、topic、headers这四种（AMQP规范里还提到两种Exchange Type，分别为system与自定义，这里不予以描述），下面分别进行介绍。

fanout

fanout类型的Exchange路由规则很是简单，它会把全部发送到该Exchange的消息路由到全部与它绑定的Queue中。

上图中，生产者（P）发送到Exchange（X）的全部消息都会路由到图中的两个Queue，并最终被两个消费者（C1与C2）消费。

direct

direct类型的Exchange路由规则也很简单，它会把消息路由到那些binding key与routing key彻底匹配的Queue中。

以上图的配置为例，咱们以routingKey=”error”发送消息到Exchange，则消息会路由到Queue1（amqp.gen-S9b…，这是由RabbitMQ自动生成的Queue名称）和Queue2（amqp.gen-Agl…）；若是咱们以routingKey=”info”或routingKey=”warning”来发送消息，则消息只会路由到Queue2。若是咱们以其余routingKey发送消息，则消息不会路由到这两个Queue中。

topic

前面讲到direct类型的Exchange路由规则是彻底匹配binding key与routing key，但这种严格的匹配方式在不少状况下不能知足实际业务需求。topic类型的Exchange在匹配规则上进行了扩展，它与direct类型的Exchage类似，也是将消息路由到binding key与routing key相匹配的Queue中，但这里的匹配规则有些不一样，它约定：

routing key为一个句点号“. ”分隔的字符串（咱们将被句点号“. ”分隔开的每一段独立的字符串称为一个单词），如“stock.usd.nyse”、“nyse.vmw”、“quick.orange.rabbit”

binding key与routing key同样也是句点号“. ”分隔的字符串。

binding key中能够存在两种特殊字符“*”与“#”，用于作模糊匹配，其中“*”用于匹配一个单词，“#”用于匹配多个单词（能够是零个）。

以上图中的配置为例，routingKey=”quick.orange.rabbit”的消息会同时路由到Q1与Q2，routingKey=”lazy.orange.fox”的消息会路由到Q1与Q2，routingKey=”lazy.brown.fox”的消息会路由到Q2，routingKey=”lazy.pink.rabbit”的消息会路由到Q2（只会投递给Q2一次，虽然这个routingKey与Q2的两个bindingKey都匹配）；routingKey=”quick.brown.fox”、routingKey=”orange”、routingKey=”quick.orange.male.rabbit”的消息将会被丢弃，由于它们没有匹配任何bindingKey。

headers

headers类型的Exchange不依赖于routing key与binding key的匹配规则来路由消息，而是根据发送的消息内容中的headers属性进行匹配。

在绑定Queue与Exchange时指定一组键值对；当消息发送到Exchange时，RabbitMQ会取到该消息的headers（也是一个键值对的形式），对比其中的键值对是否彻底匹配Queue与Exchange绑定时指定的键值对；若是彻底匹配则消息会路由到该Queue，不然不会路由到该Queue。

该类型的Exchange没有用到过（不过也应该颇有用武之地），因此不作介绍。

十一、RPC

MQ自己是基于异步的消息处理，前面的示例中全部的生产者（P）将消息发送到RabbitMQ后不会知道消费者（C）处理成功或者失败（甚至连有没有消费者来处理这条消息都不知道）。

但实际的应用场景中，咱们极可能须要一些同步处理，须要同步等待服务端将个人消息处理完成后再进行下一步处理。这至关于RPC（Remote Procedure Call，远程过程调用）。在RabbitMQ中也支持RPC。

RabbitMQ中实现RPC的机制是：

客户端发送请求（消息）时，在消息的属性（MessageProperties，在AMQP协议中定义了14中properties，这些属性会随着消息一块儿发送）中设置两个值replyTo（一个Queue名称，用于告诉服务器处理完成后将通知个人消息发送到这个Queue中）和correlationId（这次请求的标识号，服务器处理完成后须要将此属性返还，客户端将根据这个id了解哪条请求被成功执行了或执行失败）；

服务器端收到消息并处理；

服务器端处理完消息后，将生成一条应答消息到replyTo指定的Queue，同时带上correlationId属性；

客户端以前已订阅replyTo指定的Queue，从中收到服务器的应答消息后，根据其中的correlationId属性分析哪条请求被执行了，根据执行结果进行后续业务处理。

4、总结

本文介绍了RabbitMQ中我的认为最重要的概念，充分利用RabbitMQ提供的这些功能就能够处理咱们绝大部分的异步业务