RabbitMQ消息中间件

一、初始MQ

首先了解一下微服务间通讯有同步和异步两种方式:

- 同步通讯:是指两个或多个系统在进行信息交换时,必须在同一时刻进行操作
- 异步通讯:是指两个或多个系统之间的通讯方式,其中发送方和接收方不是在同一时刻进行操作。

同步调用的优点:

- 时效性较强,可以立即得到结果

同步调用的缺点:

- 多个系统间耦合,扩展及后续维护繁琐
- 性能和吞吐能力下降
- 有额外的资源消耗
- 有级联失败风险

异步通讯:
优势一:服务解耦
优势二:性能提升,吞吐量提高
优势三:服务没有强依赖,不担心级联失败问题
优势四:流量削峰

缺点:

- 架构复杂了,业务没有明显的流程线,不好管理(对程序员的技术要求高了)
- 需要依赖于Broker的可靠、安全、性能(搭建集群)

1. 技术对比

MQ,中文是消息队列(Message Queue),字面来看就是存放消息的队列。
比较常见的MQ实现(也被称为消息中间件):

- ActiveMQ
- **RabbitMQ**
- **RocketMQ**
- Kafka

几种常见MQ的对比:
对比RabbitMQActiveMQRocketMQKafka
公司/社区RabbitApache阿里Apache
开发语言ErlangJavaJavaScala&Java
协议支持AMQP,XMPP SMTP,STOMPOpenWire,STOMP REST,XMPP,AMQP自定义协议自定义协议
可用性一般
单机吞吐量一般非常高
消息延迟微秒级毫秒级毫秒级毫秒以内
消息可靠性一般一般
选择原则:

- 追求可用性:Kafka、 RocketMQ 、RabbitMQ
- 追求可靠性:RabbitMQ、RocketMQ
- 追求吞吐能力:RocketMQ、Kafka
- 追求消息低延迟:RabbitMQ、Kafka

2. 原生JavaAPI实现MQ

在这之前,先认识RabbitMQ中的一些角色:

- publisher:生产者(使用Java代码发送消息)
- consumer:消费者(使用Java代码接收消息)
- exchange:交换机,负责消息路由
- queue:队列,存储消息
- virtualHost:虚拟主机,隔离不同租户的exchange、queue、消息
2.1 MQ的消息模型
- 简单队列
- 工作队列模式
- 发布订阅模式
- Fanout广播
- Direct定向模式
- Topic主题
- 消息转换器

下面使用原生API只展示简单队列模式
2.2 原生JavaAPI实现简单队列
简单队列模式的模型图:
Publisher
Queue
Consumer
- publisher:消息发布者,将消息发送到队列queue
- queue:消息队列,负责接收并缓存消息
- consumer:订阅队列,处理队列中的消息
下面使用的是官方提供的原生JavaAPI完成的;不用自己手敲代码练习,下面有利用Spring简化开发的方案
//生产端publisher实现
public class PublisherTest {
    @Test
    public void testSendMessage() throws IOException, TimeoutException {
        // 1.建立连接
        ConnectionFactory factory = new ConnectionFactory();
        // 1.1.设置连接参数,分别是:主机名、端口号、vhost、用户名、密码
        factory.setHost("192.168.200.130");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("用户名");//设置自己的用户名和密码
        factory.setPassword("*****");
        // 1.2.建立连接
        Connection connection = factory.newConnection();

        // 2.创建通道Channel
        Channel channel = connection.createChannel();

        // 3.创建队列
        String queueName = "simple.queue";
        channel.queueDeclare(queueName, false, false, false, null);

        // 4.发送消息
        String message = "hello, rabbitmq!";
        channel.basicPublish("", queueName, null, message.getBytes());
        System.out.println("发送消息成功:【" + message + "】");

        // 5.关闭通道和连接
        channel.close();
        connection.close();

    }
}
/***********************************************************************************************/
//消费端consumer实现

public class ConsumerTest {

    public static void main(String[] args) throws IOException, TimeoutException {
        // 1.建立连接
        ConnectionFactory factory = new ConnectionFactory();
        // 1.1.设置连接参数,分别是:主机名、端口号、vhost、用户名、密码
        factory.setHost("192.168.200.130");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("用户");  //用户密码和上面的生产端保持一致
        factory.setPassword("*****");
        // 1.2.建立连接
        Connection connection = factory.newConnection();

        // 2.创建通道Channel
        Channel channel = connection.createChannel();

        // 3.创建队列
        String queueName = "simple.queue";
        channel.queueDeclare(queueName, false, false, false, null);

        // 4.订阅消息
        channel.basicConsume(queueName, true, new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, 
                                       Envelope envelope,
                                       AMQP.BasicProperties properties, 
                                       byte[] body) throws IOException {
                // 5.处理消息
                String message = new String(body);
                System.out.println("接收到消息:【" + message + "】");
            }
        });
        System.out.println("等待接收消息。。。。");
    }
}

小结:

基本消息队列的消息发送流程:

1. 建立connection

2. 创建channel

3. 利用channel声明队列

4. 利用channel向队列发送消息

基本消息队列的消息接收流程:

1. 建立connection

2. 创建channel

3. 利用channel声明队列

4. 定义consumer的消费行为handleDelivery()

5. 利用channel将消费者与队列绑定

3.基于SpringAMQP实现MQ

SpringAMQP是基于RabbitMQ封装的一套模板,并且利用SpringBoot对其实现了自动装配,使用起来非常方便。

SpringAmqp的官方地址https://spring.io/projects/spring-amqp

SpringAMQP提供了三个功能:

- 自动声明队列、交换机及其绑定关系(代码+注解)
- 封装了RabbitTemplate工具,用于发送消息 :rabbitTemplate.convertAndSend()
- 基于注解的监听器模式,异步接收消息:@RabbitListener
Publisher
Queue
Consumer
在父工程中引入依赖
<!--AMQP依赖,包含RabbitMQ-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
1.简单队列
消息发送: 

首先配置MQ地址,在publisher服务的application.yml中添加配置:
spring:
  rabbitmq:
    host: 192.168.200.130 # 主机名
    port: 5672 # 端口
    virtual-host: / # 虚拟主机
    username: 用户名 # 自己的用户名(不能为中文)和密码
    password: *****
    在publisher服务中编写测试类SpringAmqpTest,并利用RabbitTemplate实现消息发送。
    代码实现如下:
@SpringBootTest
public class SpringAmqpTest {

    @Autowired
    private RabbitTemplate rabbitTempslate;

    @Test //不要导错包,用比较长的import org.junit.jupiter.api.Test;
    public void testSimpleQueue() {
        // 队列名称
        String queueName = "simple.queue";
        // 消息
        String message = "hello, spring amqp!";
        // 发送消息:此处并不会自动创建队列
        rabbitTemplate.convertAndSend(queueName, message);
    }
}
消息接收:
    首先配置MQ地址,在consumer服务的application.yml中添加配置:
    
