当生产者发送消息的速度超过了消费者处理消息的速度,就会导致队列中的消息堆积,直到队列存储消息达到上限。最早接收到的消息,可能就会成为死信,会被丢弃,这就是消息堆积问题。
解决消息堆积有三种思路:
上面呢,我们已经 知道解决消息队列的常见三种解决方案,其中一种方案就是想办法去提高一个队列它能存储一个消息量的上限。
但是RabbitMQ呢是内存存储的,如果说在高并发的情况下消息量非常的大,这些消息我们如果都给它丢到内存当中,显然是不合适的,所以我们就要学习一个惰性队列来解决这个问题!
从RabbitMQ的3.6.0版本开始,就增加了Lazy Queues的概念,也就是惰性队列。惰性队列的特征如下:
package com.jie.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 {
/**
* @description:惰性队列
* @author: jie
* @time: 2022/3/13 11:06
*/
@Bean
public Queue lazyQueue(){
return QueueBuilder.durable("lazy.queue")
.lazy()
.build();
}
/**
* @description:普通队列
* @author: jie
* @time: 2022/3/13 11:06
*/
@Bean
public Queue normalQueue(){
return QueueBuilder.durable("normal.queue")
.build();
}
}
运行,查看浏览器。
/**
* @description:声明惰性队列
* @author: jie
* @time: 2022/3/13 14:37
*/
@RabbitListener(queuesToDeclare = @Queue(
name = "lazy.queue",
durable = "true",
arguments = @Argument(name = "x-queue-mode",value = "lazy")
))
public void listenLazyQueue(String msg){
System.out.println("消费者接收到simple.queue的消息:【" + msg + "】");
}
package com.jie.mq.spring;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
import java.nio.charset.StandardCharsets;
@Slf4j
@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
public void testLazyQueue() throws InterruptedException {
for (int i = 0; i < 1000000; i++) {
String routingKey = "delay";
//1、准备消息
Message message = MessageBuilder
.withBody("hell,Spring".getBytes(StandardCharsets.UTF_8))
.setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT)
.build();
//2、发送消息
rabbitTemplate.convertAndSend("lazy.queue", message);
}
}
@Test
public void testNormaQueue() throws InterruptedException {
for (int i = 0; i < 1000000; i++) {
String routingKey = "delay";
//1、准备消息
Message message = MessageBuilder
.withBody("hell,Spring".getBytes(StandardCharsets.UTF_8))
.setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT)
.build();
//2、发送消息
rabbitTemplate.convertAndSend("normal.queue", message);
}
}
}
消息堆积问题的解决方案?
惰性队列的优点有哪些?
惰性队列的缺点有哪些?