关于RabbitMQ-阿里云开发者社区

异步通信消息队列RabbitMQ高级.xmind

首先MQ是什么？

MQ是一种应用程序键一步通讯的技术，MQ是消息队列的缩写（Message Queue）

在MQ中，消息由一个应用程序发送到一个称为队列的中间件中，接着被中间件存储，并最终被另一个或多个消费者应用程序读取和处理；MQ组成：消息——生产者——队列——中间件——消费者！

Broker：是一个像数据总线一样的东西，所有的服务要接收数据和发送数据都发到这个总线上，这个总线就像协议一样，让服务间的通讯变得标准和可控。

virtualHost：虚拟主机，隔离不同租户的exchange、queue、消息的隔离

MQ的好处：

使用MQ可以有效解耦应用程序之间的依赖，提高应用程序的可扩展性、可靠性和可维护性。

同时，它还可以提高系统整体的吞吐量、降低响应时间，对于高并发、高可用、高性能的分布式系统架构设计具有重要意义。

同步和异步的区别：

同步：时效性强，可以立即得到效果，需要双方协同工作

异步：分离责任，双方各自独立完成自己的工作，在实际开发中，可以更好地提高系统的并发性能和效率；

RabbitMQ的安装：

下载镜像：docker pull rabbitmq:3.8-management

dockerrun\-eRABBITMQ_DEFAULT_USER=test\-eRABBITMQ_DEFAULT_PASS=***\-vmq-plugins:/plugins\--namemq\--hostnamemq\-p15672:15672\-p5672:5672\-d\rabbitmq:3.8-management

publisher生成者实现：

思路：

建立connection
创建channel
利用channel声明队列
利用channel向队列发送消息

publicclassPublisherTest {
@TestpublicvoidtestSendMessage() throwsIOException, TimeoutException {
// 1.建立连接ConnectionFactoryfactory=newConnectionFactory();
// 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码factory.setHost("192.168.***.101");
factory.setPort(5672);
factory.setVirtualHost("/");
factory.setUsername("test");
factory.setPassword("***");
// 1.2.建立连接Connectionconnection=factory.newConnection();
// 2.创建通道ChannelChannelchannel=connection.createChannel();
// 3.创建队列StringqueueName="simple.queue";
channel.queueDeclare(queueName, false, false, false, null);
// 4.发送消息Stringmessage="hello, rabbitmq!";
channel.basicPublish("", queueName, null, message.getBytes());
System.out.println("发送消息成功：【"+message+"】");
// 5.关闭通道和连接channel.close();
connection.close();
    }
}

consumer实现：

思路：

建立connection
创建channel
利用channel声明队列
定义consumer的消费行为handleDelivery()
利用channel将消费者与队列绑定

publicclassConsumerTest {
publicstaticvoidmain(String[] args) throwsIOException, TimeoutException {
// 1.建立连接ConnectionFactoryfactory=newConnectionFactory();
// 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码factory.setHost("192.168.***.101");
factory.setPort(5672);
factory.setVirtualHost("/");
factory.setUsername("test");
factory.setPassword("***");
// 1.2.建立连接Connectionconnection=factory.newConnection();
// 2.创建通道ChannelChannelchannel=connection.createChannel();
// 3.创建队列StringqueueName="simple.queue";
channel.queueDeclare(queueName, false, false, false, null);
// 4.订阅消息channel.basicConsume(queueName, true, newDefaultConsumer(channel){
@OverridepublicvoidhandleDelivery(StringconsumerTag, Envelopeenvelope,
AMQP.BasicPropertiesproperties, byte[] body) throwsIOException {
// 5.处理消息Stringmessage=newString(body);
System.out.println("接收到消息：【"+message+"】");
            }
        });
System.out.println("等待接收消息。。。。");
    }
}

SpringAMQP

pringAMQP是基于RabbitMQ封装的一套模板，并且还利用SpringBoot对其实现了自动装配，使用起来非常方便。SpringAmqp的官方地址：https://spring.io/projects/spring-amqp

SpringAMQP提供了三个功能：

