RabbitMQ深入学习

继承上一节的学习，上一节学习了RabbitMQ的根本内容，本节学习RabbitMQ的高级特性。
RocketMQ的高级特性学习见这篇博客



  

1.消息可靠性

消息从发送，到消费者接收，会司理多个过程：

此中的每一步都可能导致消息丢失，常见的丢失原因包罗：

• 发送时丢失：
  
  
  
  • 生产者发送的消息未送达exchange
    
  • 消息到达exchange后未到达queue
    
  
  
• MQ宕机，queue将消息丢失
  
• consumer接收到消息后未消费就宕机
  

针对这些题目，RabbitMQ分别给出了办理方案：

• 生产者确认机制
  
• mq长期化
  
• 消费者确认机制
  
• 失败重试机制
  

下面我们就通过案例来演示每一个步骤。

项目结构如下：

1.1生产者消息确认

RabbitMQ提供了publisher confirm机制来制止消息发送到MQ过程中丢失。这种机制必须给每个消息指定一个唯一ID。消息发送到MQ以后，会返回一个结果给发送者，表示消息是否处理成功。
返回结果有两种方式：

• publisher-confirm，发送者确认
  
  
  
  • 消息成功投递到交换机，返回ack
    
  • 消息未投递到交换机，返回nack
    
  
  
• publisher-return，发送者回执
  
  
  
  • 消息投递到交换机了，但是没有路由到队列。返回ACK，及路由失败原因。
    
  
  

如下图所示，只有成功投递到交换机而且路由到队列才是投递成功：

   注意：确认机制发送消息时，需要给每个消息设置一个全局唯一id，以区分不同消息，制止ack辩说
  下面我们进行演示。mq的安装和配置信息，以及yml中的根本配置信息好比ip端口等，前面文章已经叙述不再赘述。
1.修改配置
起首，修改publisher服务中的application.yml文件，添加下面的内容：

1. spring:
2.   rabbitmq:
3.     publisher-confirm-type: correlated
4.     publisher-returns: true
5.     template:
6.       mandatory: true

复制代码

  

• publish-confirm-type：开启publisher-confirm，在AMQP这里支持两种生产者确认范例：
  
  
  
  • simple：同步期待confirm结果，直到超时。不推荐
    
  • correlated：异步回调，定义ConfirmCallback，MQ返回结果时会回调这个ConfirmCallback。推荐
    
  
  
• publish-returns：开启publish-return功能，同样是基于callback机制，这里是定义ReturnCallback而非上面的ConfirmCallback，但是返回结果返不返回取决于template.mandatory的配置。
  
• template.mandatory：定义消息路由到队列失败时的计谋。true，则调用ReturnCallback；false：则直接抛弃消息
  

  
2.定义Return回调
开启生产者消息确认功能之后，既然说了要通过回调机制来实现，那就得编写这个回调函数，怎么编写呢？起首是ReturnCallback的编写。
每个RabbitTemplate(由spring创建的，所以是单例)只能配置一个ReturnCallback，所以不能在每次发消息时来陪，需要在项目加载时配置：
修改publisher服务，添加一个：

1. package cn.itcast.mq.config;
3. import lombok.extern.slf4j.Slf4j;
4. import org.springframework.amqp.rabbit.core.RabbitTemplate;
5. import org.springframework.beans.BeansException;
6. import org.springframework.context.ApplicationContext;
7. import org.springframework.context.ApplicationContextAware;
8. import org.springframework.context.annotation.Configuration;
10. @Slf4j
11. @Configuration
12. public class CommonConfig implements ApplicationContextAware {
13.     @Override
14.     public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
15.         // 获取RabbitTemplate对象
16.         RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);
17.         // 配置ReturnCallback
18.         rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
19.             // 投递失败，记录日志
20.             log.info("消息发送失败，应答码{}，原因{}，交换机{}，路由键{},消息{}",
21.                      replyCode, replyText, exchange, routingKey, message.toString());
22.             // 如果有业务需要，可以重发消息
23.         });
24.     }
25. }

复制代码

  ApplicationContextAware是bean容器大概说bean工厂的关照接口，意思就是当spring的bean工厂预备好了以后它会来关照你，而且你实现了这个接口，你一定要实现上面的方法叫setApplicationContext。那么他在关照你的时间就会把spring这个容器ApplicationContext 传递给你。那你想呀spring的bean工厂创建完了关照我，是不是拿到这个工厂了，我是不是就可以从工厂里取到我想要的bean了。
这里就我就取到了这个RabbitTemplate，就可以给他设置callback，我们讲的是这个代码是在并工厂创建完了以后运行，即在项目启动时就会去实行了，就是个全局callback了。


这里的回调逻辑是什么呢？lambda有五个参数，你如今不是消息路由失败了吗，好我会给你一个回执，在回执中我会告诉你诶，你发的消息message是什么，失败的状态码replyCode是什么，失败的原因replyText是什么，消息投递到了哪个交换机exchange，然后你投递时routingKey用的是什么。
那这五个参数给我可以干什么呢？第一，我可以去记载日志记载一下；第二，我既然拿到了这个失败的消息体，失败的交换机，失败的 routingKey，也当然可以重发消息了，即失败的重试。
  以上就是消息到达了交换机但是路由到队列失败时的ReturnCallback，别的呢我们还有一个ConfirmCallback，定义消息根本就没有到达交换机的回调计谋

3.定义ConfirmCallback
ConfirmCallback可以在发送消息时指定，因为每个业务处理confirm成功或失败的逻辑不一定雷同。
在publisher服务的cn.itcast.mq.spring.SpringAmqpTest类中，定义一个单元测试方法：

1. public void testSendMessage2SimpleQueue() throws InterruptedException {
2.     // 1.准备好一条消息消息体，在发送过程中去添加回调逻辑
3.     String message = "hello, spring amqp!";
4.     // 1.2全局唯一的消息ID(使用UUID)，需要封装到CorrelationData中
5.     CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
6.     // 1.3添加callback，就是我们的ConfirmCallback
7.     correlationData.getFuture().addCallback(
8.         result -> {
9.                 // 接收到了mq的回执
10.             if(result.isAck()){
11.                 // 1.3.1.ack，消息成功投递到交换机
12.                 log.debug("消息发送成功, ID:{}", correlationData.getId());
13.             }else{
14.                 // 1.3.2.nack，消息没能投递到交换机
15.                 log.error("消息发送失败, ID:{}, 原因{}",correlationData.getId(), result.getReason());
16.             }
17.         },
18.         // 1.4发送失败连回调都收不到
19.         ex -> log.error("消息发送异常, ID:{}, 原因{}",correlationData.getId(),ex.getMessage())
20.     );
21.     // 2.发送消息，三个参数：交换机名称；routingKey名称；消息体
22.     // 之前就传前三，这里多了第四个参数：correlationData，对应我们在yml里配置的correlated
23.     // correlationData封装了消息唯一id和回调逻辑
24.     rabbitTemplate.convertAndSend("task.direct", "task", message, correlationData);
26.     // 休眠一会儿，等待ack回执
27.     Thread.sleep(2000);
28. }

