RabbitMQ深入学习

继续上一节的学习，上一节学习了RabbitMQ的基本内容，本节学习RabbitMQ的高级特性。

RocketMQ的高级特性学习见这篇博客

1.消息可靠性
- 1.1生产者消息确认
- 1.2消息持久化
- 1.3消费者消息确认
- 1.4消费失败重试机制
- 1.5消息可靠性保证总结
2.什么是死信交换机
- 2.1利用死信交换机接收死信
- 2.2TTL结合死信交换机实现延迟消息
- 2.3延迟队列
3.消息堆积问题与惰性队列
4.MQ集群
- 4.1普通集群
- 4.2镜像集群
- 4.3仲裁队列

在这里插入图片描述

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文件，添加下面的内容：

spring:rabbitmq:publisher-confirm-type: correlatedpublisher-returns: truetemplate: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服务，添加一个：

package cn.itcast.mq.config;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Configuration;@Slf4j
@Configuration
public class CommonConfig implements ApplicationContextAware {@Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {// 获取RabbitTemplate对象RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);// 配置ReturnCallbackrabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {// 投递失败，记录日志log.info("消息发送失败，应答码{}，原因{}，交换机{}，路由键{},消息{}",replyCode, replyText, exchange, routingKey, message.toString());// 如果有业务需要，可以重发消息});}
}

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类中，定义一个单元测试方法：

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

发送之前得将交换机跟队列绑定好，，否则即使消息到了交换机，也会因为找不到队列而触发回退机制ReturnCallback。
方式一：通过代码方式进行绑定（推荐），前面文章有介绍。
方式二：手动在 RabbitMQ 管理界面中绑定。这种方式适用于测试或临时使用，但不推荐用于生产，因为无法版本控制和自动部署。

测试发现提示投递成功，修改投递的交换机名称模拟写错，模拟投递失败，测试ConfirmCallback是否正常触发；修改投递的routingKey模拟写错，模拟路由失败，测试ReturnCallback是否正常触发。

1.2消息持久化

生产者确认可以确保消息投递到RabbitMQ的队列中，但是消息发送到RabbitMQ以后，如果突然宕机，也可能导致消息丢失。要想确保消息在RabbitMQ中安全保存，必须开启消息持久化机制。

我们先去看一下现象，这是我们之前准备的一条队列，然后在这个队列里面已经有一条消息了在这里插入图片描述

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

在这里插入图片描述

然后我们通过过docker restart mq完成重启mq，重启完成发现交换机都在，但是队列没了，队列都没了何谈队列当中的消息呢。那为什么我们自己创建的队列消息都没了但系统的交换机还在呢。这说明系统的交换机是持久化的。

我们观察可知其实系统创建的交换机都带了一个特征(Features)——D，对应持久化。那因此呢我们要想让我们的队列也能持久化，让我们的交换机也能持久化，都应该带上这个durable的参数，在前端我们创建一个交换机的时候Durability参数设置为Durable即可，创建队列时同理。控制代码去创建我们这节将会学习。

消息持久化分为：

交换机持久化
队列持久化
消息持久化

1.交换机持久化

RabbitMQ中交换机默认是非持久化的，mq重启后就丢失。SpringAMQP中声明交换机时可以通过代码指定交换机持久化：

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

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

在这里插入图片描述

2.队列持久化

RabbitMQ中队列默认是非持久化的，mq重启后就丢失。

SpringAMQP中声明队列时可以通过代码指定交换机持久化：

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

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

在这里插入图片描述

3.消息持久化

队列持久化不代表消息就持久化，当交换机和队列持久化时如果宕机，消息仍然会丢失。

利用SpringAMQP发送消息时，可以设置消息的属性（MessageProperties），指定delivery-mode：

1：非持久化
2：持久化

用java代码指定：

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

我们之前的文章中发消息没使用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文件，添加下面内容：

spring:rabbitmq:listener:simple:acknowledge-mode: none # none,关闭ack;manual,手动ack;auto,自动ack

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

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

测试可以发现，当消息处理抛异常时，消息依然被RabbitMQ删除了。

2.演示auto模式

再次把确认机制修改为auto:

spring:rabbitmq:listener:simple: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文件，添加内容：

spring:rabbitmq:listener:simple:retry:enabled: true # 开启消费者失败重试initial-interval: 1000 # 初始的失败等待时长为1秒multiplier: 1 # 失败的等待时长倍数，下次等待时长 = multiplier * 上一次等待时长，这里等待时长永远是1s；若设置为2就是1 2 4 8s...max-attempts: 3 # 最大重试次数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服务中定义处理失败消息的交换机和队列

@Bean
public DirectExchange errorMessageExchange(){return new DirectExchange("error.direct");
}
@Bean
public Queue errorQueue(){return new Queue("error.queue", true);
}
@Bean
public Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange){return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with("error");
}

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

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

