RabbitMQ 详解及实例（含错误信息处理）

 Doker官网：Doker 多克

一、AMQP

AMQP协议是一个高级抽象层消息通信协议，RabbitMQ是AMQP协议的实现。它主要包括以下组件：

1.Server(broker): 接受客户端连接，实现AMQP消息队列和路由功能的进程。

2.Virtual Host:其实是一个虚拟概念，类似于权限控制组，一个Virtual Host里面可以有若干个Exchange和Queue，但是权限控制的最小粒度是Virtual Host

3.Exchange:接受生产者发送的消息，并根据Binding规则将消息路由给服务器中的队列。ExchangeType决定了Exchange路由消息的行为，例如，在RabbitMQ中，ExchangeType有direct、Fanout和Topic三种，不同类型的Exchange路由的行为是不一样的。

   3.1 直接式交换器类型(Direct)

Direct Exchange：直接交互式处理路由键。需要将一个队列绑定到交换机上，要求该消息与一个特定的路由键完全匹配，这是一个完整的匹配。路由键就是BindingKey。如果一个队列绑定到该交换机上要求路由键“dog”，则只有被标记为“dog”的消息才被转发，不会转发dog.puppy，也不会转发dog.guard，只会转发dog。

   3.2. 广播式交换机类型(Fanout)

Fanout Exchange：广播式路由键。你只需要简单的将队列绑定到交换机上。一个发送到交换机的消息都被转发到与该交换机绑定的所有队列上。很像子网广播，每台子网内的主机都获得了一份复制的消息。Fanout交换机转发消息是最快的。

  3. 3  主题式交换金类型(Topic)

Topic Exchange：主题式交换器。通过消息的路由关键字和绑定关键字的模式匹配，将消息路由到被绑定的队列中。这种路由器类型可以被用来支持经典的发布/订阅消息传输类型——使用主题名字空间作为消息寻址模式，将消息传递给那些部分或者全部匹配主题模式的多个消费者。主题交换器类型的工作方式如下：绑定关键字用零个或多个标记构成，每一个标记之间用“.”字符分隔。绑定关键字必须用这种形式明确说明，并支持通配符：“*”匹配一个词组，“#”零个或多个词组。因此绑定关键字“*.stock.#”匹配路由关键字“usd.stock”和“eur.stock.db”，但是不匹配“stock.nasdaq”。

4.Message Queue：消息队列，用于存储还未被消费者消费的消息。

5.Message: 由Header和Body组成，Header是由生产者添加的各种属性的集合，包括Message是否被持久化、由哪个Message Queue接受、优先级是多少等。而Body是真正需要传输的APP数据。

6.Binding:Binding联系了Exchange与Message Queue。Exchange在与多个Message Queue发生Binding后会生成一张路由表，路由表中存储着Message Queue所需消息的限制条件即Binding Key。当Exchange收到Message时会解析其Header得到Routing Key，Exchange根据Routing Key与Exchange Type将Message路由到Message Queue。Binding Key由Consumer在Binding Exchange与Message Queue时指定，而Routing Key由Producer发送Message时指定，两者的匹配方式由Exchange Type决定。 

7.Connection:连接，对于RabbitMQ而言，其实就是一个位于客户端和Broker之间的TCP连接。

8.Channel:信道，仅仅创建了客户端到Broker之间的连接后，客户端还是不能发送消息的。需要为每一个Connection创建Channel，AMQP协议规定只有通过Channel才能执行AMQP的命令。一个Connection可以包含多个Channel。之所以需要Channel，是因为TCP连接的建立和释放都是十分昂贵的，如果一个客户端每一个线程都需要与Broker交互，如果每一个线程都建立一个TCP连接，暂且不考虑TCP连接是否浪费，就算操作系统也无法承受每秒建立如此多的TCP连接。RabbitMQ建议客户端线程之间不要共用Channel，至少要保证共用Channel的线程发送消息必须是串行的，但是建议尽量共用Connection。

9.Command:AMQP的命令，客户端通过Command完成与AMQP服务器的交互来实现自身的逻辑。例如在RabbitMQ中，客户端可以通过publish命令发送消息，txSelect开启一个事务，txCommit提交一个事务。

在了解了AMQP模型以后，需要简单介绍一下AMQP的协议栈，AMQP协议本身包括三层：

1.Module Layer，位于协议最高层，主要定义了一些供客户端调用的命令，客户端可以利用这些命令实现自己的业务逻辑，例如，客户端可以通过queue.declare声明一个队列，利用consume命令获取一个队列中的消息。

2.Session Layer，主要负责将客户端的命令发送给服务器，在将服务器端的应答返回给客户端，主要为客户端与服务器之间通信提供可靠性、同步机制和错误处理。

3.Transport Layer，主要传输二进制数据流，提供帧的处理、信道复用、错误检测和数据表示。

二、RabbitMQ集群种类