复制代码

  发送之前得将交换机跟队列绑定好，，否则即使消息到了交换机，也会因为找不到队列而触发回退机制ReturnCallback。
方式一：通过代码方式进行绑定（推荐），前面文章有介绍。
方式二：手动在 RabbitMQ 管理界面中绑定。这种方式适用于测试或临时使用，但不推荐用于生产，因为无法版本控制和主动摆设。
  测试发现提示投递成功，修改投递的交换机名称模拟写错，模拟投递失败，测试ConfirmCallback是否正常触发；修改投递的routingKey模拟写错，模拟路由失败，测试ReturnCallback是否正常触发。


1.2消息长期化

生产者确认可以确保消息投递到RabbitMQ的队列中，但是消息发送到RabbitMQ以后，假如忽然宕机，也可能导致消息丢失。要想确保消息在RabbitMQ中安全生存，必须开启消息长期化机制。
我们先去看一下征象，这是我们之前预备的一条队列，然后在这个队列内里已经有一条消息了

交换机界面如下，这些交换机啊是我们系统主动创建的交换机：

然后我们通过过docker restart mq完成重启mq，重启完成发现交换机都在，但是队列没了，队列都没了何谈队列当中的消息呢。那为什么我们自己创建的队列消息都没了但系统的交换机还在呢。这说明系统的交换机是长期化的。
我们观察可知其实系统创建的交换机都带了一个特征(Features)——D，对应长期化。那因此呢我们要想让我们的队列也能长期化，让我们的交换机也能长期化，都应该带上这个durable的参数，在前端我们创建一个交换机的时间Durability参数设置为Durable即可，创建队列时同理。控制代码去创建我们这节将会学习。
消息长期化分为：

• 交换机长期化
  
• 队列长期化
  
• 消息长期化
  

1.交换机长期化
RabbitMQ中交换机默认黑白长期化的，mq重启后就丢失。SpringAMQP中声明交换机时可以通过代码指定交换机长期化：

1. @Bean
2. public DirectExchange simpleExchange(){
3.     // 三个参数：交换机名称、是否持久化、当没有queue与其绑定时是否自动删除
4.     return new DirectExchange("simple.direct", true, false);
5. }

复制代码

  究竟上，默认情况下，由SpringAMQP声明的交换机都是长期化的。
可以在RabbitMQ控制台看到长期化的交换机都会带上D的标示：
  

2.队列长期化
RabbitMQ中队列默认黑白长期化的，mq重启后就丢失。
SpringAMQP中声明队列时可以通过代码指定交换机长期化：

1. @Bean
2. public Queue simpleQueue(){
3.     // 使用QueueBuilder构建队列，durable就是持久化的
4.     return QueueBuilder.durable("simple.queue").build();
5. }

复制代码

  究竟上，默认情况下，由SpringAMQP声明的队列都是长期化的。
可以在RabbitMQ控制台看到长期化的队列都会带上D的标示：
  

3.消息长期化
队列长期化不代表消息就长期化，当交换机和队列长期化时假如宕机，消息仍然会丢失。
利用SpringAMQP发送消息时，可以设置消息的属性（MessageProperties），指定delivery-mode：

• 1：非长期化
  
• 2：长期化
  

用java代码指定：

1. // @Test
2. public void testDurableMessage() {
3.     // 创建消息
4.     Message message = MessageBuilder
5.         .withBody("hello, ttl queue".getBytes(StandardCharsets.UTF_8))
6.         .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
7.         .build();
8.     // 消息ID，需要封装到CorrelationData中
9.     CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
10.     // 发送消息
11.     rabbitTemplate.convertAndSend(routingKey: "simple.queue", message, correlationData);
12.     // 记录日志
13.     log.debug("发送消息成功");
14. }

复制代码

  我们之前的文章中发消息没使用MessageBuilder那么麻烦。默认情况下，SpringAMQP发出的任何消息都是长期化的，不消特意指定。
  

1.3消费者消息确认

RabbitMQ是阅后即焚机制，RabbitMQ确认消息被消费者消费并ACK 后会立刻删除。
RabbitMQ是通过消费者回执来确认消费者是否成功处理消息的：消费者获取消息后，应该向RabbitMQ发送ACK回执，表明自己已经处理消息。
假想如许的场景：

• 1）RabbitMQ投递消息给消费者
  
• 2）消费者获取消息后，返回ACK给RabbitMQ
  
• 3）RabbitMQ删除消息
  
• 4）消费者宕机，消息尚未处理
  

如许，消息就丢失了。因此消费者返回ACK的机遇非常紧张。而SpringAMQP则允许配置三种确认模式：

• manual：手动ack，需要在业务代码竣事后，调用api发送ack。
  
  
  
  • 即自己根据业务情况，判定什么时间该ack。一般情况下先写处理消息的业务，处理完了最后一行发送ack，假如业务实行过程中抛了异常则在catch里发nack，这种模式有代码侵入，不推荐。
    
  
  
• auto：主动ack，由spring监测listener代码(即消费者的业务逻辑)是否出现异常，没有异常则返回ack；抛出异常则返回nack
  
  
  
  • auto模式雷同事务机制，出现异常时返回nack，消息回滚到mq；没有异常，返回ack。不需要我们自己try，底层用了spring的aop技能。推荐使用。
    
  
  
• none：关闭ack，MQ假定消费者获取消息后会成功处理，因此消息投递后立即被删除
  
  
  
  • none模式下，消息投递是不可靠的，可能丢失
    
  
  

   一般，我们都是使用默认的auto即可。
  1.演示none模式
修改consumer服务的application.yml文件，添加下面内容：

1. spring:
2.   rabbitmq:
3.     listener:
4.       simple:
5.         acknowledge-mode: none # none,关闭ack;manual,手动ack;auto,自动ack

复制代码

修改consumer服务的SpringRabbitListener类中的方法，模拟一个消息处理异常：

1. @RabbitListener(queues = "simple.queue")
2. public void listenSimpleQueue(String msg) {
3.     log.info("消费者接收到simple.queue的消息：【{}】", msg);
4.     // 模拟异常
5.     System.out.println(1 / 0);
6.     log.debug("消息处理完成！");
7. }

复制代码

  测试可以发现，当消息处理抛异常时，消息依然被RabbitMQ删除了。
  
2.演示auto模式
再次把确认机制修改为auto:

1. spring:
2.   rabbitmq:
3.     listener:
4.       simple:
5.         acknowledge-mode: auto # 关闭ack