自动声明队列、交换机及其绑定关系
基于注解的监听器模式，异步接收消息
封装了RabbitTemplate工具，用于发送消息

<!--AMQP依赖，包含RabbitMQ--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency>spring:
  rabbitmq:
    host: 192.168.***.101 # 主机名
    port: 5672 # 端口
    virtual-host: / # 虚拟主机
    username: test # 用户名
    password: *** # 密码

队列：

Basic Queue 简单队列：是一种先进先出（FIFO）的队列，通过将任务依次放入队列中，然后逐个处理队列中的每个任务。这种机制通常是同步的，即当前任务执行完毕后才会开始下一个任务的处理。

只有一个队列多个消费者

workQueue 工作队列：是一种异步的任务处理机制，它在内核中运行，并使用线程池来处理任务。它允许多个任务同时进行处理，并能够自动调整线程池中的线程数量以适应系统负载。

总之，Basic queue适合于串行处理任务，而Workqueue适合于并行处理任务。

交换机：

Exchange有以下3种类型：

Fanout：广播，将消息交给所有绑定到交换机的队列
Direct：定向，把消息交给符合指定routing key 的队列
Topic：通配符，把消息交给符合routing pattern（路由模式）的队列

交换机只负责转发信息，不具备存储消息的能力。

Fanout（广播）交换机：

@ConfigurationpublicclassFanoutConfig {
/*** 声明交换机* @return Fanout类型交换机*/@BeanpublicFanoutExchangefanoutExchange(){
returnnewFanoutExchange("test.fanout");
    }
/*** 第1个队列*/@BeanpublicQueuefanoutQueue1(){
returnnewQueue("fanout.queue1");
    }
/*** 绑定队列和交换机*/@BeanpublicBindingbindingQueue1(QueuefanoutQueue1, FanoutExchangefanoutExchange){
returnBindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
    }
/*** 第2个队列*/@BeanpublicQueuefanoutQueue2(){
returnnewQueue("fanout.queue2");
    }
/*** 绑定队列和交换机*/@BeanpublicBindingbindingQueue2(QueuefanoutQueue2, FanoutExchangefanoutExchange){
returnBindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
    }
}

消息发送：

@TestpublicvoidtestFanoutExchange() {
// 队列名称StringexchangeName="test.fanout";
// 消息Stringmessage="hello, everyone!";
rabbitTemplate.convertAndSend(exchangeName, "", message);
}

消息接收：

@RabbitListener(queues="fanout.queue1")
publicvoidlistenFanoutQueue1(Stringmsg) {
System.out.println("消费者1接收到Fanout消息：【"+msg+"】");
}
@RabbitListener(queues="fanout.queue2")
publicvoidlistenFanoutQueue2(Stringmsg) {
System.out.println("消费者2接收到Fanout消息：【"+msg+"】");
}

声明队列、交换机、绑定关系的Bean是什么？

Queue
FanoutExchange
Binding

Direct交换机：（定向交换机）

基于@Bean的方式声明队列和交换机比较麻烦，Spring还提供了基于注解方式来声明。

添加两个消费者，同时基于注解来声明队列和交换机：

@RabbitListener(bindings=@QueueBinding(
value=@Queue(name="direct.queue1"),
exchange=@Exchange(name="test.direct", type=ExchangeTypes.DIRECT),
key= {"red", "blue"}
))
publicvoidlistenDirectQueue1(Stringmsg){
System.out.println("消费者接收到direct.queue1的消息：【"+msg+"】");
}
@RabbitListener(bindings=@QueueBinding(
value=@Queue(name="direct.queue2"),
exchange=@Exchange(name="test.direct", type=ExchangeTypes.DIRECT),
key= {"red", "yellow"}
))
publicvoidlistenDirectQueue2(Stringmsg){
System.out.println("消费者接收到direct.queue2的消息：【"+msg+"】");
}

描述下Direct交换机与Fanout交换机的差异？

Fanout交换机将消息路由给每一个与之绑定的队列
Direct交换机根据RoutingKey判断路由给哪个队列
如果多个队列具有相同的RoutingKey，则与Fanout功能类似