RabbitMQ是用erlang开发的，集群非常方便，因为erlang天生就是一门分布式语言,但其本身并不支持负载均衡。 Rabbit模式大概分为以下三种：单一模式、普通模式、镜像模式。

1.单一模式：最简单的情况，非集群模式。

2.普通模式：默认的集群模式。

对于Queue来说，消息实体只存在于其中一个节点，A、B两个节点仅有相同的元数据，即队列结构。 当消息进入A节点的Queue中后，consumer从B节点拉取时，RabbitMQ会临时在A、B间进行消息传输，把A中的消息实体取出并经过B发送给consumer。 所以consumer应尽量连接每一个节点，从中取消息。即对于同一个逻辑队列，要在多个节点建立物理Queue。否则无论consumer连A或B，出口总在A，会产生瓶颈。 该模式存在一个问题就是当A节点故障后，B节点无法取到A节点中还未消费的消息实体。 如果做了消息持久化，那么得等A节点恢复，然后才可被消费；如果没有持久化的话，然后就没有然后了……

3.镜像模式：把需要的队列做成镜像队列，存在于多个节点，属于RabbitMQ的HA方案。

该模式解决了上述问题，其实质和普通模式不同之处在于，消息实体会主动在镜像节点间同步，而不是在consumer取数据时临时拉取。 该模式带来的副作用也很明显，除了降低系统性能外，如果镜像队列数量过多，加之大量的消息进入，集群内部的网络带宽将会被这种同步通讯大大消耗掉。 所以在对可靠性要求较高的场合中适用。

三、集群基本概念

RabbitMQ的集群节点包括内存节点、磁盘节点。内存节点就是将所有数据放在内存，磁盘节点将数据放在磁盘。不过，如果在投递消息时，打开了消息的持久化，那么即使是内存节点，数据还是安全的放在磁盘。

一个rabbitmq集群中可以共享 user，vhost，queue，exchange等，所有的数据和状态都是必须在所有节点上复制的，一个例外是，那些当前只属于创建它的节点的消息队列，尽管它们可见且可被所有节点读取。rabbitmq节点可以动态的加入到集群中，一个节点它可以加入到集群中，也可以从集群环集群会进行一个基本的负载均衡。 集群中有两种节点：

• 内存节点：只保存状态到内存（一个例外的情况是：持久的queue的持久内容将被保存到disk）
• 磁盘节点：保存状态到内存和磁盘。

内存节点虽然不写入磁盘，但是它执行比磁盘节点要好。集群中，只需要一个磁盘节点来保存状态就足够了。

如果集群中只有内存节点，那么不能停止它们，否则所有的状态，消息等都会丢失

四、高级特性

1.过期时间

Time To Live，也就是生存时间，是一条消息在队列中的最大存活时间，单位是毫秒，下面看看RabbitMQ过期时间特性：

• RabbitMQ可以对消息和队列设置TTL。
• RabbitMQ支持设置消息的过期时间，在消息发送的时候可以进行指定，每条消息的过期时间可以不同。
• RabbitMQ支持设置队列的过期时间，从消息入队列开始计算，直到超过了队列的超时时间配置，那么消息会变成死信，自动清除。
• 如果两种方式一起使用，则过期时间以两者中较小的那个数值为准。
• 当然也可以不设置TTL，不设置表示消息不会过期；如果设置为0，则表示除非此时可以直接将消息投递到消费者，否则该消息将被立即丢弃。

2. 消息确认

为了保证消息从队列可靠地到达消费者，RabbitMQ提供了消息确认机制。

消费者订阅队列的时候，可以指定autoAck参数，当autoAck为true的时候，RabbitMQ采用自动确认模式，RabbitMQ自动把发送出去的消息设置为确认，然后从内存或者硬盘中删除，而不管消费者是否真正消费到了这些消息。

当autoAck为false的时候，RabbitMQ会等待消费者回复的确认信号，收到确认信号之后才从内存或者磁盘中删除消息。

消息确认机制是RabbitMQ消息可靠性投递的基础，只要设置autoAck参数为false，消费者就有足够的时间处理消息，不用担心处理消息的过程中消费者进程挂掉后消息丢失的问题。

3.持久化

消息的可靠性是RabbitMQ的一大特色，那么RabbitMQ是如何保证消息可靠性的呢？答案就是消息持久化。持久化可以防止在异常情况下丢失数据。RabbitMQ的持久化分为三个部分：交换器持久化、队列持久化和消息的持久化。

交换器持久化可以通过在声明队列时将durable参数设置为true。如果交换器不设置持久化，那么在RabbitMQ服务重启之后，相关的交换器元数据会丢失，不过消息不会丢失，只是不能将消息发送到这个交换器了。

队列的持久化能保证其本身的元数据不会因异常情况而丢失，但是不能保证内部所存储的消息不会丢失。要确保消息不会丢失，需要将其设置为持久化。队列的持久化可以通过在声明队列时将durable参数设置为true。

设置了队列和消息的持久化，当RabbitMQ服务重启之后，消息依然存在。如果只设置队列持久化或者消息持久化，重启之后消息都会消失。