复制代码

在异常位置打断点，再次发送消息，程序卡在断点时，可以发现此时消息状态为unack（未确定状态）：

抛出异常后，因为Spring会主动返回nack，所以消息规复至Ready状态，而且没有被RabbitMQ删除：

1.4消费失败重试机制

当消费者出现异常后，消息会不断requeue（重入队）到队列，再重新发送给消费者，然后再次异常，再次requeue，无穷循环，导致mq的消息处理飙升，带来不必要的压力，怎么办呢？

1.本地重试
我们可以利用Spring的retry机制，在消费者出现异常时利用本地重试，而不是无穷定的requeue到mq队列。
   从前是抛了异常就把消息重新投递给mq，mq再把消息给消费者，消费者再处理，消费者再抛异常返回nack…来归去做)。如今是利用Spring自己的retry机制，消费者抛了异常后不再返回nack，而是在本地去做重试，直到成功大概达到重试上限，达到上限时可以采用其他计谋，好比说再把消息投递给mq大概是把消息放到一个额外的地方将来由人工参与。
  修改consumer服务的application.yml文件，添加内容：

1. spring:
2.   rabbitmq:
3.     listener:
4.       simple:
5.         retry:
6.           enabled: true # 开启消费者失败重试
7.           initial-interval: 1000 # 初始的失败等待时长为1秒
8.           multiplier: 1 # 失败的等待时长倍数，下次等待时长 = multiplier * 上一次等待时长，这里等待时长永远是1s；若设置为2就是1 2 4 8s...
9.           max-attempts: 3 # 最大重试次数
10.           stateless: true # 代表有无状态，默认为true无状态；false为有状态。一般情况下用不到，如果业务中包含事务，一定要改为false，即有状态。因为这个时候spring需要在重试的时候保留事务，不会去导致事务失效，对业务性能产生影响

复制代码

重启consumer服务，重复之前的测试。可以发现：

• 在重试3次后，SpringAMQP会抛出异常AmqpRejectAndDontRequeueException，说明本地重试触发了
  
• 查看RabbitMQ控制台，发现消息被删除了，说明最后SpringAMQP返回的是ack，mq删除消息了
  

结论：

• 开启本地重试时，消息处理过程中抛出异常，不会requeue到队列，而是在消费者本地重试
  
• 重试达到最大次数后，Spring会返回ack，消息会被抛弃
  

2.失败计谋
在之前的测试中，达到最大重试次数后，消息会被抛弃，这是由Spring内部机制决定的。
在开启重试模式后，重试次数耗尽，假如消息依然失败，则需要有MessageRecovery接口来处理，它包含三种不同的实现：

• RejectAndDontRequeueRecoverer：重试耗尽后，直接reject，抛弃消息。默认就是这种方式
  
• ImmediateRequeueMessageRecoverer：重试耗尽后，返回nack，消息重新入队
  
• RepublishMessageRecoverer：重试耗尽后，将失败消息投递到指定的交换机
  

比较优雅的一种处理方案是RepublishMessageRecoverer，失败后将消息投递到一个指定的，专门存放异常消息的队列，后续由人工会合处理。下面演示
1）在consumer服务中定义处理失败消息的交换机和队列

1. @Bean
2. public DirectExchange errorMessageExchange(){
3.     return new DirectExchange("error.direct");
4. }
5. @Bean
6. public Queue errorQueue(){
7.     return new Queue("error.queue", true);
8. }
9. @Bean
10. public Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange){
11.     return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with("error");
12. }

复制代码

2）定义一个RepublishMessageRecoverer，关联队列和交换机

1. @Bean
2. public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){
3.     return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");
4. }

复制代码

完备代码：

1. package cn.itcast.mq.config;
3. import org.springframework.amqp.core.Binding;
4. import org.springframework.amqp.core.BindingBuilder;
5. import org.springframework.amqp.core.DirectExchange;
6. import org.springframework.amqp.core.Queue;
7. import org.springframework.amqp.rabbit.core.RabbitTemplate;
8. import org.springframework.amqp.rabbit.retry.MessageRecoverer;
9. import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer;
10. import org.springframework.context.annotation.Bean;
12. @Configuration
13. public class ErrorMessageConfig {
14.     @Bean
15.     public DirectExchange errorMessageExchange(){
16.         return new DirectExchange("error.direct");
17.     }
18.     @Bean
19.     public Queue errorQueue(){
20.         return new Queue("error.queue", true);
21.     }
22.     @Bean
23.     public Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange){
24.         return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with("error");
25.     }
27.     @Bean
28.     public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){
29.         return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");
30.     }
31. }

复制代码

1.5消息可靠性保证总结

如何确保RabbitMQ消息的可靠性？

• 开启生产者确认机制，确保生产者的消息能到达队列
  
• 开启长期化功能，确保消息未消费前在队列中不会丢失
  
• 开启消费者确认机制为auto或maual，由spring确认消息处理成功后完成ack
  
• 开启消费者失败重试机制，并设置MessageRecoverer，多次重试失败后将消息投递到异常交换机，交由人工处理
  

2.什么是死信交换机

什么是死信？当一个队列中的消息满足下列情况之一时，可以成为死信（dead letter）：

• 消费者使用basic.reject或 basic.nack声明消费失败，而且消息的requeue参数设置为false
  
• 消息是一个过期消息，超时无人消费
  
• 要投递的队列消息满了，无法投递
  

假如这个包含死信的队列配置了dead-letter-exchange属性，指定了一个交换机，那么队列中的死信就会投递到这个交换机中，而这个交换机称为死信交换机（Dead Letter Exchange，查抄DLX）。
如图，一个消息被消费者拒绝了，酿成了死信，因为simple.queue绑定了死信交换机 dl.direct，因此死信会投递给这个交换机：

假如这个死信交换机也绑定了一个队列，则消息终极会进入这个存放死信的队列：

别的，队列将死信投递给死信交换机时，必须知道两个信息，如许才华确保投递的消息能到达死信交换机，而且正确的路由到死信队列。

• 死信交换机名称
  
• 死信交换机与死信队列绑定的RoutingKey
  

   聊到这可能感觉跟上节中的RepublishMessageRecoverer模式很相似，但上节中消息终极到达消费者并在重试次数耗尽的情况下消息被抛弃，同样把消息发送到一个交换机，再到一个队列，我们称之为异常消息的交换机和异常消息队列，专门去接收无法处理的消息去做独立处理。


但两者还是有差别的，如下图所示。在RepublishMessageRecoverer模式中所有的失败消息是由消费者去做投递的，而在本节是由队列投递到死信交换机的，这是两者的差别。


所以死信交换机也能做到RepublishMessageRecoverer模式这种兜底的结果，只不外死信交换机还会有一些其他的功能和作用。
所以假如只做消息的兜底和异常消息的处理，其实建议只用RepublishMessageRecoverer模式就可以了。