基于@RabbitListener注解声明队列和交换机有哪些常见注解？

Topic（主题交换机）：

Topic类型的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符！

Routingkey 一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert

通配符规则：

#：匹配0个或多个词

*：匹配不多不少恰好1个词

举例：

item.#：能够匹配item.spu.insert 或者 item.spu

item.*：只能匹配item.spu

消息接收：

@RabbitListener(bindings=@QueueBinding(
value=@Queue(name="topic.queue1"),
exchange=@Exchange(name="test.topic", type=ExchangeTypes.TOPIC),
key="***.#"))
publicvoidlistenTopicQueue1(Stringmsg){
System.out.println("消费者接收到topic.queue1的消息：【"+msg+"】");
}
@RabbitListener(bindings=@QueueBinding(
value=@Queue(name="topic.queue2"),
exchange=@Exchange(name="test.topic", type=ExchangeTypes.TOPIC),
key="#.news"))
publicvoidlistenTopicQueue2(Stringmsg){
System.out.println("消费者接收到topic.queue2的消息：【"+msg+"】");
}

Topic交换机接收的消息RoutingKey必须是多个单词，以 **.** 分割
Topic交换机与队列绑定时的bindingKey可以指定通配符
#：代表0个或多个词
*：代表1个词

消息转换器：

配置JSON转换器：

引入依赖

<dependency><groupId>com.fasterxml.jackson.dataformat</groupId><artifactId>jackson-dataformat-xml</artifactId><version>2.9.10</version></dependency>

在启动类中添加一个Bean即可：

每个工程启动类均增加：

@BeanpublicMessageConverterjsonMessageConverter(){
returnnewJackson2JsonMessageConverter();
    }

MQ生产环境遇到的问题

消息可靠性：

消息从发送到消费者接收，会经历多个阶段，每个阶段都有可能导致消息丢失，常见因数：

发送时丢失：

生产者发送的消息未送达exchange
消息到达exchange后未到达queue

MQ宕机，queue将消息丢失
consumer接收到消息后未消费就宕机

针对这些问题，RabbitMQ分别给出了解决方案：

生产者确认机制
mq持久化
消费者确认机制
失败重试机制

生成者确认机制：

修改配置：

spring:
  rabbitmq:
    publisher-confirm-type: correlated
    publisher-returns: true
    template:
      mandatory: true

说明：

publish-confirm-type：开启publisher-confirm，这里支持两种类型：

simple：同步等待confirm结果，直到超时
correlated：异步回调，定义ConfirmCallback，MQ返回结果时会回调这个ConfirmCallback

publish-returns：开启publish-return功能，同样是基于callback机制，不过是定义ReturnCallback
template.mandatory：定义消息路由失败时的策略。true，则调用ReturnCallback；false：则直接丢弃消息

定义Return回调：

//每个RabbitTemplate只能配置一个ReturnCallback@Slf4j@ConfigurationpublicclassCommonConfigimplementsApplicationContextAware {
@OverridepublicvoidsetApplicationContext(ApplicationContextapplicationContext) throwsBeansException {
// 获取RabbitTemplateRabbitTemplaterabbitTemplate=applicationContext.getBean(RabbitTemplate.class);
// 设置ReturnCallbackrabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
// 投递失败，记录日志log.info("消息发送失败，应答码{}，原因{}，交换机{}，路由键{},消息{}",
replyCode, replyText, exchange, routingKey, message.toString());
// 如果有业务需要，可以重发消息        });
    }
}

定义ConfirmCallback：

//Test启动类中定义：publicvoidtestSendMessage2SimpleQueue() throwsInterruptedException {
// 1.消息体Stringmessage="hello, spring amqp!";
// 2.全局唯一的消息ID，需要封装到CorrelationData中CorrelationDatacorrelationData=newCorrelationData(UUID.randomUUID().toString());
// 3.添加callbackcorrelationData.getFuture().addCallback(
result-> {
if(result.isAck()){
// 3.1.ack，消息成功log.debug("消息发送成功, ID:{}", correlationData.getId());
            }else{
// 3.2.nack，消息失败log.error("消息发送失败, ID:{}, 原因{}",correlationData.getId(), result.getReason());
            }
        },