spring:
  rabbitmq:
    host: 192.168.200.130 # 主机名
    port: 5672 # 端口
    virtual-host: / # 虚拟主机
    username: itcast # 用户名
    password: 123321 # 密码    
在consumer服务的中新建一个类SpringRabbitListener
@Component
public class SpringRabbitListener {

    @RabbitListener(queues = "simple.queue")
    public void listenSimpleQueueMessage(String msg) {
        System.out.println("spring 消费者接收到消息:【" + msg + "】");
    }
}
2.工作队列(Work queues)
    当消息处理比较耗时的时候,可能生产消息的速度会远远大于消息的消费速度。长此以往,
    消息就会堆积越来越多,无法及时处理。如何解决呢?
    
  - 那我们可以让多个消费者绑定到一个队列,共同消费队列中的消息。
 这个就称为Work queues,也被称为(Task queues),任务模型。可以使用work 模型,
 多个消费者共同处理消息处理,速度就能大大提高了。
Publisher
Queue
Consumer 1
Consumer 2
消息发送:

    在publisher服务中的SpringAmqpTest类中添加一个测试方法:
/**
     * workQueue
     * 向队列中不停发送消息,模拟消息堆积。
     */
@Test
public void testWorkQueue() throws InterruptedException {
    // 队列名称
    String queueName = "simple.queue";
    // 消息
    String message = "hello, message_";
    for (int i = 1; i <= 50; i++) {
        // 发送消息
        rabbitTemplate.convertAndSend(queueName, message + i);
        Thread.sleep(20);
    }
}
消息接收:
     要模拟多个消费者绑定同一个队列,我们在consumer中添加2个新的方法:
//@RabbitListener(queues = "simple.queue")
//public void listenSimpleQueueMessage(String msg) {
//    System.out.println(msg);
//}

@RabbitListener(queues = "simple.queue")
public void listenWorkQueue1(String msg) throws InterruptedException {
    System.out.println(LocalTime.now() + "消费者1:" + msg);
    Thread.sleep(20);
}

@RabbitListener(queues = "simple.queue")
public void listenWorkQueue2(String msg) throws InterruptedException {
    System.err.println(LocalTime.now() + "消费者2:" + msg);
    Thread.sleep(200);
}
运算之后得到结果:消息是平均分配给每个消费者,并没有考虑到消费者的处理能力。这样显然是有问题的。
怎样解决这个问题呢?

我们可以修改consumer服务的application.yml文件,添加配置:

spring:
  rabbitmq:
    host: 192.168.200.130 # 主机名
    port: 5672 # 端口
    virtual-host: / # 虚拟主机
    username: 用户名 # 自己的用户名和密码
    password: **** 
    listener: #监听
      simple: #简单消息模型
        prefetch: 1  #每次只能获取一条消息,处理完成才能获取下一个消息
        
 Work模型的使用:

- 多个消费者绑定到一个队列,同一条消息只会被一个消费者处理
- 通过设置prefetch来控制消费者预取的消息数量
3.发布/订阅
发布订阅的模型如图:
Publisher
exchange
Queue1
Consumer 1
Consumer 2
Queue2
Consumer 3
在订阅模型中,多了一个exchange角色,而且过程略有变化:

- Publisher:生产者,也就是要发送消息的程序,但是不再发送到队列中,而是发给exchage(交换机)

- Consumer:消费者,与以前一样,订阅队列,没有变化

- Queue:消息队列也与以前一样,接收消息、缓存消息。

- Exchange:交换机(消息路由)。一方面,接收生产者发送的消息。另一方面,知道如何处理消息,
- 例如递交给某个特别队列、递交给所有队列、或将消息丢弃。到底如何操作,取决于Exchange的类型。

  Exchange有以下3种类型:

  - Fanout(扇出):广播,将消息交给所有绑定到交换机的队列
  - Direct:定向,把消息交给符合指定routing key 的队列
  - Topic:通配符,把消息交给符合routing pattern(路由模式) 的队列

Exchange(交换机)只负责转发消息,不具备存储消息的能力,因此如果没有任何队列与Exchange绑定,
或者没有符合路由规则的队列,那么消息会丢失!
4.Fanout广播
Fanout,英文翻译是扇出,在MQ中理解成广播更合适。
Publisher
exchange
Queue1
Consumer 1
Queue2
Consumer 2
在广播模式下,消息发送流程是这样的:

- 1)  可以有多个队列
- 2)  每个队列都要绑定到Exchange(交换机)
- 3)  生产者发送的消息,只能发送到交换机,交换机来决定要发给哪个队列,生产者无法决定
- 4)  交换机把消息发送给绑定过的所有队列
- 5)  订阅队列的消费者都能拿到消息

声明队列和交换机

Spring提供了一个接口Exchange,来表示所有不同类型的交换机:UML类图

在这里插入图片描述

在consumer服务中创建一个类,声明队列和交换机:
    
@Configuration
public class FanoutConfig {
    /**
     * 声明交换机
     * @return Fanout类型交换机
     */
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("itcast.fanout");
    }

    /**
     * 第1个队列
     */
    @Bean
    public Queue fanoutQueue1(){
        return new Queue("fanout.queue1");
    }

    /**
     * 绑定队列和交换机
     */
    @Bean // 方法中的参数,从IoC容器中获取
    public Binding bindingQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange){
        return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
    }

    /**
     * 第2个队列
     */
    @Bean
    public Queue fanoutQueue2(){
        return new Queue("fanout.queue2");
    }

    /**
     * 绑定队列和交换机
     */
    @Bean
    public Binding bindingQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange){
        return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
    }
} 
消息发送:
在publisher服务的SpringAmqpTest类中添加测试方法:
@Test
public void testFanoutExchange() {
    // 交换机名称
    String exchangeName = "itcast.fanout";
    // 消息
    String message = "hello, everyone!";
    rabbitTemplate.convertAndSend(exchangeName, "", message);
}
消息接收
在consumer服务的SpringRabbitListener中添加两个方法,作为消费者:

@RabbitListener(queues = "fanout.queue1")
public void listenFanoutQueue1(String msg) {
    System.out.println("消费者1接收到Fanout消息:【" + msg + "】");
}

@RabbitListener(queues = "fanout.queue2")
public void listenFanoutQueue2(String msg) {
    System.out.println("消费者2接收到Fanout消息:【" + msg + "】");
}
交换机的作用是什么?

- 接收publisher发送的消息
- 将消息按照规则路由到与之绑定的队列
- 不能缓存消息,路由失败,消息丢失
- FanoutExchange的会将消息路由到每个绑定的队列

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

- Queue
- FanoutExchange
- Binding
5.Direct定向
在Fanout模式中,一条消息,会被所有订阅的队列都消费。
但是,在某些场景下,我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。
key:blue
key:red
Publisher
exchange
Queue
Consumer 1
Queue2
Consumer 2
 在Direct模型下:

- 队列与交换机的绑定,不能是任意绑定了,而是要指定一个`RoutingKey`(路由key)。
- 消息的发送方在 向 Exchange发送消息时,也必须指定消息的 `RoutingKey`。
- Exchange不再把消息交给每一个绑定的队列,而是根据消息的`Routing Key`进行判断,
只有队列的`Routingkey`与消息的 `Routing key`完全一致,才会接收到消息。
案例需求如下:

1. 利用@RabbitListener声明Exchange、Queue、RoutingKey

2. 在consumer服务中,编写两个消费者方法,分别监听direct.queue1和direct.queue2

3. 在publisher中编写测试方法,向itcast. direct发送消息
声明队列和交换机
    基于@Bean的方式声明队列和交换机比较麻烦,Spring还提供了基于注解方式来声明。
    
    在consumer的SpringRabbitListener中添加两个消费者,同时基于注解来声明队列和交换机:
@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "direct.queue1"), //创建队列
    exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),//创建交换机
    key = {"red", "blue"} //绑定接受消息的key
))
public void listenDirectQueue1(String msg){
    System.out.println("消费者接收到direct.queue1的消息:【" + msg + "】");
}

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "direct.queue2"),
    exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),
    key = {"red", "yellow"}
))
public void listenDirectQueue2(String msg){
    System.out.println("消费者接收到direct.queue2的消息:【" + msg + "】");
}
消息发送

在publisher服务的SpringAmqpTest类中添加测试方法

@Test
public void testSendDirectExchange() {
    // 交换机名称
    String exchangeName = "itcast.direct";
    // 消息
    String message = "红色警报!日本乱排核废水,导致海洋生物变异,惊现哥斯拉!";
    // 发送消息:key=red,两个消费者都能收到消息
    rabbitTemplate.convertAndSend(exchangeName, "red", message);
    
    // 发送消息:key=blue,消费者1 能收到消息
    rabbitTemplate.convertAndSend(exchangeName, "blue", message);
    
    // 发送消息:key=yellow,消费者2 能收到消息
    rabbitTemplate.convertAndSend(exchangeName, "yellow", message);
}
总结:
Direct交换机与Fanout交换机的差异

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

基于@RabbitListener注解声明队列和交换机的常见注解

- 开始声明:bindings= ? 
- 指定一个绑定关系: @QueueBinding
- 声明队列:value = @Queue
- 声明交换机:exchange = @Exchange
- 指定路由key:key = {一个或多个}
6.Topic主题
`Topic`类型的`Exchange`与`Direct`相比:

- 相同点:都可以根据`RoutingKey`把消息路由到不同的队列
- 不同点:`Topic`类型`Exchange`可以让队列在绑定`Routing key` 的时候使用通配符!

> `Routingkey` 一般都是有一个或多个单词组成,多个单词之间以”.”分割,
例如: `item.insert`, item.del

> 通配符规则:
>
> `#`:匹配零个,一个或多个词,任意多个【常用】
>
> `*`:匹配不多不少必须是1个词
topic
bindingKey1
bindingKey2
bindingKey3
bindingKey4
Publisher
exchange
Queue1
Consumer 1
Queue2
Consumer 2
Queue3
Consumer 3
Queue4
Consumer 4
举例:
demo.#:能够匹配demo, demo.spu, demo.spu.insert
demo.*:只能匹配demo.spu
实现思路如下:

1. 并利用@RabbitListener声明Exchange、Queue、RoutingKey

2. 在consumer服务中,编写两个消费者方法,分别监听topic.queue1和topic.queue2

3. 在publisher中编写测试方法,向itcast. topic发送消息

- Queue1:假设绑定的是`china.#` ,因此凡是以 `china.`开头的`routing key` 都会被匹配到。
- 包括china.news和china.weather
- Queue2:假设绑定的是`#.news` ,因此凡是以 `.news`结尾的 `routing key` 都会被匹配。
- 包括china.news和japan.news
消息接收
    在consumer服务的SpringRabbitListener中添加方法:

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "demo.queue1"),
    exchange = @Exchange(name = "demo.topic", type = ExchangeTypes.TOPIC),
    key = "china.#"
))
public void listenTopicQueue1(String msg){
    System.out.println("消费者接收到topic.queue1的消息:【" + msg + "】");
}

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "demo.queue2"),
    exchange = @Exchange(name = "demo.topic", type = ExchangeTypes.TOPIC),
    key = "#.news"
))
public void listenTopicQueue2(String msg){
    System.out.println("消费者接收到demo.queue2的消息:【" + msg + "】");
}
消息发送
    在publisher服务的SpringAmqpTest类中添加测试方法:
/**
     * topicExchange
     */
@Test
public void testSendTopicExchange() {
    // 交换机名称
    String exchangeName = "demo.topic";
    // 消息
    String message = "喜报!孙悟空大战哥斯拉,胜!";
    // 发送消息
    rabbitTemplate.convertAndSend(exchangeName, "china.news", message);
    
     // 发送消息
    rabbitTemplate.convertAndSend(exchangeName, "china.weather", "明天天气晴,20-36度");
}
# 总结

描述下Direct交换机与Topic交换机的差异?

- Topic交换机接收的消息RoutingKey必须是多个单词,以 `**.**` 分割
- Topic交换机与队列绑定时的bindingKey可以指定通配符
  - `#`:代表0个,1个或多个词
  - `*`:代表1个词
7.消息转换器
Spring会把你发送的消息序列化为字节发送给MQ,接收消息的时候,还会把字节反序列化为下面的message对象。
void convertAndSend(String exchange, String routingKey,Object message) throw AmqpException;
默认情况下Spring采用的序列化方式是JDK序列化。众所周知,JDK序列化存在下列问题:

- 数据体积过大
- 可读性差
测试默认转换器:

@Test
public void testSendMap() throws InterruptedException {
    // 准备消息
    Map<String,Object> msg = new HashMap<>();
    msg.put("name", "Jack");
    msg.put("age", 21);
    // 发送消息
    rabbitTemplate.convertAndSend("simple.queue", msg);
}
1、执行前:先停止consumer服务,防止消息被消费掉,无法在RabbitMQ控制台看到
2、MQ服务上没有simple.queue(临时),通过管理端快速创建一个
发送消息后查看控制台:

在这里插入图片描述

# 配置JSON转换器

   显然,JDK序列化方式并不合适。我们希望消息体的体积更小、可读性更高,因此可以使用JSON方式来做序列化
   和反序列化。
   
1、在publisher和consumer两个服务中都引入依赖,因此咱们选择在父工程添加:
<!-- mq-demo的pom.xml,jackson(SpringBoot用的) -->
<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
</dependency>
2、配置消息转换器

#在PublisherApplication和ConsumerApplication两个启动类中都添加一个Bean
PublisherApplication,作用:Java对象 =====》JSON字符串

import org.springframework.amqp.support.converter.MessageConverter;

@SpringBootApplication
public class PublisherApplication {
    public static void main(String[] args) {
        SpringApplication.run(PublisherApplication.class);
    }