完整代码：

package cn.itcast.mq.config;import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.retry.MessageRecoverer;
import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer;
import org.springframework.context.annotation.Bean;@Configuration
public class ErrorMessageConfig {@Beanpublic DirectExchange errorMessageExchange(){return new DirectExchange("error.direct");}@Beanpublic Queue errorQueue(){return new Queue("error.queue", true);}@Beanpublic Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange){return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with("error");}@Beanpublic MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");}
}

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服务中，定义一组死信交换机、死信队列：

// 声明普通的 simple.queue队列，并且为其指定死信交换机：dl.direct
@Bean
public Queue simpleQueue2(){return QueueBuilder.durable("simple.queue") // 指定队列名称，并持久化.deadLetterExchange("dl.direct") // 指定死信交换机.build();
}
// 声明死信交换机 dl.direct
@Bean
public DirectExchange dlExchange(){return new DirectExchange("dl.direct", true, false);
}
// 声明存储死信的队列 dl.queue
@Bean
public Queue dlQueue(){return new Queue("dl.queue", true);
}
// 将死信队列 与 死信交换机绑定
@Bean
public Binding dlBinding(){return BindingBuilder.bind(dlQueue()).to(dlExchange()).with("simple");
}

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

消息被消费者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中，定义一个新的消费者，并且声明死信交换机、死信队列：

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

2.声明一个队列，并且指定TTL

要给队列设置超时时间，需要在声明队列时配置x-message-ttl属性：

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

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

@Bean
public DirectExchange ttlExchange(){return new DirectExchange("ttl.direct");
}
@Bean
public Binding ttlBinding(){return BindingBuilder.bind(ttlQueue()).to(ttlExchange()).with("ttl");
}

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

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

发送消息的日志：

在这里插入图片描述

查看下接收消息的日志：

在这里插入图片描述

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

3.发送消息时，设定TTL

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

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

查看发送消息日志：

在这里插入图片描述

接收消息日志：

在这里插入图片描述

这次，发送与接收的延迟只有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，所以我们使用下面命令查看数据卷：

docker volume inspect mq-plugins

可以得到下面结果：

在这里插入图片描述

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

在这里插入图片描述

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

docker exec -it mq bash

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

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即可。可以通过命令行将一个运行中的队列修改为惰性队列：

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通信端口
mq1	8081 —> 15672	8071 —> 5672
mq2	8082 —> 15672	8072 —> 5672
mq3	8083 —> 15672	8073 —> 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。执行下面的命令：

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

可以看到cookie值如下：

FXZMCVGLBIXZCDEMMVZQ

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

docker rm -f mq

清理一下数据卷：

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

2.准备集群配置

在/tmp目录新建一个配置文件 rabbitmq.conf：

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

文件内容如下：

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

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

再创建一个文件，记录cookie

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

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

cd /tmp
# 创建目录
mkdir mq1 mq2 mq3

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

# 进入/tmp
cd /tmp
# 拷贝
cp rabbitmq.conf mq1
cp rabbitmq.conf mq2
cp rabbitmq.conf mq3
cp .erlang.cookie mq1
cp .erlang.cookie mq2
cp .erlang.cookie mq3

3.启动集群

创建一个网络以供三个mq互联：

docker network create mq-net

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