别的这里的小细节需要注意，在RepublishMessageRecoverer模式中，消息是由消费者重新投递到交换机的，在投递过程中我们必须指定交换机的名称，还有就是routingkey，弄错了就无法到达队列。而如今我们在死信交换机当中投递消息的人不是消费者了，而是我们的队列(如图中的simple.queue)，那队列再去做消息投递的时间除了要知道交换机的名字，还得知道死信交换机与死信队列绑定的key，否则不知道routingkey消息也不能到达死信队列。
  

2.1利用死信交换机接收死信

在失败重试计谋中，默认的RejectAndDontRequeueRecoverer失败计谋会在本地重试次数耗尽后，发送reject给RabbitMQ，消息酿成死信，被抛弃。
我们可以给simple.queue添加一个死信交换机，给死信交换机绑定一个队列。如许消息酿成死信后也不会抛弃，而是终极投递到死信交换机，路由到与死信交换机绑定的队列。

我们在consumer服务中，定义一组死信交换机、死信队列：

1. // 声明普通的 simple.queue队列，并且为其指定死信交换机：dl.direct
2. @Bean
3. public Queue simpleQueue2(){
4.     return QueueBuilder.durable("simple.queue") // 指定队列名称，并持久化
5.         .deadLetterExchange("dl.direct") // 指定死信交换机
6.         .build();
7. }
8. // 声明死信交换机 dl.direct
9. @Bean
10. public DirectExchange dlExchange(){
11.     return new DirectExchange("dl.direct", true, false);
12. }
13. // 声明存储死信的队列 dl.queue
14. @Bean
15. public Queue dlQueue(){
16.     return new Queue("dl.queue", true);
17. }
18. // 将死信队列 与 死信交换机绑定
19. @Bean
20. public Binding dlBinding(){
21.     return BindingBuilder.bind(dlQueue()).to(dlExchange()).with("simple");
22. }

复制代码

  总结
什么样的消息会成为死信？
  

• 消息被消费者reject大概返回nack
  
• 消息超时未消费
  
• 队列满了
  

  如何给队列绑定死信交换机?
  

• 给队列设置dead-letter-exchange属性，指定一个交换机；
  
• 给队列设置dead-letter-routing-key属性，设置死信交换机与死信队列的RoutingKey
  

  死信交换机的使用场景是什么？
  

• 假如队列绑定了死信交换机，死信会投递到死信交换机；
  
• 可以利用死信交换机收集所有消费者处理失败的消息（死信），交由人工处理，进一步提高消息队列的可靠性
  

  

2.2TTL联合死信交换机实现延迟消息

上一节我们已经简朴认识了死信交换机，它的作用就是来接收队列中的死信，而死信有三种情况。
第一就是指这个消息消费者处理的时间失败了且终极呢被拒绝了，而且也不重入队就酿成死信了，那死信交换机就可以用来作为一种兜底的方案来去用；第二种是要投递的队列消息满了，无法投递；第三种是指消息超时没有被消费，结果过期了酿成了死信，这个就跟我们的延迟消息啊有关系了，所以呢这节我们就一起来了解一下MQ中的TTL超机遇制，来实现一下这种延迟消息结果。
TTL，也就是Time-To-Live。假如一个队列中的消息在TTL竣事时仍未消费，则会变为死信，ttl超时分为两种情况:

• 消息所在的队列设置了存活时间
  
• 消息本身设置了存活时间
  

   如下图所示，有一个消息的发布者，他把消息投递给一个ttl.direct交换机，而这个交换机呢再与一个叫ttl.queue的队列形成了一个绑定，ttl.queue呢又绑定了一个死信交换机dl.direct，同时呢这个交换机也绑定了一个死信队列dl.queue。
如今假设我们有一个消息他自己设置了ttl时间为5000单位是毫秒，那么当我们消息发出去到达了队列的那一刻他就可以开始倒计时了，倒计时竣事啊这个消息就酿成了死信，从而投递到死信交换机，终极到达了我们的死信队列。这个时间假如恰恰有一个消费者在监听死信队列，那他肯定就可以大概收到这个消息了，那从最后的结果来看是不是生产者发了个消息，结果呢五秒钟后消费者才拿到，实现了一个延迟投递消息的结果呀。
除了我们给消息设置ttl以外，我们也可以给这个队列去设置啊，好比说我给ttl.queue去设置一个ttl，x-message-ttl=10000，那假如说我们的消息和队列都有自己的ttl，这个时间呢我们就会看哪一个更短，以较小的那个数值为准啊。这个呢就是用死信交换机联合ttl实现延迟消息的原理。
  

下面根据上图，我们来实现一下用死信交换机联合ttl实现延迟消息。
1.接收超时死信的死信交换机
在consumer服务的SpringRabbitListener中，定义一个新的消费者，而且声明 死信交换机、死信队列：

1. @RabbitListener(bindings = @QueueBinding(
2.     value = @Queue(name = "dl.ttl.queue", durable = "true"),
3.     exchange = @Exchange(name = "dl.ttl.direct"),
4.     key = "ttl"
5. ))
6. public void listenDlQueue(String msg){
7.     log.info("接收到 dl.ttl.queue的延迟消息：{}", msg);
8. }

复制代码

2.声明一个队列，而且指定TTL
要给队列设置超时时间，需要在声明队列时配置x-message-ttl属性：

1. @Bean
2. public Queue ttlQueue(){
3.     return QueueBuilder.durable("ttl.queue") // 指定队列名称，并持久化
4.         .ttl(10000) // 设置队列的超时时间，10秒
5.         .deadLetterExchange("dl.ttl.direct") // 指定死信交换机
6.         .build();
7. }

复制代码

注意，这个队列设定了死信交换机为dl.ttl.direct。声明交换机，将ttl与交换机绑定：

1. @Bean
2. public DirectExchange ttlExchange(){
3.     return new DirectExchange("ttl.direct");
4. }
5. @Bean
6. public Binding ttlBinding(){
7.     return BindingBuilder.bind(ttlQueue()).to(ttlExchange()).with("ttl");
8. }

复制代码

发送消息，但是不要指定TTL：

1. @Test
2. public void testTTLQueue() {
3.     // 创建消息
4.     String message = "hello, ttl queue";
5.     // 消息ID，需要封装到CorrelationData中
6.     CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
7.     // 发送消息
8.     rabbitTemplate.convertAndSend("ttl.direct", "ttl", message, correlationData);
9.     // 记录日志
10.     log.debug("发送消息成功");
11. }

复制代码

发送消息的日志：

查看下接收消息的日志：

因为队列的TTL值是10000ms，也就是10秒。可以看到消息发送与接收之间的时差刚好是10秒。