    @Bean //注意导包:org.springframework.amqp.support.converter.MessageConverter
    public MessageConverter messageConverter(){
        return new Jackson2JsonMessageConverter();
    }
}
3、重新发送消息:通过管理平台查询效果

在这里插入图片描述

4、接受消息:SpringRabbitListener
@RabbitListener(queues = "simple.queue")
public void listenObjectQueue(Map<String,Object> msg){
    System.out.println("接收到object.queue的消息:" + msg);
}

二、MQ高级

1.消息可靠性

消息从发送到消费者接收,会经历多个过程:
Publisher
exchange
Queue1
Consumer 1
Queue2
Consumer 2
其中的每一步都可能导致消息丢失,常见的丢失原因包括:

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

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

- 生产者确认机制(发送时丢失)
- 消息持久化(MQ宕机)
- 消费者确认机制(消费者宕机)
- 失败重试机制(消费失败)
1.1.生产者消息确认
    RabbitMQ提供了生产者确认机制来避免消息发送到MQ过程中丢失。这种机制必须给每个消息指定一个唯一ID。
    消息发送到MQ以后,会返回一个结果给发送者,表示消息是否处理成功。
    
返回结果有两种方式:

- publisher-confirm,发送者确认
  - 消息成功投递到交换机,返回ack
  - 消息未投递到交换机,返回nack
  
- publisher-return,发送者回执
  - 消息投递到交换机了,但是没有路由到队列。返回通知,及路由失败原因。
  - 正常到达队列,没有任何回复(没有回复就是成功)
确认机制发送消息时,需要给每个消息设置一个全局唯一Id,以区分不同消息,避免ack冲突。
举个栗子:

在这里插入图片描述

修改publisher服务中的application.yml文件,添加下面的内容:

spring:
  rabbitmq:
    host: 192.168.200.130 # rabbitMQ的ip地址
    port: 5672 # 端口
    username: itcast
    password: 123321
    virtual-host: /
    publisher-confirm-type: correlated #判断是否到达交换机(异步通知)
    publisher-returns: true #判断是否到达队列
    template:
      mandatory: true #定义消息路由失败时的策略
解释说明一下:

- `publish-confirm-type`:开启publisher-confirm,这里支持两种类型:
  - `simple`:同步等待confirm结果,直到超时【一般不使用,影响性能】
  - `correlated`:异步回调,定义ConfirmCallback,MQ返回结果时会回调这个ConfirmCallback
  
- `publish-returns`:
  - `true`:开启publish-return功能,同样是基于callback机制,不过是定义ReturnCallback
  
  - `false`:关闭publish-return功能
- `template.mandatory`:定义消息路由失败时的策略。
  - true则调用ReturnCallback
  - false则直接丢弃消息
修改consumer服务中的application.yml:改为自己的虚拟机IP

spring:
  rabbitmq:
    host: 192.168.200.130 # rabbitMQ的ip地址
    port: 5672 # 端口
    username: itcast
    password: 123321
    virtual-host: /
定义Return回调:
    每个RabbitTemplate只能配置一个ReturnCallback,因此需要在项目启动过程中配置
作用: 如果消息没有到达队列,会执行回调方法

修改publisher服务,添加一个ReturnCallback:
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
//ApplicationContextAware: 在Spring容器(Bean工厂)创建好的时候,通知咱们
public class CommonConfig implements ApplicationContextAware {
    
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        // 获取RabbitTemplate
        RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);
        // 设置ReturnCallback,先用匿名内部类
        rabbitTemplate.setReturnCallback(
            (message, replyCode, replyText, exchange, routingKey) -> {
            // 投递失败(没有到达队列),记录日志
            log.error("消息队列接收失败,应答码{},原因{},交换机{},路由键{},消息{}",
                     replyCode, replyText, exchange, routingKey, message.toString());
            // 如果有业务需要,可以重发消息
            //rabbitTemplate.convertAndSend(exchange, routingKey, message);
        });
    }
}
定义Confirm回调:

    ConfirmCallback可以在发送消息时指定,因为每个业务处理confirm成功或失败的逻辑不一定相同。
    在publisher服务的cn.itcast.mq.spring.SpringAmqpTest类中,定义一个单元测试方法:
@Test
public void testSendMessage2SimpleQueue() throws InterruptedException {
    // 1.消息体
    String message = "hello, spring amqp!";
    // 2.全局唯一的消息ID,需要封装到CorrelationData中
    //uuid, 雪花算法
    CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
    // 3.添加callback
    correlationData.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.发送消息,其中"simple.test"是路由key
    rabbitTemplate.convertAndSend(
            "amq.topic", "simple.test", message, correlationData);

    // 休眠一会儿,等待ack回执
    //如果不休眠,程序就直接结束了;RabbitMQ服务器就无法回调咱们写的代码
    Thread.sleep(2000);
}
登录到MQ的管理端:
# 测试:

1、发送到一个不存在的交换机:camq.topic
rabbitTemplate.convertAndSend(
      "camq.topic", "simple.test", message, correlationData);

//查看日志:会有一个没有到达交换机的信息
2、发送到一个已经存在的交换机:amq.topic(系统自带的),但没有绑定指定的路由
rabbitTemplate.convertAndSend(
      "amq.topic", "simple.test", message, correlationData);

//查看日志:没有路由到队列
3、通过管理端指定amq.topic交换机的路由key到simple.queue

在这里插入图片描述

rabbitTemplate.convertAndSend(
      "amq.topic", "simple.test", message, correlationData);

//成功发送,需要到管理端查看一下队列中是否有消息
1.2.消息持久化
生产者确认可以确保消息投递到RabbitMQ的队列中,但是消息发送到RabbitMQ以后,如果突然宕机,
也可能导致消息丢失。

要想确保消息在RabbitMQ中安全保存,必须开启消息持久化机制。

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

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

SpringAMQP中可以通过代码指定交换机持久化:
@Bean
public DirectExchange simpleExchange(){
    // 三个参数:交换机名称、是否持久化、当没有queue与其绑定时是否自动删除
    //durable: 持久化
    return new DirectExchange("simple.direct", true, false);
    
    //默认创建就是持久化的交换机
    //return new DirectExchange("simple.direct");
}
提示:由SpringAMQP声明的交换机都是持久化的
可以在RabbitMQ控制台看到持久化的交换机都会带上`D`的标示:

在这里插入图片描述

# 队列持久化

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

SpringAMQP中可以通过代码指定交换机持久化:
@Bean
public Queue simpleQueue(){
    // 使用QueueBuilder构建队列,durable就是持久化的
    return QueueBuilder.durable("simple.queue").build();
    
    //return new Queue("simple.queue");
}
提示:由SpringAMQP声明的交换机都是持久化的

在这里插入图片描述

# 消息持久化

利用SpringAMQP发送消息时,可以设置消息的属性(MessageProperties),指定delivery-mode:

- 非持久化:MessageDeliveryMode.NON_PERSISTENT
- 持久化:MessageDeliveryMode.PERSISTENT

用java代码指定:
@Test
public void testDurableMessage() {
    // 1.准备消息
    Message message = MessageBuilder
            .withBody("hello, spring".getBytes(StandardCharsets.UTF_8))
            .setDeliveryMode(MessageDeliveryMode.PERSISTENT) //设置消息的属性:持久化
            .build();
    // 2.发送消息
    rabbitTemplate.convertAndSend("simple.queue", message);
}
提示:由SpringAMQP声明的交换机都是持久化的
1.3.消费者消息确认
RabbitMQ确认消息被消费者消费后会立刻删除。

而RabbitMQ是通过消费者回执来确认消费者是否成功处理消息的:消费者获取消息后,应该向RabbitMQ发送ACK回
执,表明自己已经处理消息。

设想这样的场景:

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

这样,消息就丢失了。因此消费者返回ACK的时机非常重要。

/********************************************************************************

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

- manual:手动ack,需要在处理完消息后,调用api发送ack【麻烦,一般不使用】。
- auto:自动ack,由spring监测listener代码是否出现异常,没有异常则返回ack;抛出异常则返回nack
- none:关闭ack,MQ假定消费者获取消息后肯定会成功处理,因此消息投递后立即被删除



由此可知:

- manual:自己根据业务情况,判断什么时候该ack(太麻烦,不使用)
- auto模式类似事务机制,出现异常时返回nack,消息回滚到mq;没有异常,返回ack
- none模式下,消息投递是不可靠的,可能丢失(不适合用在项目中)

因此,我们都是使用默认的auto即可。
# none模式
修改consumer服务的application.yml文件,添加下面内容:

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: none # 关闭ack
修改consumer服务的SpringRabbitListener类中的方法,模拟一个消息处理异常:
@RabbitListener(queues = "simple.queue")
public void listenSimpleQueue(String msg) {
    log.info("消费者接收到simple.queue的消息:【{}】", msg);
    // 模拟异常 ->给MQ返回nack
    System.out.println(1 / 0);
    log.debug("消息成功处理完成!");
}

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

spring:
  rabbitmq:
    listener:
      simple:
        #消费成功,返回ack
        #消费失败,返回nack
        acknowledge-mode: auto # 根据异常自动ack
在异常位置打断点,再次发送消息,程序卡在断点时,可以发现此时消息状态为unack(未确定状态):

在这里插入图片描述

抛出异常后,因为Spring会自动返回nack,所以消息恢复至Ready状态,并且没有被RabbitMQ删除:

在这里插入图片描述

1.4.消费失败重试机制
当消费者出现异常后,消息会不断requeue(重入队)到队列,再重新发送给消费者,然后再次异常,再次requeue,
无限循环,导致mq的消息处理飙升,带来不必要的压力:

在这里插入图片描述

怎么办呢?
# 本地重试

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

修改consumer服务的application.yml文件,添加内容:
spring:
  rabbitmq:
    listener:
      simple:
        retry: #本地重试
          enabled: true # 开启消费者失败重试
          initial-interval: 1000 # 初始的失败等待时长为1秒
          multiplier: 2 # 失败的等待时长倍数,下次等待时长 = multiplier * last-interval
          max-attempts: 3 # 最大重试次数,包含服务器推送的第一次
          stateless: true # true无状态;false有状态。如果业务中包含事务,这里改为false
重启consumer服务,重复之前的测试。可以发现:

- 在重试3次后,SpringAMQP会抛出异常AmqpRejectAndDontRequeueException,说明本地重试触发了

- 查看RabbitMQ控制台,发现消息被删除了(RejectAndDontRequeue),说明最后SpringAMQP返回的是ack,
mq删除消息了

reject: 拒绝
don't re queue: 不要重新放到队列


# 结论:

- 开启本地重试时,消息处理过程中抛出异常,不会requeue到队列,而是在消费者本地重试
- 重试达到最大次数后,Spring会返回ack给MQ服务器(reject+ not re queue),消息会被丢弃
失败策略:

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

在开启重试模式后,重试次数耗尽,如果消息依然失败,会有MessageRecoverer接口来处理,
它包含三种不同的实现:
- RejectAndDontRequeueRecoverer:重试耗尽后,直接reject,丢弃消息。默认就是这种方式

- ImmediateRequeueMessageRecoverer:本地重试耗尽后,返回nack,消息重新入队,重新推送消息

- RepublishMessageRecoverer【最优方法】:重试耗尽后,将失败消息投递到指定的交换机(后续人工介入来处理)


处理方案是RepublishMessageRecoverer,失败后将消息投递到一个指定的,专门存放异常消息的队列,
后续由人工集中处理。
1)在consumer服务中定义处理失败消息的交换机和队列
    
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.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class ErrorMessageConfig {
    
    @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");
    }
    
    //TODO 指定失败处理策略
}
/**************************************************************************************************/
2)定义一个RepublishMessageRecoverer,关联队列和交换机
    
@Bean //非常特殊,方法上有@Bean,方法中所有的参数自动就有一个@Autowired
public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){
    //最终效果:将重试失败的消息重新发送到指定的交换机+路由key
    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 {
    @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");
    }

    @Bean //修改本地重试耗尽之后,消息处理策略:把消息发到指定的交换+key
    public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){
        return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");
    }
}
1.5.总结
如何确保RabbitMQ消息的可靠性?

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

2.死信交换机

2.1.认识死信
当一个队列中的消息满足下列情况之一时,可以成为死信(dead letter):

- 消费者使用basic.reject或 basic.nack声明消费失败,并且消息的requeue参数设置为false
- 消息是一个过期消息,超时无人消费【利用此机制,实现延迟消息】
- 要投递的队列消息满了,无法投递

如果一个消息被消费者拒绝了,变成了死信:

在这里插入图片描述

如果这个包含死信的队列配置了`dead-letter-exchange`属性,指定了一个交换机,那么队列中的死信就会投递到
这个交换机中,而这个交换机称为死信交换机(Dead Letter Exchange,简称DLX)。

如果这个死信交换机也绑定了一个队列,则消息最终会进入这个只存放死信的队列:

在这里插入图片描述

因为simple.queue绑定了死信交换机 dl.direct,并且设置了路由key,因此死信最终会经过死信交换机路由给死信队列。

- 指定死信交换机名称:dl.direct
- 指定死信交换机与死信队列绑定的RoutingKey:dl

这样才能确保投递的消息能到达死信交换机,并且正确的路由到死信队列。

> 下边代码只是为了演示,对应图片中的配置,不用添加到项目中
@Bean
public Queue simpleQueue(){
    //return new Queue("simple.queue");
    return QueueBuilder
        .durable("simple.queue") // 指定队列名称,并持久化
        .deadLetterExchange("dl.direct") // 指定死信交换机
        .deadLetterRoutingKey("dl") //指定路由key
        .build();
}
# 总结:

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