当然，也可以将所有的消息都设置为持久化，但是这样做会影响RabbitMQ的性能，因为磁盘的写入速度比内存的写入要慢得多。

对于可靠性不是那么高的消息可以不采用持久化处理以提高整体的吞吐量。鱼和熊掌不可兼得，关键在于选择和取舍。在实际中，需要根据实际情况在可靠性和吞吐量之间做一个权衡。

4.死信队列

当消息在一个队列中变成死信之后，他能被重新发送到另一个交换器中，这个交换器成为死信交换器，与该交换器绑定的队列称为死信队列。

消息变成死信有下面几种情况：

• 消息被拒绝。
• 消息过期
• 队列达到最大长度

DLX也是一个正常的交换器，和一般的交换器没有区别，他能在任何的队列上面被指定，实际上就是设置某个队列的属性。当这个队列中有死信的时候，RabbitMQ会自动将这个消息重新发送到设置的交换器上，进而被路由到另一个队列，我们可以监听这个队列中消息做相应的处理。

死信队列有什么用？当发生异常的时候，消息不能够被消费者正常消费，被加入到了死信队列中。后续的程序可以根据死信队列中的内容分析当时发生的异常，进而改善和优化系统。

5. 延迟队列

一般的队列，消息一旦进入队列就会被消费者立即消费。延迟队列就是进入该队列的消息会被消费者延迟消费，延迟队列中存储的对象是的延迟消息，“延迟消息”是指当消息被发送以后，等待特定的时间后，消费者才能拿到这个消息进行消费。

延迟队列用于需要延迟工作的场景。最常见的使用场景：淘宝或者天猫我们都使用过，用户在下单之后通常有30分钟的时间进行支付，如果这30分钟之内没有支付成功，那么订单就会自动取消。

除了延迟消费，延迟队列的典型应用场景还有延迟重试。比如消费者从队列里面消费消息失败了，可以延迟一段时间以后进行重试。

五、开发实测

 1、pom文件：

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

2、基本配置文件：

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/*rabbitMq基本配置*/
@Configuration
public class RabbitMqConfig {public final static String productExchange = "productExchange ";private String productQueue = "productQueue";public final static String ERROR_EXCHANGE = "le.error.exchange";private String topicExchangeName = "";@Beanpublic FanoutExchange productExchange() {return new FanoutExchange(productExchange);}public TopicExchange topicExchange() {return new TopicExchange(topicExchangeName);}@Beanpublic DirectExchange errorExchang() {return new DirectExchange(ERROR_EXCHANGE, true, false);}@Beanpublic Queue productQueue() {return new Queue(productQueue, true);}@Beanpublic Binding productBinding(Queue productQueue, FanoutExchange productExchange) {return BindingBuilder.bind(productQueue).to(productExchange);}
}

3、生产者:

@Component
public class ProductProducer {Logger logger = LoggerFactory.getLogger(ProductProducer.class);@AutowiredRabbitTemplate rabbitTemplate;@AutowiredGson gson;public void push(ProfitEntity profitEntity) {String json = gson.toJson("");logger.info("###消息发送者: json＝" + json); rabbitTemplate.convertAndSend(RabbitmqConfig.Product_EXCHANGE, null, json);}}

4、消费者：

@Component
public class TransProductConsumer {Logger logger = LoggerFactory.getLogger(TransproductConsumer.class);@AutowiredRabbitTemplate rabbitTemplate;@AutowiredErrorProducer errorProducer;@AutowiredProductService productService;@RabbitHandler@RabbitListener(queues = RabbitmqConfig.productQueue )public void comsume(String json) {try {logger.info("##start##消费者：TransProductConsumer ，接收到消息：json＝" + json);} catch (Exception e) {logger.error("计算消息队列出错 zf.Product.caculate.queue ", e);errorProducer.push(RabbitmqConfig.ERROR_PROFIT_CACULATE_QUEUE, json);}}
}

5、错误消息处理生产者：

@Component
public class ErrorProducer {Logger logger = LoggerFactory.getLogger(TransDetailProducer.class);@AutowiredRabbitTemplate rabbitTemplate;@AutowiredGson gson;public void push(String queueName, String errorJson) {logger.info("###异常消息发送者: queueName=" + queueName + " errorJson＝" + errorJson);rabbitTemplate.convertAndSend(RabbitmqConfig.ERROR_EXCHANGE, queueName, errorJson);}public void push(String queueName, Object obj) {String errorJson = gson.toJson(obj);logger.info("###异常消息发送者: queueName=" + queueName + " errorJson＝" + errorJson);rabbitTemplate.convertAndSend(RabbitmqConfig.ERROR_EXCHANGE, queueName, errorJson);}}

RabbitMQ 官网：Messaging that just works — RabbitMQ

大家好，我是Doker品牌的Sinbad,欢迎点赞和评论,您的鼓励是我们持续更新的动力！欢迎加微信进入技术群聊！