3.发送消息时，设定TTL
在发送消息时，也可以指定TTL：

1. @Test
2. public void testTTLMsg() {
3.     // 创建消息
4.     Message message = MessageBuilder
5.         .withBody("hello, ttl message".getBytes(StandardCharsets.UTF_8))
6.         .setExpiration("5000")
7.         .build();
8.     // 消息ID，需要封装到CorrelationData中
9.     CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
10.     // 发送消息
11.     rabbitTemplate.convertAndSend("ttl.direct", "ttl", message, correlationData);
12.     log.debug("发送消息成功");
13. }

复制代码

查看发送消息日志：

接收消息日志：

这次，发送与接收的延迟只有5秒。说明当队列、消息都设置了TTL时，任意一个到期就会成为死信。
   总结
消息超时的两种方式是？
  

• 给队列设置ttl属性，进入队列后超过ttl时间的消息变为死信
  
• 给消息设置ttl属性，队列接收到消息超过ttl时间后变为死信
  

  如何实现发送一个消息20秒后消费者才收到消息？
  

• 给消息的目的队列指定死信交换机
  
• 将消费者监听的队列绑定到死信交换机
  
• 发送消息时给消息设置超时时间为20秒
  

  

2.3延迟队列

利用TTL联合死信交换机，我们实现了消息发出后，消费者延迟收到消息的结果。这种消息模式就称为延迟队列（Delay Queue）模式。
延迟队列的使用场景包罗：

• 延迟发送短信
  
• 用户下单，假如用户在15 分钟内未付出，则主动取消
  
• 预约工作会议，20分钟后主动关照所有参会人员
  

因为延迟队列的需求非常多，所以RabbitMQ的官方也推出了一个插件，原生支持延迟队列结果。
这个插件就是DelayExchange插件。参考RabbitMQ的插件列表页面：

为什么叫DelayExchange呢，使用这个插件的时间它不是基于队列去做的，而是是基于交换机去做的，我们发消息到交换机，而交换机呢会将这个消息帮你延迟的投递到队列里。使用方式可以参考官网所在
下面进行这部门的学习。

1.安装DelayExchange插件
官方的安装指南所在
上述文档是基于linux原生安装RabbitMQ，然后安装插件。
因为我们之前是基于Docker安装RabbitMQ，所以下面我们会解说基于Docker来安装RabbitMQ插件。
RabbitMQ有一个官方的插件社区，所在，此中包含各种各样的插件，包罗我们要使用的DelayExchange插件
各人可以去对应的GitHub页面下载3.8.9版本的插件，所在

因为我们是基于Docker安装，所以需要先查看RabbitMQ的插件目录对应的数据卷。假如不是基于Docker的，请参考第一章部门，重新创建Docker容器。
我们之前设定的RabbitMQ的数据卷名称为mq-plugins，所以我们使用下面下令查看数据卷：

1. docker volume inspect mq-plugins

复制代码

可以得到下面结果：

接下来，将插件上传到这个目录即可：

最后就是安装了，需要进入MQ容器内部来实行安装。我的容器名为mq，所以实行下面下令：

1. docker exec -it mq bash

复制代码

实行时，请将此中的 -it 后面的mq替换为你自己的容器名.进入容器内部后，实行下面下令开启插件：

1. rabbitmq-plugins enable rabbitmq_delayed_message_exchange

复制代码

结果如下：

2.DelayExchange原理
DelayExchange插件的原理是对官方原生的Exchange做了功能的升级:

• 将DelayExchange接受到的消息暂存在内存中(官方的Exchange是无法存储消息的)
  
• 在DelayExchange中计时，超时后才投递消息到队列中
  

在RabbitMQ的管理平台声明一个DelayExchange：

消息的延迟时间需要在发送消息的时间指定:

DelayExchange需要将一个交换机声明为delayed范例。当我们发送消息到delayExchange时，流程如下：

• 接收消息
  
• 判定消息是否具备x-delay属性
  
• 假如有x-delay属性，说明是延迟消息，长期化到硬盘，读取x-delay值，作为延迟时间
  
• 返回routing not found结果给消息发送者
  
• x-delay时间到期后，重新投递消息到指定队列
  

消息发送者发送完消息后：

但是这个报错是我们不需要的，可以去ReturnCallback中配置一下不报这个错：

3.使用DelayExchange
插件的使用也非常简朴：声明一个交换机，交换机的范例可以是任意范例，只需要设定delayed属性为true即可，然后声明队列与其绑定即可。
1）声明DelayExchange交换机
基于注解方式（推荐）：

也可以基于@Bean的方式：

2）发送消息
发送消息时，一定要携带x-delay属性，指定延迟的时间：

   总结
延迟队列插件的使用步骤包罗哪些？
  

• 声明一个交换机，添加delayed属性为true
  
• 发送消息时，添加x-delay头，值为超时时间
  

  

3.消息堆积题目与惰性队列



当生产者发送消息的速度超过了消费者处理消息的速度，就会导致队列中的消息堆积，直到队列存储消息达到上限。之后发送的消息就会成为死信，可能会被抛弃，这就是消息堆积题目。

办理消息堆积有两种思路：

• 增长更多消费者，提高消费速度。也就是我们之前说的work queue模式
  
• 扩大队列容积，提高堆积上限
  

要提升队列容积，把消息生存在内存中显然是不行的。
从RabbitMQ的3.6.0版本开始，就增长了Lazy Queues的概念，也就是惰性队列。惰性队列的特征如下：

• 接收到消息后直接存入磁盘而非内存
  
• 消费者要消费消息时才会从磁盘中读取并加载到内存
  
• 支持数百万条的消息存储
  

下面进行学习。
1.基于下令行设置lazy-queue
而要设置一个队列为惰性队列，只需要在声明队列时，指定x-queue-mode属性为lazy即可。可以通过下令行将一个运行中的队列修改为惰性队列：

1. rabbitmqctl set_policy Lazy "^lazy-queue$" '{"queue-mode":"lazy"}' --apply-to queues  

复制代码

下令解读：

• rabbitmqctl ：RabbitMQ的下令行工具
  
• set_policy ：添加一个计谋
  
• Lazy ：计谋名称，可以自定义
  
• "^lazy-queue$" ：用正则表达式匹配队列的名字
  
• '{"queue-mode":"lazy"}' ：设置队列模式为lazy模式
  
• --apply-to queues ：计谋的作用对象，是所有的队列
  

2.基于@Bean声明lazy-queue

3.基于@RabbitListener声明LazyQueue

下面测试普通队列和惰性队列的区别：

快速向两个队列发送100万条消息，观察控制台发现惰性队列的消息全在磁盘中，内存中一条都没有。

而普通队列中是内存在波动，每次达到了3万多条的时间就会刷去一部门到磁盘，会有一个停息，刷出去了再允许消息来，所以他的稳固性就不是很好。

   总结