- 消息被消费者reject或者返回nack,并且设置了requeue=false
- 消息超时未消费
- 队列满了

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

- 如果队列绑定了死信交换机,死信会投递到死信交换机;
- 可以利用死信交换机收集所有消费者处理失败的消息(死信),交由人工处理,进一步提高消息队列的可靠性。
2.2.TTL(过期时间)
TTL,也就是Time-To-Live(过期时间)。如果一个队列中的消息TTL结束仍未消费,则会变为死信。

TTL超时分为两种情况:

- 消息本身设置了超时时间
- 消息所在的队列设置了超时时间

> 思考:为什么要给消息或者队列设置过期时间呢?

> 目的:实现延迟任务的功能

> 比如要实现如下功能:
>
> - 延迟10分钟发送短信给用户(ttl = 10分钟)
> - 用户下单,如果用户在15 分钟内未支付,则自动取消
> - 预约工作会议,20分钟后自动通知所有参会人员
# 创建死信交换机

在consumer服务的SpringRabbitListener中,定义一个新的消费者,并且声明死信交换机 dl.direct、
死信队列 dl.queue:
@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "dl.queue", durable = "true"),
    exchange = @Exchange(name = "dl.direct"),
    key = "dl"
))
public void listenDlQueue(String msg){
    log.info("接收到 dl.queue的延迟消息:{}", msg);
}
# 声明队列,指定超时时间

在consumer服务中新建TTLMessageConfig,创建ttl队列:

- 设置超时时间:ttl(10000)
- 指定死信交换机:deadLetterExchange("dl.direct")
- 指定死信的路由key:deadLetterRoutingKey("dl")
package cn.itcast.mq.config;

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class TTLMessageConfig {

    @Bean
    public Queue ttlQueue(){
        return QueueBuilder
                .durable("ttl.queue") // 指定队列名称,并持久化
                .ttl(10000) // 设置队列的超时时间,10秒
                .deadLetterExchange("dl.direct") // 指定死信交换机
                .deadLetterRoutingKey("dl")
                .build();
    }
    
    /**
     * 声明交换机,将ttl队列与交换机绑定
     */
    @Bean
    public DirectExchange ttlExchange(){
        return new DirectExchange("ttl.direct");
    }
    
    @Bean
    public Binding ttlBinding(Queue ttlQueue, DirectExchange ttlExchange){
        return BindingBuilder.bind(ttlQueue).to(ttlExchange).with("ttl");
    }	
}
在publisher服务中发送消息:
    
@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("发送消息成功");
}
注意:先启动消费者,再发送消息

执行完之后观察时间戳,可以看到消息发送与接收之间的时差大概是10秒。
# 发送消息时,设定TTL

在发送消息时,也可以指定TTL:
@Test
public void testTTLMsg() {
    // 创建消息
    Message message = MessageBuilder
        .withBody("hello, ttl message".getBytes(StandardCharsets.UTF_8))
        //setex : set expire
        .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的插件列表页面:https://www.rabbitmq.com/community-plugins.html

在这里插入图片描述

使用方式可以参考官网地址:https://blog.rabbitmq.com/posts/2015/04/scheduling-messages-with-rabbitmq
# 使用DelayExchange

插件的使用也非常简单:

- 声明一个交换机,交换机的类型可以是任意类型
- 设定delayed属性为true
- 声明队列与其绑定

# 1)声明DelayExchange交换机

基于注解方式【常用】:

注意:如果MQ容器没有安装DelayExchange插件,直接指定delayed=true,启动项目时会报错
@RabbitListener(bindings = @QueueBinding(
          value = @Queue(name = "delay.queue",durable = "true"),
          exchange = @Exchange(name = "delay.direct",delayed = "true"),
           key="delay"
))
public void listenDelayedQueue(String msg){
    log.info("接受到 delay.queue的延迟消息: {}",msg);
}
在consumer服务的SpringRabbitListener中添加:

> 优势:代码简单

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "delay.queue", durable = "true"),
        exchange = @Exchange(name = "delay.direct", delayed = "true"),
        key = "delay"
))
public void listenDelayExchange(String msg) {
    log.info("消费者接收到了delay.queue的延迟消息{}", msg);
}
第二种方式:也可以基于@Bean的方式:

> 优势:清晰明了

在这里插入图片描述

# 2)发送消息

发送消息时,一定要携带x-delay属性,指定延迟的时间:
@Test
public void testDelayedMsg(){
    Message message = MessageBuilder
        .withBody("hello,delayed message",getBytes(StandardCharsets.UTF_8))
        .setHeader("x-delay",10000)
        .build();
    CorrelationData correlationData = new CorrelationData(UUID.random.UUID().toString());
    rabbitTemplate.convertAndSend("delay.direct","delay",message,correlationData);
    log.debug("发送消息成功");
}
@Test
public void testSendDelayMessage() throws InterruptedException {
    // 1.准备消息
    Message message = MessageBuilder
            .withBody("hello, delayed messsage".getBytes(StandardCharsets.UTF_8))
            .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
            .setHeader("x-delay", 10000) //时间必须是数字,不能是字符串
            .build();
    // 2.准备CorrelationData
    CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
    // 3.发送消息
    rabbitTemplate.convertAndSend("delay.direct", "delay", message, correlationData);

    log.info("发送消息成功");
}
此时,idea控制台会有一个报错信息:

在这里插入图片描述

原因很简单,在之前课程中我们添加了定义发送者Return回调,如果消息发送之后没有到达队列就会报错。

当使用插件发送消息时设置了x-delay=10000,那消息只要没有到过期时间,就不会路由到队列中,
而是存在一个叫Mnesia的分布式数据库管理系统中。

在这里插入图片描述

因此需要在publisher服务的CommonConfig中,判断是否为延迟消息:

在这里插入图片描述

// 判断是否是延迟消息
Integer receivedDelay = message.getMessageProperties().getReceivedDelay();
if (receivedDelay != null && receivedDelay > 0) {
    // 是一个延迟消息,忽略这个错误提示
    return;
}
# 总结

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

- 声明一个交换机,添加delayed属性为true
- 发送消息时,添加x-delay头,值为超时时间(必须是int值)

3.惰性队列

# 消息堆积问题

   当生产者发送消息的速度超过了消费者处理消息的速度,就会导致队列中的消息堆积,
   直到队列存储消息达到上限。之后发送的消息就会成为死信,可能会被丢弃,这就是消息堆积问题。
队列满丢弃
Publisher
Queue
Consumer
死信
解决消息堆积有三种思路:

- 增加更多消费者,提高消费速度
- 在消费者内开启线程池多线程处理,加快消息处理速度
- 惰性队列:扩大队列容积,提高堆积上限
从RabbitMQ的3.6.0版本开始,就增加了Lazy Queues的概念,也就是惰性队列。惰性队列的特征如下:

- 接收到消息后直接存入磁盘而非内存,缺点:速度会变慢
- 消费者要消费消息时才会从磁盘中读取并加载到内存,最终推送给消费者
- 支持数百万条的消息存储
3.1.基于命令行设置lazy-queue
注:(本操作是Linux操作系统进行的)