ex->log.error("消息发送异常, ID:{}, 原因{}",correlationData.getId(),ex.getMessage())
    );
// 4.发送消息rabbitTemplate.convertAndSend("task.direct", "task", message, correlationData);
// 休眠一会儿，等待ack回执Thread.sleep(2000);
}

消息持久化：

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

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

交换机持久化：

RabbitMQ中交换机默认是非持久化的，mq重启后就丢失

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

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

队列持久化：

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

@BeanpublicQueuesimpleQueue(){
// 使用QueueBuilder构建队列，durable就是持久化的returnQueueBuilder.durable("simple.queue").build();
}

消息持久化：

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

1：非持久化
2：持久化

默认情况下，SpringAMQP发出的任何消息都是持久化的，不用特意指定。

消费者消息确认：

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

SpringAMQP则允许配置三种确认模式：

•manual：手动ack，需要在业务代码结束后，调用api发送ack。

•auto：自动ack，由spring监测listener代码是否出现异常，没有异常则返回ack；抛出异常则返回nack

•none：关闭ack，MQ假定消费者获取消息后会成功处理，因此消息投递后立即被删除

由此可知：

none模式下，消息投递是不可靠的，可能丢失
auto模式类似事务机制，出现异常时返回nack，消息回滚到mq；没有异常，返回ack
manual：自己根据业务情况，判断什么时候该ack

一般，我们都是使用默认的auto即可。

演示auto模式：

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: auto # 关闭ack

消费失败重试机制：

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

本地重试：

我们可以利用Spring的retry机制，在消费者出现异常时利用本地重试，而不是无限制的requeue到mq队列

spring:
  rabbitmq:
    listener:
      simple:
        retry:
          enabled: true # 开启消费者失败重试
          initial-interval: 1000 # 初识的失败等待时长为1秒
          multiplier: 1 # 失败的等待时长倍数，下次等待时长 = multiplier * last-interval
          max-attempts: 3 # 最大重试次数
          stateless: true # true无状态；false有状态。如果业务中包含事务，这里改为false

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

失败策略：

在之前的测试中，达到最大重试次数后，消息会被丢弃，这是由Spring内部机制决定的。

在开启重试模式后，重试次数耗尽，如果消息依然失败，则需要有MessageRecovery接口来处理，它包含三种不同的实现：

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

比较优雅的一种处理方案是RepublishMessageRecoverer，失败后将消息投递到一个指定的，专门存放异常消息的队列，后续由人工集中处理。

@ConfigurationpublicclassErrorMessageConfig {
@BeanpublicDirectExchangeerrorMessageExchange(){
returnnewDirectExchange("error.direct");
    }
@BeanpublicQueueerrorQueue(){
returnnewQueue("error.queue", true);
    }
@BeanpublicBindingerrorBinding(QueueerrorQueue, DirectExchangeerrorMessageExchange){
returnBindingBuilder.bind(errorQueue).to(errorMessageExchange).with("error");
    }
//定义一个RepublishMessageRecoverer，关联队列和交换机@BeanpublicMessageRecovererrepublishMessageRecoverer(RabbitTemplaterabbitTemplate){
returnnewRepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");
    }
}

总结：

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

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

死信交换机：

什么是死信交换机：

消息被消费者拒绝后，会成为死信，死信投递给死信交换机，然后进入存放死信的队列

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

// 声明普通的 simple.queue队列，并且为其指定死信交换机：dl.direct@BeanpublicQueuesimpleQueue2(){
returnQueueBuilder.durable("simple.queue") // 指定队列名称，并持久化        .deadLetterExchange("dl.direct") // 指定死信交换机        .build();
}
// 声明死信交换机 dl.direct@BeanpublicDirectExchangedlExchange(){
returnnewDirectExchange("dl.direct", true, false);
}
// 声明存储死信的队列 dl.queue@BeanpublicQueuedlQueue(){
returnnewQueue("dl.queue", true);
}
// 将死信队列 与 死信交换机绑定@BeanpublicBindingdlBinding(){
returnBindingBuilder.bind(dlQueue()).to(dlExchange()).with("simple");
}

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

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

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

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