消息堆积题目的办理方案？
  

• 队列上绑定多个消费者，提高消费速度
  
• 使用惰性队列，可以再mq中生存更多消息
  

  惰性队列的长处有哪些？
  

• 基于磁盘存储，消息上限高
  
• 没有间歇性的page-out，性能比较稳固
  

  惰性队列的缺点有哪些？
  

• 基于磁盘存储，消息时效性会降低
  
• 性能受限于磁盘的IO
  

  

4.MQ集群

RabbitMQ的是基于Erlang语言编写，而Erlang又是一个面向并发的语言，自然支持集群模式。RabbitMQ的集群有两种模式：

• 普通集群：是一种分布式集群，将队列分散到集群的各个节点，从而提高整个集群的并发能力。
  
• 
  
  
  
  • 普通模式集群不进行数据同步，每个MQ都有自己的队列、数据信息（别的元数据信息如交换机等会同步）。比方我们有2个MQ：mq1，和mq2，假如你的消息在mq1，而你毗连到了mq2，那么mq2会去mq1拉取消息，然后返回给你。假如mq1宕机，消息就会丢失。
    
  
  
• 镜像集群：是一种主从集群，普通集群的根本上，添加了主从备份功能，提高集群的数据可用性。假如一个节点宕机，并不会导致数据丢失。不外，这种方式增长了数据同步的带宽消耗。
  

镜像集群虽然支持主从，但主从同步并不是强同等的，某些情况下可能有数据丢失的风险。因此在RabbitMQ的3.8版本以后，推出了新的功能：使用仲裁队列来代替镜像集群，底层采用Raft协议确保主从的数据同等性。


4.1普通集群

普通集群，大概叫尺度集群（classic cluster），具备下列特征：

• 会在集群的各个节点间共享部门数据，包罗：交换机、队列元信息。不包含队列中的消息。
  
• 当访问集群某节点时，假如队列不在该节点，会从数据所在节点传递到当前节点并返回
  
• 队列所在节点宕机，队列中的消息就会丢失
  

我们先来摆设普通模式集群，我们的筹划摆设3节点的mq集群：
主机名控制台端口amqp通讯端口mq18081 —> 156728071 —> 5672mq28082 —> 156728072 —> 5672mq38083 —> 156728073 —> 5672   这里控制台端口的暴露是8081~8083，为什么暴露成8081~8083而不是15672呢?因为我们是在一个捏造机的机器里，都是15672就辩说了。除了控制台端口还有一个mq的通讯端口，也是一样的，mq通讯端口默认是5672，这里用8071~8073进行区分。
  集群中的节点标示默认都是：rabbit@[hostname]，因此以上三个节点的名称分别为：

• rabbit@mq1
  
• rabbit@mq2
  
• rabbit@mq3
  

1.获取cookie
RabbitMQ底层依赖于Erlang，而Erlang捏造机就是一个面向分布式的语言，默认就支持集群模式。集群模式中的每个RabbitMQ 节点使用 cookie 来确定它们是否被允许相互通讯。
要使两个节点可以大概通讯，它们必须具有雷同的共享秘密，称为Erlang cookie。cookie 只是一串最多 255 个字符的字母数字字符。
每个集群节点必须具有雷同的 cookie。实例之间也需要它来相互通讯。
我们先在之前启动的mq容器中获取一个cookie值，作为集群的cookie。实行下面的下令：

1. docker exec -it mq cat /var/lib/rabbitmq/.erlang.cookie

复制代码

可以看到cookie值如下：

1. FXZMCVGLBIXZCDEMMVZQ

复制代码

接下来，制止并删除当前的mq容器，我们重新搭建集群。

1. docker rm -f mq

复制代码

清理一下数据卷：

1. docker volume prune
2. # 删除所有未使用的数据卷（最好别随便用）

复制代码

2.预备集群配置
在/tmp目录新建一个配置文件 rabbitmq.conf：

1. cd /tmp
2. # 创建文件
3. touch rabbitmq.conf

复制代码

文件内容如下：

1. loopback_users.guest = false
2. listeners.tcp.default = 5672
3. cluster_formation.peer_discovery_backend = rabbit_peer_discovery_classic_config
4. cluster_formation.classic_config.nodes.1 = rabbit@mq1
5. cluster_formation.classic_config.nodes.2 = rabbit@mq2
6. cluster_formation.classic_config.nodes.3 = rabbit@mq3

复制代码

  loopback_users.guest = false禁用guest用户，rabbitmq默认有个guest用户，禁用防止不法分子访问
listeners.tcp.default = 5672 监听的端口，5672mq消息通讯用的这个端口
下面三个配的是集群中的节点信息
  再创建一个文件，记载cookie

1. cd /tmp# 创建cookie文件touch .erlang.cookie# 写入cookieecho "FXZMCVGLBIXZCDEMMVZQ
2. " > .erlang.cookie# 修改cookie文件的权限，改成只读，只有root能读写chmod 600 .erlang.cookie

复制代码

预备三个目录,mq1、mq2、mq3：

1. cd /tmp
2. # 创建目录
3. mkdir mq1 mq2 mq3

复制代码

然后拷贝rabbitmq.conf、cookie文件到mq1、mq2、mq3：

1. # 进入/tmp
2. cd /tmp
3. # 拷贝
4. cp rabbitmq.conf mq1
5. cp rabbitmq.conf mq2
6. cp rabbitmq.conf mq3
7. cp .erlang.cookie mq1
8. cp .erlang.cookie mq2
9. cp .erlang.cookie mq3

复制代码

3.启动集群
创建一个网络以供三个mq互联：

1. docker network create mq-net

复制代码

运行下令，第一个mq结点：

1. docker run -d --net mq-net \
2. -v ${PWD}/mq1/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf \
3. -v ${PWD}/.erlang.cookie:/var/lib/rabbitmq/.erlang.cookie \
4. -e RABBITMQ_DEFAULT_USER=itcast \
5. -e RABBITMQ_DEFAULT_PASS=123321 \
6. --name mq1 \
7. --hostname mq1 \
8. -p 8071:5672 \
9. -p 8081:15672 \
10. rabbitmq:3.8-management

复制代码

  docker run -d --net mq-net \ 加入网络
-v ${PWD}/mq1/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf \ 数据卷挂载：将当前的配置文件挂载到容器里
-v ${PWD}/.erlang.cookie:/var/lib/rabbitmq/.erlang.cookie \ 再挂载cookie文件到容器中的cookie文件中去
-e RABBITMQ_DEFAULT_USER=itcast \用户名与暗码
--name mq1 \ 容器名
--hostname mq1 \ 主机名
-p 8071:5672 \端口映射 5672映射成8071
-p 8081:15672 \ 15672映射成8081
  第二个mq结点：

