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.

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)