TTL：

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

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

@BeanpublicQueuettlQueue(){
returnQueueBuilder.durable("ttl.queue") // 指定队列名称，并持久化        .ttl(10000) // 设置队列的超时时间，10秒        .deadLetterExchange("dl.ttl.direct") // 指定死信交换机        .deadLetterRountingKey("dl")
        .build();
}
//注意，这个队列设定了死信交换机为dl.ttl.direct，声明交换机，将ttl与交换机绑定：@BeanpublicDirectExchangettlExchange(){
returnnewDirectExchange("ttl.direct");
}
@BeanpublicBindingttlBinding(){
returnBindingBuilder.bind(ttlQueue()).to(ttlExchange()).with("ttl");
}

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

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

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

发送消息时，设定TTL:

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

查看发送消息日志：

接收消息日志：

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

总结：

消息超时的两种方式是？

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

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

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

延迟队列：

消息超时的两种方式是？

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

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

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

安装DelayExchange插件：

参考：部署

使用DelayExchange：

插件的使用也非常简单：声明一个交换机，交换机的类型可以是任意类型，只需要设定delayed属性为true即可，然后声明队列与其绑定即可。

声明DelayExchange交换机：

基于注解方式（推荐）：

也可以基于@Bean的方式：

发送消息

总结：

延迟队列插件的使用步骤包括哪些？

•声明一个交换机，添加delayed属性为true

•发送消息时，添加x-delay头，值为超时时间

惰性队列：

消息堆积问题：

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

解决消息堆积有两种思路：

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

要提升队列容积，把消息保存在内存中显然是不行的。

惰性队列：

从RabbitMQ的3.6.0版本开始，就增加了Lazy Queues的概念，也就是惰性队列。惰性队列的特征如下：

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

基于命令行设置lazy-queue：

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：策略的作用对象，是所有的队列

基于@Bean声明lazy-queue：

基于@RabbitListener声明LazyQueue：

总结：

消息堆积问题的解决方案？

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

惰性队列的优点有哪些？

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

惰性队列的缺点有哪些？

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

MQ集群：

普通集群：

是一种分布式集群，将队列分散到集群的各个节点，从而提高整个集群的并发能力。

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

部署：部署

镜像集群：

是一种主从集群，普通集群的基础上，添加了主从备份功能，提高集群的数据可用性。

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

部署：部署

仲裁队列

仲裁队列：仲裁队列是3.8版本以后才有的新功能，用来替代镜像队列，具备下列特征：

与镜像队列一样，都是主从模式，支持主从数据同步
使用非常简单，没有复杂的配置
主从同步基于Raft协议，强一致

@BeanpublicQueuequorumQueue() {
returnQueueBuilder        .durable("quorum.queue") // 持久化        .quorum() // 仲裁队列        .build();
}

SpringAMQP连接MQ集群：

spring:
  rabbitmq:
    addresses: 192.168.***.105:8071, 192.168.***.105:8072, 192.168.***.105:8073
    username: test
    password: ***
    virtual-host: /

上述整理知识点仅供个人学习，不作为其他用途；欢迎大家补充

关于RabbitMQ

首先MQ是什么？

MQ的好处：

同步和异步的区别：

RabbitMQ的安装：

publisher生成者实现：

consumer实现：

SpringAMQP

队列：

交换机：

消息转换器：

MQ生产环境遇到的问题

消息可靠性：

生成者确认机制：

消息持久化：

交换机持久化：

队列持久化：

消息持久化：

消费者消息确认：

演示auto模式：

消费失败重试机制：

本地重试：

失败策略：

总结：

死信交换机：

什么是死信交换机：

TTL：

总结：

延迟队列：

使用DelayExchange：

总结：

惰性队列：

消息堆积问题：

惰性队列：

总结：

MQ集群：

普通集群：

镜像集群：

仲裁队列

SpringAMQP连接MQ集群：

热门文章

最新文章

相关课程

相关电子书

相关实验场景