docker run -d --net mq-net \
-v ${PWD}/mq1/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf \
-v ${PWD}/.erlang.cookie:/var/lib/rabbitmq/.erlang.cookie \
-e RABBITMQ_DEFAULT_USER=itcast \
-e RABBITMQ_DEFAULT_PASS=123321 \
--name mq1 \
--hostname mq1 \
-p 8071:5672 \
-p 8081:15672 \
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结点：

docker run -d --net mq-net \
-v ${PWD}/mq2/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf \
-v ${PWD}/.erlang.cookie:/var/lib/rabbitmq/.erlang.cookie \
-e RABBITMQ_DEFAULT_USER=itcast \
-e RABBITMQ_DEFAULT_PASS=123321 \
--name mq2 \
--hostname mq2 \
-p 8072:5672 \
-p 8082:15672 \
rabbitmq:3.8-management

容器名、主机名还有端口映射改了

第三个mq结点：

docker run -d --net mq-net \
-v ${PWD}/mq3/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf \
-v ${PWD}/.erlang.cookie:/var/lib/rabbitmq/.erlang.cookie \
-e RABBITMQ_DEFAULT_USER=itcast \
-e RABBITMQ_DEFAULT_PASS=123321 \
--name mq3 \
--hostname mq3 \
-p 8073:5672 \
-p 8083:15672 \
rabbitmq:3.8-management

4.测试

可以通过下面的命令看mq1的运行状态

docker logs -f mq1

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

在这里插入图片描述

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

在这里插入图片描述

1）数据共享测试

点击这个队列，进入管理页面：

在这里插入图片描述

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

在这里插入图片描述

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

在这里插入图片描述

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

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-mode	ha-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）。
nodes	node names	指定队列创建到哪些节点，如果指定的节点全部不存在，则会出现异常。如果指定的节点在集群中存在，但是暂时不可用，会创建节点到当前客户端连接到的节点。

这里我们以rabbitmqctl命令作为案例来讲解配置语法。

1）exactly模式

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模式

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

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

3）nodes模式

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.

运行下面的命令，进入任意一个结点都行：

docker exec -it mq1 bash

然后输入命令：

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

或者直接一个命令：

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容器：

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

进入容器控制台：

docker exec -it mq4 bash

停止mq进程

rabbitmqctl stop_app

重置RabbitMQ中的数据：

rabbitmqctl reset

加入mq1：

rabbitmqctl join_cluster rabbit@mq1

再次启动mq进程：

rabbitmqctl start_app

在这里插入图片描述

（2）增加仲裁队列副本

我们先查看下quorum.queue这个队列目前的副本情况，进入mq1容器：

docker exec -it mq1 bash

执行命令：

rabbitmq-queues quorum_status "quorum.queue"

结果：

在这里插入图片描述

现在，我们让mq4也加入进来：

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

结果：

在这里插入图片描述

再次查看：

rabbitmq-queues quorum_status "quorum.queue"

在这里插入图片描述

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

在这里插入图片描述

4.Java代码创建仲裁队列

上面搭建了仲裁队列集群，是基于浏览器的控制台去做的。但最终我们肯定还要用iava代码来做。

其实啊仲裁队列本身就是一个普通的队列，集群搭建好了以后你只要创建它就能形成集群了，那因此呢我们这里的代码也非常的简单，仅仅是声明了一个bean，然后呢就是一个队列，区别在于我们这里多了一个.quorum()，那么这个就是标记当前队列是一个仲裁队列，那么将来呢我们的集群就会自动给它形成主和镜像了，所以非常的简单。

@Bean
public Queue quorumQueue() {return QueueBuilder.durable("quorum.queue") // 持久化.quorum() // 仲裁队列.build();
}

5.SpringAMQP连接MQ集群

之前的代码我们连接的是单点mq:

在这里插入图片描述

那你连一个mq这么写，但现在是集群呀，有8071、8072、8073好几个，怎么写呢?

注意，这里用address来代替host、port方式

spring:rabbitmq:addresses: 192.168.150.105:8071, 192.168.150.105:8072, 192.168.150.105:8073username: itcastpassword: 123321virtual-host: /

在这里插入图片描述

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