目录
一、惰性队列
1.1、消息堆积问题
1.2、消息堆积问题的解决方法
从消费者的角度：
从队列的角度：
1.3、引入惰性队列
1.3.1、什么是惰性队列
1.3.2、惰性队列的使用
1.3.3、效果演示
一、惰性队列
1.1、消息堆积问题
当生产者发送消息的速度超过了消费者处理消息的速度，就会导致队列中的消息堆积，直到队列存储消息到达上限.
这就像是有一个蓄水池，一边往里注水，一边向外排水，但是注水速度比排水快，因此这个水池最终就会填满.
接着，最早收到的消息，就可能成为死信，默认情况下会把死信丢弃，丢弃了之后队列就有了空间，就可以新消息进入队列.
1.2、消息堆积问题的解决方法
从消费者的角度：
1. 增加更多的消费者，调高消费速度.
如果说现在只有单个消费者，处理速度是 100ms 一个消息，那么如果有三个消费者，理论上就处理速度就是 300ms 一个消息.
2.在消费者内开启一个线程池，每当一个消息过来了，就交给一个线程去处理，那么当前这个消费者内部处理消息的速度就加快了.
但是，这个方法有一个限制，就是如果消息特别多的情况下，你是不是就需要分配很多线程. 线程越多，对 cpu 来讲也是一种浪费，因为 cpu 需要在多个线程之间做上下文切换，所以这个方案比较适合消息处理业务比较耗时的情况，开多个线程，让 cpu 并行处理.
从队列的角度：
3. 扩大队列容积，提高堆积上限.
就像一个蓄水池，我把这个水池搞的像黄河那么大，你再往里进水，啥时候才能填满呢~
前两种方法我们已经是比较的熟悉的了，但是最后扩大队列容积，具体该如何实现呢？
惰性队列就可以用来解决这个问题~
1.3、引入惰性队列
1.3.1、什么是惰性队列
对于传统的队列来讲，如果没有开启消息持久化，所有接收到的消息都是放在内存里面的，目的就是为了加快消息投递的速度，这也是 RabbitMQ 的一个很优势——响应速度快.
但是他也带来了一个问题，RabbitMQ 设置了一个内存预警值（内存存储的上限，默认是 40%），如果在消息堆积的情况下，很容易就到达这个预警值，此时，RabbitMQ 就会处于一个暂停状态，会阻止生产者投递消息进来，然后把内存中的一部分消息清理出来，刷出到磁盘中，这动作也叫 “page out” . 进而导致 mq 的并发能力，忽高忽低，性能不稳定（每次page out 都会比较耗时，停顿一段时间）.
而惰性队列，就是是专门用来处理消息堆积问题的~
他有以下三个特点：
1. 收到的消息会直接写入磁盘，而非内存，也因此很难触发 mq 的内存预警，几乎不会出现 page out 的情况.
2. 消费者消费惰性队列的消息，也需要先从磁盘中读取并加载到内存中. 实际上这也会使延迟稍微高一点，毕竟磁盘的性能和内存还是有很大差距的，但是也在可以接收到范围内.
3. 支持数百万的消息存储，这也是因为他是磁盘存储（空间非常大）.
1.3.2、惰性队列的使用
设置一个队列为 惰性队列，有以下三种方式：
1. 在 RabbitMQ Managerment 中，我们只需要在声明队列的时候，指定 x-queue-mode 属性为 lazy 即可.
2. 用 SpringAMQP 声明惰性队列，@Bean 方式如下：
@Configuration
public class LazyConfig {
@Bean
public Queue lazyQueue() {
return QueueBuilder
.durable("lazy.queue")
.lazy() //开启 x-queue-mode 属性为 lazy
.build();
}
}
3. 用 SpringAMQP 声明惰性队列，注解方式如下：
@RabbitListener(queuesToDeclare = @Queue(
name = "lazy.queue",
durable = "true",
arguments = @Argument(name = "x-queue-mode", value = "lazy")
))
public void listenLazyQueue(String msg) {
log.info("消费者接收到了惰性队列的消息! msg=" + msg);
}
1.3.3、效果演示
a）使用 @Bean 声明一个普通队列和一个惰性队列，来对比效果.
@Configuration
public class LazyConfig {
@Bean
public Queue lazyQueue() {
return QueueBuilder
.durable("lazy.queue")
.lazy() //开启 x-queue-mode 属性为 lazy
.build();
}
@Bean
public Queue normalQueue() {
return QueueBuilder
.durable("normal.queue")
.build();
}
}
b）使用两个生产者，分别向两个队列中发送 100000 消息.
@Test
public void testLazyMessage() {
for(int i = 0; i < 1000000; i++) {
//1.准备消息
Message message = MessageBuilder.withBody("hello lazy message".getBytes())
.setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT)
.build();
//2.发送消息
rabbitTemplate.convertAndSend("lazy.queue", message);
}
}
@Test
public void testNormalMessage() {
for(int i = 0; i < 1000000; i++) {
//1.准备消息
Message message = MessageBuilder.withBody("hello normal message".getBytes())
.setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT)
.build();
//2.发送消息
rabbitTemplate.convertAndSend("normal.queue", message);
}
}
普通队列情况如下：
普通队列每次往 memory 中写入一定量的数据后，暂停一段时间，向磁盘写数据（page out）.
lazy 队列情况如下：
lazy 一直都往磁盘上写数据，所以 total 等于 paged out.
文章来源：https://www.toymoban.com/news/detail-753252.html
文章来源地址https://www.toymoban.com/news/detail-753252.html
到了这里，关于RabbitMQ - 消息堆积问题的最佳解决方案？惰性队列的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
