• 首页 首页 icon
  • 工具库 工具库 icon
    • IP查询 IP查询 icon
  • 内容库 内容库 icon
    • 快讯库 快讯库 icon
    • 精品库 精品库 icon
    • 问答库 问答库 icon
  • 更多 更多 icon
    • 服务条款 服务条款 icon

RabbitMQ消息堆积问题

武飞扬头像
阿杰学编程
帮助2

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

解决消息堆积有三种思路:

  1. 增加更多消费者,提高消费速度

  2. 在消费者内开启线程池加快消息处理速度

  3. 扩大队列容积,提高堆积上限

1、惰性队列

上面呢,我们已经 知道解决消息队列的常见三种解决方案,其中一种方案就是想办法去提高一个队列它能存储一个消息量的上限。

但是RabbitMQ呢是内存存储的,如果说在高并发的情况下消息量非常的大,这些消息我们如果都给它丢到内存当中,显然是不合适的,所以我们就要学习一个惰性队列来解决这个问题!

RabbitMQ3.6.0版本开始,就增加了Lazy Queues的概念,也就是惰性队列。惰性队列的特征如下:

  1. 接收到消息后直接存入磁盘而非内存

  2. 消费者要消费消息时才会从磁盘中读取并加载到内存

  3. 支持数百万条的消息存储

1.1 基于@Bean声明lazy-queue

学新通

  1.  
    package com.jie.mq.config;
  2.  
     
  3.  
    import org.springframework.amqp.core.Queue;
  4.  
    import org.springframework.amqp.core.QueueBuilder;
  5.  
    import org.springframework.context.annotation.Bean;
  6.  
    import org.springframework.context.annotation.Configuration;
  7.  
     
  8.  
    @Configuration
  9.  
    public class LazyConfig {
  10.  
    /**
  11.  
    * @description:惰性队列
  12.  
    * @author: jie
  13.  
    * @time: 2022/3/13 11:06
  14.  
    */
  15.  
    @Bean
  16.  
    public Queue lazyQueue(){
  17.  
     
  18.  
    return QueueBuilder.durable("lazy.queue")
  19.  
    .lazy()
  20.  
    .build();
  21.  
    }
  22.  
    /**
  23.  
    * @description:普通队列
  24.  
    * @author: jie
  25.  
    * @time: 2022/3/13 11:06
  26.  
    */
  27.  
    @Bean
  28.  
    public Queue normalQueue(){
  29.  
    return QueueBuilder.durable("normal.queue")
  30.  
    .build();
  31.  
    }
  32.  
     
  33.  
    }

运行,查看浏览器。学新通

1.2 基于@RabbitListener声明LazyQueue

  1.  
    /**
  2.  
    * @description:声明惰性队列
  3.  
    * @author: jie
  4.  
    * @time: 2022/3/13 14:37
  5.  
    */
  6.  
    @RabbitListener(queuesToDeclare = @Queue(
  7.  
    name = "lazy.queue",
  8.  
    durable = "true",
  9.  
    arguments = @Argument(name = "x-queue-mode",value = "lazy")
  10.  
    ))
  11.  
    public void listenLazyQueue(String msg){
  12.  
    System.out.println("消费者接收到simple.queue的消息:【" msg "】");
  13.  
    }

学新通

1.3 发送消息

  1.  
    package com.jie.mq.spring;
  2.  
     
  3.  
    import lombok.extern.slf4j.Slf4j;
  4.  
    import org.junit.Test;
  5.  
    import org.junit.runner.RunWith;
  6.  
    import org.springframework.amqp.core.Message;
  7.  
    import org.springframework.amqp.core.MessageBuilder;
  8.  
    import org.springframework.amqp.core.MessageDeliveryMode;
  9.  
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
  10.  
    import org.springframework.beans.factory.annotation.Autowired;
  11.  
    import org.springframework.boot.test.context.SpringBootTest;
  12.  
    import org.springframework.test.context.junit4.SpringRunner;
  13.  
     
  14.  
    import java.nio.charset.StandardCharsets;
  15.  
     
  16.  
    @Slf4j
  17.  
    @RunWith(SpringRunner.class)
  18.  
    @SpringBootTest
  19.  
    public class SpringAmqpTest {
  20.  
    @Autowired
  21.  
    private RabbitTemplate rabbitTemplate;
  22.  
     
  23.  
    @Test
  24.  
    public void testLazyQueue() throws InterruptedException {
  25.  
    for (int i = 0; i < 1000000; i ) {
  26.  
    String routingKey = "delay";
  27.  
    //1、准备消息
  28.  
    Message message = MessageBuilder
  29.  
    .withBody("hell,Spring".getBytes(StandardCharsets.UTF_8))
  30.  
    .setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT)
  31.  
    .build();
  32.  
    //2、发送消息
  33.  
    rabbitTemplate.convertAndSend("lazy.queue", message);
  34.  
    }
  35.  
    }
  36.  
     
  37.  
    @Test
  38.  
    public void testNormaQueue() throws InterruptedException {
  39.  
    for (int i = 0; i < 1000000; i ) {
  40.  
    String routingKey = "delay";
  41.  
    //1、准备消息
  42.  
    Message message = MessageBuilder
  43.  
    .withBody("hell,Spring".getBytes(StandardCharsets.UTF_8))
  44.  
    .setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT)
  45.  
    .build();
  46.  
    //2、发送消息
  47.  
    rabbitTemplate.convertAndSend("normal.queue", message);
  48.  
    }
  49.  
    }
  50.  
     
  51.  
    }

2、总结

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

  1. 队列上绑定多个消费者,提高消费速度

  2. 使用惰性队列,可以再mq中保存更多消息

惰性队列的优点有哪些?

  1. 基于磁盘存储,消息上限高

  2. 没有间歇性的page-out,性能比较稳定

惰性队列的缺点有哪些?

  1. 基于磁盘存储,消息时效性会降低

  2. 性能受限于磁盘的IO

这篇好文章是转载于:学新通技术网

  • 版权申明: 本站部分内容来自互联网,仅供学习及演示用,请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系,请提供相关证据及您的身份证明,我们将在收到邮件后48小时内删除。
  • 本站站名: 学新通技术网
  • 本文地址: /boutique/detail/tanhgbhbef
系列文章
更多 icon
同类精品
更多 icon
继续加载