设置一个队列为惰性队列,只需要在声明队列时,指定x-queue-mode属性为lazy即可。

可以通过命令行将一个运行中的队列修改为惰性队列:


#进入MQ容器
docker exec -it mq bash

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  `:策略的作用对象,是所有的队列
3.2.@Bean声明lazy-queue

@Bean
public Queue lazyQueue(){
    return QueueBuilder
          .durable("lazy.queue")
          .lazy()//开启x-queue-mode为lazy
          .build();
}
package cn.itcast.mq.config;

import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


@Configuration
public class LazyConfig {

    @Bean
    public Queue lazyQueue() {
        return QueueBuilder
            	.durable("lazy.queue")
                .lazy() //指定是惰性队列
                .build();
    }
    
    @Bean
    public Queue normalQueue() {
        return QueueBuilder.durable("normal.queue").build();
    }
}
重启cousumer服务:确认已经创建了以上两个队列:

在这里插入图片描述

3.3.注解声明LazyQueue
此处没有给队列绑定交换机,因此使用的是queuesToDeclare = ?,而不是bindings = ?
@RabbitListener(queuesToDeclare = @Queue(
        name = "lazy.queue",
        durable = "true",
        arguments = @Argument(name = "x-queue-mode", value = "lazy")
))
public void listLazyQueue(String msg) {
    log.info("接收到 lazy.queue 的消息:{}", msg);
}
3.4.测试
先把cousumer服务停掉,不然发送的消息都被消费掉了,无法观察效果
1、在publisher服务的SpringAmqpTest中发送消息到惰性队列:
@Test
public void testLazyQueue() throws InterruptedException {
    long b = System.currentTimeMillis();
    for (int i = 0; i < 1000; i++) {
        // 1.准备消息
        Message message = MessageBuilder
                .withBody("hello, Spring".getBytes(StandardCharsets.UTF_8))
                .setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT)
                .build();
        // 2.发送消息
        rabbitTemplate.convertAndSend("lazy.queue", message);
    }
    long e = System.currentTimeMillis();
    System.out.println(e - b);
}
发现消息都在磁盘中:

在这里插入图片描述

2、也可以发送到普通队列,做为对比:
@Test
public void testNormalQueue() throws InterruptedException {
    long b = System.currentTimeMillis();
    for (int i = 0; i < 1000; i++) {
        // 1.准备消息
        Message message = MessageBuilder
                .withBody("hello, Spring".getBytes(StandardCharsets.UTF_8))
                .setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT)
                .build();
        // 2.发送消息
        rabbitTemplate.convertAndSend("normal.queue", message);
    }
    long e = System.currentTimeMillis();
    System.out.println(e - b);
}
发现消息都在内存中:

在这里插入图片描述

# 总结

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

- 队列上绑定多个消费者,提高消费速度
- 在消费者内开启线程池多线程处理,加快消息处理速度
- 使用惰性队列,可以再mq中保存更多消息

惰性队列的优点有哪些?

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

惰性队列的缺点有哪些?

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

4.MQ集群

4.1.集群分类
RabbitMQ的是基于Erlang语言编写,而Erlang又是一个面向并发的语言,天然支持集群模式。

RabbitMQ的集群有两种模式:

- 普通集群:是一种分布式集群,将队列分散到集群的各个节点,从而提高整个集群的并发能力。
- 镜像集群:是一种主从集群,在普通集群的基础上,添加了主从备份功能,提高集群的数据可用性。

镜像集群虽然支持主从,但主从同步并不是强一致的,某些情况下可能有数据丢失的风险。

因此在RabbitMQ的3.8版本以后,推出了新的功能:仲裁队列来代替镜像集群,底层采用Raft协议确保
主从的数据一致性。
4.2.普通集群
# 集群结构和特征

普通集群,或者叫标准集群(classic cluster),具备下列特征:

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

结构如图:

在这里插入图片描述

4.3.镜像集群
# 集群结构和特征

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

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

结构如图:

在这里插入图片描述

4.4.仲裁队列
# 集群特征

镜像集群虽然支持主从,但主从同步并不是强一致的,某些情况下可能有数据丢失的风险。

仲裁队列:仲裁队列是3.8版本以后才有的新功能,用来替代镜像队列,底层采用Raft协议确保主从的数据一致性,
具备下列特征:

- 与镜像队列一样,都是主从模式,支持主从数据同步
- 使用非常简单,没有复杂的配置
- 主从同步基于Raft协议,强一致
Java代码创建仲裁队列
    
@Bean
public Queue quorumQueue() {
    return QueueBuilder
        .durable("quorum.queue2") // 持久化
        //.layzy() //惰性队列
        .quorum() // 仲裁队列
        .build();
}
# SpringAMQP连接MQ集群

注意,这里用address来代替host、port方式
spring:
  rabbitmq:
	#host: 192.168.200.130
    #port: 5672
    addresses: 192.168.200.130:8071, 192.168.200.130:8072, 192.168.200.130:8073
    username: itcast
    password: 123321
    virtual-host: /
注意:因为重新创建的3MQ集群,还没有安装延迟队列插件,因此原来练习延迟队列的代码需要注释掉:
1、创建交换机时
    
@RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "delay.queue", durable = "true"),
        exchange = @Exchange(name = "delay.direct", delayed = "true"),
        key = "delay"
))
public void listenDelayExchange(String msg) {
    log.info("消费者接收到了delay.queue的延迟消息{}", msg);
}
/******************************************************************************************/
2、发送消息时
    
@Test
public void testSendDelayMessage() throws InterruptedException {
    // 1.准备消息
    Message message = MessageBuilder
            .withBody("hello, delayed messsage".getBytes(StandardCharsets.UTF_8))
            .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
            .setHeader("x-delay", 10000)
            .build();
    // 2.准备CorrelationData
    CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
    // 3.发送消息
    rabbitTemplate.convertAndSend("delay.direct", "delay", message, correlationData);

    log.info("发送消息成功");
}

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:/a/110107.html

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈qq邮箱809451989@qq.com,一经查实,立即删除!

相关文章

Hadoop学习总结(搭建Hadoop集群(伪分布式模式))

如果前面有搭建过Hadoop集群完全分布式模式&#xff0c;现在搭建Hadoop伪分布式模式可以选择直接克隆完全分布式模式中的主节点(hadoop001)。以下是在搭建过完全分布式模式下的Hadoop集群的情况进行 伪分布式模式下的Hadoop功能与完全分布式模式下的Hadoop功能相同。 一、克隆…

day55--动态规划13

300.最长递增子序列 674. 最长连续递增序列 718. 最长重复子数组 第一题&#xff1a;最长递增子序列 给你一个整数数组 nums &#xff0c;找到其中最长严格递增子序列的长度。 子序列是由数组派生而来的序列&#xff0c;删除&#xff08;或不删除&#xff09;数组中的元素而…

软测推荐第二期:10本高质量测试书籍

在不断发展的软件开发领域&#xff0c;测试是质量的守护者&#xff0c;确保产品不仅满足功能要求&#xff0c;而且提供无缝的用户体验。随着软件复杂性的增加&#xff0c;对完善的测试方法和见解的需求也随之增加。 上次给大家推荐了五本书&#xff0c;获得了大家的积极反馈&a…

二叉搜索树的最小绝对差[简单]

优质博文&#xff1a;IT-BLOG-CN 一、题目 给你一个二叉搜索树的根节点root&#xff0c;返回树中任意两不同节点值之间的最小差值。差值是一个正数&#xff0c;其数值等于两值之差的绝对值。 示例 1&#xff1a; 输入&#xff1a;root [4,2,6,1,3] 输出&#xff1a;1 示例 …

当『后设学习』碰上『工程学思维』

只要我成为一个废物&#xff0c;就没人能够利用我&#xff01; 雷猴啊&#xff0c;我是一只临期程序猿。打过几年工&#xff0c;写过几行代码。但今天我不想聊代码&#xff0c;我们聊聊学习这件事。 技术年年更新&#xff0c;尤其是前端框架&#xff0c;很多时候觉得学习速度都…

asp.net学生考试报名管理系统VS开发sqlserver数据库web结构c#编程Microsoft Visual Studio

一、源码特点 asp.net学生考试报名管理系统是一套完善的web设计管理系统系统&#xff0c;系统具有完整的源代码和数据库&#xff0c;系统主要采用B/S模式开发。开发环境为vs2010&#xff0c;数据库为sqlserver2008&#xff0c;使 用c#语言开发 应用技术&#xff1a;asp…

在VM虚拟机上安装centos并了解Linux常用命令

一. centos安装 新建一个虚拟机&#xff0c;使用ISO映像文件&#xff08;在浏览器上直接搜索阿里云镜像站&#xff0c;下载合适的镜像文件&#xff09; 安装后设置密码然后重启 重启后输入账号和密码 查看IP 输入命令&#xff1a; vi ifcfg-ens33&#xff0c;进入编辑界面&a…

程序的编译和链接

目录 翻译环境 linux下的测试 ​编辑 预定义符号 执行环境 #define定义宏 #和## # ## 宏参数的副作用 宏和函数对比 优点 缺点 #undef 条件编译 头文件包含 在标准c的任何实现中&#xff0c;存在两种环境——翻译环境和执行环境 翻译环境 翻译环境生成目标文件…

docker 安装minio,访问地址进不去

文章目录 黑马头条P37docker安装minio文图一、启动后页面一直是加载状态进不去 黑马头条P37docker安装minio文图 一、启动后页面一直是加载状态进不去 通过docker logs -f (容器id)查看日志 通过这个报错信息&#xff0c;得知最近minio 升级&#xff0c;一些启动信息和之前不…

2023年MathorCup高校数学建模挑战赛大数据挑战赛赛题浅析

比赛时长为期7天的妈杯大数据挑战赛如期开赛&#xff0c;为了帮助大家更好的选题&#xff0c;首先给大家带来赛题浅析&#xff0c;为了方便大家更好的选题。 赛道 A&#xff1a;基于计算机视觉的坑洼道路检测和识别 A题&#xff0c;图像处理类题目。这种题目的难度数模独一档…

学习鸟哥Linux shell 时遇到的unexpected operator错误

最近在学习鸟哥Linux&#xff0c;其中一个章节讲解了Linux shell script使用语法&#xff0c;运行总是错误&#xff0c;源码如下&#xff1a; #!/bin/bashread -p "Please input &#xff08;Y/N&#xff09;: " yn[ "${yn}" "Y" -o "${y…

一文详解如何从 Oracle 迁移数据到 DolphinDB

Oracle 是一个广泛使用的关系型数据库管理系统&#xff0c;它支持 ACID 事务处理&#xff0c;具有强大的安全性和可靠性&#xff0c;因此被广泛应用于各种企业级应用程序。但是&#xff0c;随着数据规模的增加和业务需求的变化&#xff0c;Oracle 的一些限制和缺点也逐渐暴露出…

【AD9361 数字接口CMOS LVDSSPI】C 并行数据 LVDS

接上一部分&#xff0c;AD9361 数字接口CMOS &LVDS&SPI 目录 一、LVDS模式数据路径和时钟信号LVDS模式数据通路信号[1] DATA_CLK[2] FB_CLK[3] Rx_FRAME[4] Rx_D[5&#xff1a;0][5] Tx_FRAME[6]Tx_D[5&#xff1a;0][7] ENABLE[8] TXNRX系列 二、LVDS最大时钟速率和信…

框架安全-CVE 复现SpringStrutsLaravelThinkPHP漏洞复现

目录 服务攻防-框架安全&CVE 复现&Spring&Struts&Laravel&ThinkPHP概述PHP-开发框架安全-Thinkphp&Laravel漏洞复现Thinkphp-3.X RCEThinkphp-5.X RCELaravel框架安全问题- CVE-2021-3129 RCE JAVAWEB-开发框架安全-Spring&Struts2Struts2框架安全…

windows下使用FFmpeg开源库进行视频编解码完整步聚

最终解码效果: 1.UI设计 2.在控件属性窗口中输入默认值 3.复制已编译FFmpeg库到工程同级目录下 4.在工程引用FFmpeg库及头文件 5.链接指定FFmpeg库 6.使用FFmpeg库 引用头文件 extern "C" { #include "libswscale/swscale.h" #include "libavdevic…

文章分类管理接口

目录 前言 新建表 获取文章分类列表接口 初始化路由模块 将路由对象导出并使用 初始化路由对象处理函数 修改路由代码 导入数据库 定义sql语句 调用db.query() 完整的获取文章分类列表处理函数 新增文章分类接口 定义路由和处理函数 验证表单数据 查询分类名称与…

CSS基础入门04

目录 1.内边距 1.1基础写法 1.2复合写法 2.外边距 2.1基础写法 2.2复合写法 2.3块级元素水平居中 3.去除浏览器默认样式 4.弹性布局 4.1初体验 5.flex 布局基本概念 6.常用属性 6.1justify-content 6.2align-items 1.内边距 padding 设置内容和边框之间的距离. …

3D RPG Course | Core 学习日记一:初识URP

前言 最近开始学习Unity中文课堂M_Studio&#xff08;麦大&#xff09;的3D RPG Course&#xff0c;学习一下3D RPG游戏核心功能的实现&#xff0c;第一课我们学习到的是地图场景的编辑&#xff0c;其中涉及到了URP渲染。 我们首先进入Unity资源商店把地图素材和人物素材导入好…

前端将图片储存table表格中,页面回显

<el-table :data"tableData" v-loading"loading" style"width: 100%" height"calc(100vh - 270px)" :size"tableSize"row-dblclick"enterClick"><el-table-column prop"name" label"文档…