1. docker run -d --net mq-net \
2. -v ${PWD}/mq2/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf \
3. -v ${PWD}/.erlang.cookie:/var/lib/rabbitmq/.erlang.cookie \
4. -e RABBITMQ_DEFAULT_USER=itcast \
5. -e RABBITMQ_DEFAULT_PASS=123321 \
6. --name mq2 \
7. --hostname mq2 \
8. -p 8072:5672 \
9. -p 8082:15672 \
10. rabbitmq:3.8-management

复制代码

  容器名、主机名还有端口映射改了
  第三个mq结点：

1. docker run -d --net mq-net \
2. -v ${PWD}/mq3/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf \
3. -v ${PWD}/.erlang.cookie:/var/lib/rabbitmq/.erlang.cookie \
4. -e RABBITMQ_DEFAULT_USER=itcast \
5. -e RABBITMQ_DEFAULT_PASS=123321 \
6. --name mq3 \
7. --hostname mq3 \
8. -p 8073:5672 \
9. -p 8083:15672 \
10. rabbitmq:3.8-management

复制代码

4.测试
可以通过下面的下令看mq1的运行状态

1. docker logs -f mq1

复制代码

前端访问mq：192.168.150.101:8081，在mq1这个节点上添加一个队列：

如图，在mq2和mq3两个控制台也都能看到：

1）数据共享测试
点击这个队列，进入管理页面：

然后利用控制台发送一条消息到这个队列：

结果在mq2、mq3上都能看到这条消息，因为之前介绍了，虽然在mq2和3里这仅仅是个引用，但是当我来取消息的时间可以从别的一节点把消息传过来给你，所以你在任意一个节点其实都能看到这个数据。

我们让此中一台节点mq1宕机：

1. docker stop mq1

复制代码

然后登录mq2或mq3的控制台，发现simple.queue也不可用了：

因为他其实只是个引用，假如声明这个队列的节点挂了，那这个队列就不能用了。说明普通集群的队列是没有实现共享的。要想数据规复，除非mq1重新启动。


4.2镜像集群

镜像集群：本质是主从模式，具备下面的特征：

• 交换机、队列、队列中的消息会在各个mq的镜像节点之间同步备份。
  
• 创建队列的节点被称为该队列的主节点，备份到的别的节点叫做该队列的镜像节点。
  
• 一个队列的主节点可能是另一个队列的镜像节点
  
• 所有操纵都是主节点完成，然后同步给镜像节点
  
• 主宕机后，镜像节点会替代成新的主
  

   好比上图，q1的备份在节点2上，又来个q2在结点2上，但是它在结点3上有个备份，q3在结点3上，可以在结点1上有个备份。
此时，结点1是队列1的主节点，同时又是队列3的镜像节点，这种相互备份也可以保证安全性。
  在刚刚的案例中，一旦创建队列的主机宕机，队列就会不可用。不具备高可用能力。假如要办理这个题目，必须使用官方提供的镜像集群方案。官方文档所在

1.镜像模式的特征
默认情况下，队列只生存在创建该队列的节点上。而镜像模式下，创建队列的节点被称为该队列的主节点，队列还会拷贝到集群中的别的节点，也叫做该队列的镜像节点。
但是，不同队列可以在集群中的任意节点上创建，因此不同队列的主节点可以不同。乃至，一个队列的主节点可能是另一个队列的镜像节点。
用户发送给队列的一切哀求，比方发送消息、消息回执默认都会在主节点完成，假如是从节点接收到哀求，也会路由到主节点去完成。镜像节点仅仅起到备份数据作用。
当主节点接收到消费者的ACK时，所有镜像都会删除节点中的数据。
总结如下：

• 镜像队列结构是一主多从（从就是镜像）
  
• 所有操纵都是主节点完成，然后同步给镜像节点
  
• 主宕机后，镜像节点会替代成新的主（假如在主从同步完成前，主就已经宕机，可能出现数据丢失）
  
• 不具备负载均衡功能，因为所有操纵都会有主节点完成（但是不同队列，其主节点可以不同，可以利用这个提高吞吐量）
  

2.镜像模式的配置
镜像模式的配置有3种模式，通过ha-mode指定模式，分别是exactly、all和nodes三种模式。
好比使用exactly模式，那么需要指定一个count参数，count = 镜像数目 + 1，假如count为1，那么镜像数目就为0，所以一般count大于等于2。好比说我如今有一个三节点形成的mq集群，count为2，除了我当前主节点以外，还会从别的两个节点里随机挑一个作为镜像，这就是准确模式的含义。
all模式就简朴了，我不管三七二十一，好比说你有三个节点，如今呢你在第一个节点上创建了队列，那剩下两个自然就成为我的镜像了。同样的我在第二个节点创建的队列，那么其他两个节点一和三也会成为我的镜像，即全部节点都会去备份。这个模式对磁盘和网络都会有额外的压力。
nodes模式就是我在去创建主节点和镜像节点的时间，不再随机了，而他是个指定说哎我就要那个节点，把节点的名称，就是它的标识而设置出来(好比之前我们搭建的某个节点rabbit@mq1)，那这种呢一般情况下就是你对这个机器的特征，好比说它的硬件配置你有详细的了解才去指定。
ha-modeha-params结果准确模式exactly队列的副本量count集群中队列副本（主服务器和镜像服务器之和）的数目。count假如为1意味着单个副本：即队列主节点。count值为2表示2个副本：1个队列主和1个队列镜像。换句话说：count = 镜像数目 + 1。假如群会合的节点数少于count，则该队列将镜像到所有节点。假如有集群总数大于count+1，而且包含镜像的节点出现故障，则将在另一个节点上创建一个新的镜像。all(none)队列在群会合的所有节点之间进行镜像。队列将镜像到任何新加入的节点。镜像到所有节点将对所有聚集节点施加额外的压力，包罗网络I / O，磁盘I / O和磁盘空间使用情况。推荐使用exactly，设置副本数为（N / 2 +1）。nodesnode names指定队列创建到哪些节点，假如指定的节点全部不存在，则会出现异常。假如指定的节点在集群中存在，但是暂时不可用，会创建节点到当前客户端毗连到的节点。 这里我们以rabbitmqctl下令作为案例来解说配置语法。

1）exactly模式

1. rabbitmqctl set_policy ha-two "^two\." '{"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"}'

复制代码

• rabbitmqctl set_policy：固定写法
  
• ha-two：计谋名称，自定义
  
• "^two\."：匹配队列的正则表达式，符合命名规则的队列才生效，这里是任何以two.开头的队列名称
  
• '{"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"}': 计谋内容
  
  
  
  • "ha-mode":"exactly"：计谋模式，此处是exactly模式，指定副本数目
    
  • "ha-params":2：计谋参数，这里是count=2，就是副本数目为2，1主1镜像
    
  • "ha-sync-mode":"automatic"：同步计谋，默认是manual，即新加入的镜像节点不会同步旧的消息。假如设置为automatic，则新加入的镜像节点会把主节点中所有消息都同步，会带来额外的网络开销
    
  
  

2）all模式

1. rabbitmqctl set_policy ha-all "^all\." '{"ha-mode":"all"}'

复制代码

• ha-all：计谋名称，自定义
  
• "^all\."：匹配所有以all.开头的队列名
  
• '{"ha-mode":"all"}'：计谋内容
  
  
  
  • "ha-mode":"all"：计谋模式，此处是all模式，即所有节点都会称为镜像节点
    
  
  

3）nodes模式

1. rabbitmqctl set_policy ha-nodes "^nodes\." '{"ha-mode":"nodes","ha-params":["rabbit@nodeA", "rabbit@nodeB"]}'

复制代码

• rabbitmqctl set_policy：固定写法
  
• ha-nodes：计谋名称，自定义
  
• "^nodes\."：匹配队列的正则表达式，符合命名规则的队列才生效，这里是任何以nodes.开头的队列名称
  
• '{"ha-mode":"nodes","ha-params":["rabbit@nodeA", "rabbit@nodeB"]}': 计谋内容
  
  
  
  • "ha-mode":"nodes"：计谋模式，此处是nodes模式
    
  • "ha-params":["rabbit@mq1", "rabbit@mq2"]：计谋参数，这里指定副本所在节点名称
    
  
  

3.测试
我们测试使用exactly模式的镜像，因为集群节点数目为3，因此镜像数目就设置为2.
运行下面的下令，进入任意一个结点都行：

1. docker exec -it mq1 bash

复制代码

然后输入下令：

1. rabbitmqctl set_policy ha-two "^two\." '{"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"}'

复制代码

大概直接一个下令：

1. docker exec -it mq1 rabbitmqctl set_policy ha-two "^two\." '{"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"}'

复制代码

此时随便访问一个结点的Admin的Policies界面可以看到我们配的计谋：

下面，我们创建一个新的队列：

在任意一个mq控制台查看队列，这个节点它的名称后面来了个加一，就是有一个mirror，镜像的意思，那不消说，我在这个节点上去写消息，别的的镜像节点也能看到了。这里节点2是节点1的镜像节点，节点3也能看到，但仅仅是引用(雷同普通集群)。

此时让mq1宕机，mq2就成为了该队列的主节点，而且mq3成为了镜像节点，非常强盛。此时再启动mq1，这个队列就跟它不要紧了。

4.3仲裁队列

镜像队列他虽然可以大概做主从，但是啊主从同步时可能会有数据丢失的风险，因为它不是这种强同等的啊，但仲裁队列了就可以办理这些题目。
仲裁队列：仲裁队列是3.8版本以后才有的新功能，用来替代镜像队列，但使用更加方便，具备下列特征：

• 与镜像队列一样，都是主从模式，支持主从数据同步
  
• 使用非常简朴，没有复杂的配置(镜像队列下令复杂，设置模式、还有模式对应参数)，而仲裁队列已经把这些东西设成默认的了，默认count=5
  
• 主从同步基于Raft协议，强同等，不消担心数据丢失
  

1.添加仲裁队列
仲裁队列的配置非常简朴，直接在任意控制台添加一个队列，但是一定要选择队列范例为Quorum范例。这里我们的集群已有三个健康的结点，我们进行配置，设置mq1为主节点：

在任意控制台查看队列：

可以看到，仲裁队列的 + 2字样。代表这个队列有2个镜像节点。
因为仲裁队列默认的镜像数为5。假如你的集群有7个节点，那么从结点数肯定是4；而我们集群只有3个节点，那就成all模式了，因此从结点数目就是2.

2.测试
测试这里省略。可以参考对镜像集群的测试，结果是一样的。

3.集群扩容
（1）加入集群
启动一个新的MQ容器：

1. docker run -d --net mq-net \
2. -v ${PWD}/.erlang.cookie:/var/lib/rabbitmq/.erlang.cookie \
3. -e RABBITMQ_DEFAULT_USER=itcast \
4. -e RABBITMQ_DEFAULT_PASS=123321 \
5. --name mq4 \
6. --hostname mq5 \
7. -p 8074:15672 \
8. -p 8084:15672 \
9. rabbitmq:3.8-management

复制代码

进入容器控制台：

1. docker exec -it mq4 bash

复制代码

制止mq进程

1. rabbitmqctl stop_app

复制代码

重置RabbitMQ中的数据：

1. rabbitmqctl reset

复制代码

加入mq1：

1. rabbitmqctl join_cluster rabbit@mq1

复制代码

再次启动mq进程：

1. rabbitmqctl start_app

复制代码

（2）增长仲裁队列副本
我们先查看下quorum.queue这个队列目前的副本情况，进入mq1容器：

1. docker exec -it mq1 bash

复制代码

实行下令：

1. rabbitmq-queues quorum_status "quorum.queue"

复制代码

结果：

如今，我们让mq4也加入进来：

1. rabbitmq-queues add_member "quorum.queue" "rabbit@mq4"

复制代码

结果：

再次查看：

1. rabbitmq-queues quorum_status "quorum.queue"

复制代码

查看控制台，发现quorum.queue的镜像数目也从原来的 +2 酿成了 +3：

4.Java代码创建仲裁队列
上面搭建了仲裁队列集群，是基于欣赏器的控制台去做的。但终极我们肯定还要用iava代码来做。
其实啊仲裁队列本身就是一个普通的队列，集群搭建好了以后你只要创建它就能形成集群了，那因此呢我们这里的代码也非常的简朴，仅仅是声明了一个bean，然后呢就是一个队列，区别在于我们这里多了一个.quorum()，那么这个就是标记当前队列是一个仲裁队列，那么将来呢我们的集群就会主动给它形成主和镜像了，所以非常的简朴。

1. @Bean
2. public Queue quorumQueue() {
3.     return QueueBuilder
4.         .durable("quorum.queue") // 持久化
5.         .quorum() // 仲裁队列
6.         .build();
7. }

复制代码

5.SpringAMQP毗连MQ集群
之前的代码我们毗连的是单点mq:

那你连一个mq这么写，但如今是集群呀，有8071、8072、8073好几个，怎么写呢?
注意，这里用address来代替host、port方式

1. spring:
2.   rabbitmq:
3.     addresses: 192.168.150.105:8071, 192.168.150.105:8072, 192.168.150.105:8073
4.     username: itcast
5.     password: 123321
6.     virtual-host: /

复制代码

这个时间启动项目(记得把之前声明的一些配置类给注释掉)，前端控制台就能看到我们在代码里声明的仲裁队列了。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。