当前位置:   article > 正文

9. RabbitMq之优先级队列与惰性队列_如何调整rabbitmq消费的优先级

如何调整rabbitmq消费的优先级

1. 优先级队列

当队列中堆积了很多消息时,比如对于重要客户的消息要优先消费处理时,此时优先级队列就比较适用,否则按照队列的先入先消费的话,可能重要客户要等很长时间。
在这里插入图片描述
创建优先级队列有2种方式,一种在管理端进行创建,另一种通过代码创建。通过管理端创建的话就是如下形式,创建完成后的优先级队列属性上会有一个Pri的标志。创建优先级队列时,需要指定x-max-priority参数的值,该值代表优先级,最大可以设置255,表示消息的优先级可以在0-255之间随意设置。在这里插入图片描述
下面通过案例演示上图优先级队列,队列最大优先级设置10
1.1. 首先创建配置类,用于声明交换机以及优先级队列,并把队列绑定到交换机

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

import java.util.HashMap;

@Configuration
public class PriorityConfig {
    public static final String PRIORITY_EXCHANGE = "priority_exchange";
    public static final String PRIORITY_QUEUE = "priority_queue";
    public static final String ROUTING_KEY = "priority";

    @Bean("priorityExchange")
    public DirectExchange priorityExchange(){
        return ExchangeBuilder.directExchange(PRIORITY_EXCHANGE).durable(true).build();
    }

    @Bean("priorityQueue")
    public Queue priorityQueue(){
        HashMap map = new HashMap();
        /*队列优先级最大设置10*/
        map.put("x-max-priority", 10);
        return QueueBuilder.durable(PRIORITY_QUEUE).withArguments(map).build();
    }

    @Bean
    public Binding priorityExchangeBindQueue(@Qualifier("priorityExchange")DirectExchange priorityExchange,
                                             @Qualifier("priorityQueue")Queue priorityQueue){
        return BindingBuilder.bind(priorityQueue).to(priorityExchange).with(ROUTING_KEY);
    }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32

1.2 创建生产者
本生产者发布6条消息,发布前5条消息hello rabbitmq 1 ~ hello rabbitmq 5时,未设置消息的优先级,未设置优先级表示优先级最低,在发布第6条消息hello rabbitmq 6时设置了优先级3。

import com.lzj.config.PriorityConfig;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class PriorityProducer {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void produceMessage(){
        String message = null;
        /*前5条消息不设置优先级,如不设置优先级,默认级别最低*/
        MessagePostProcessor messagePostProcessor = msg -> msg;
        for (int i=1; i<=5; i++){
            message = "hello rabbitmq " + i;
            rabbitTemplate.convertAndSend(PriorityConfig.PRIORITY_EXCHANGE, PriorityConfig.ROUTING_KEY, message, messagePostProcessor);
        }
        /*为第6条消息设置优先级为3*/
        messagePostProcessor = msg -> {
            MessageProperties messageProperties = msg.getMessageProperties();
            messageProperties.setPriority(3);
            return msg;
        };

        message = "hello rabbitmq " + 6;
        rabbitTemplate.convertAndSend(PriorityConfig.PRIORITY_EXCHANGE, PriorityConfig.ROUTING_KEY, message, messagePostProcessor);
    }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31

1.3 创建测试类
待上面生产者创建好后,下面进行测试。注意此时可先不必创建消费者,因为一旦生产者和消费者都创建好后进行测试,由于消费者消费消息的速度比较快,复现不出优先消息的功能。如果消费者速度快,生产者每发布到队列一条消息就立即被消费了,体现不出消息的优先级特性,一定要等到队列中挤压了消息后,优先级高的消息才会被优先消费,只创建生产者就启动测试的目的就是先向队列中挤压几条消息,这也是模拟消费者由于其它原因下线的场景。

package com.lzj;

import com.lzj.producer.PriorityProducer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ConfigurableApplicationContext;

@SpringBootApplication
@Slf4j
public class SpringbootDemo {

    public static void main(String[] args) {
        ConfigurableApplicationContext app = SpringApplication.run(SpringbootDemo.class, args);
        PriorityProducer priorityProducer = app.getBean(PriorityProducer.class);
        priorityProducer.produceMessage();
    }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18

1.4. 创建消费者
待生产者已经发消息后,下面创建消费者

import com.lzj.config.ConfirmConfig;
import com.lzj.config.PriorityConfig;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class PriorityConsumer {

    @RabbitListener(queues = PriorityConfig.PRIORITY_QUEUE)
    public void consuemrMessage(Message message, Channel channel){
        String msg = new String(message.getBody());
        log.info("消费者消费消息{}", msg);
    }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18

1.5 再次启动测试
消费者上线后,重新启动测试,测试代码可以注释掉执行生产者代码的地方,观察消费者消费消息情况。

package com.lzj;

import com.lzj.producer.PriorityProducer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ConfigurableApplicationContext;

@SpringBootApplication
@Slf4j
public class SpringbootDemo {

    public static void main(String[] args) {
        ConfigurableApplicationContext app = SpringApplication.run(SpringbootDemo.class, args);
//        PriorityProducer priorityProducer = app.getBean(PriorityProducer.class);
//        priorityProducer.produceMessage();
    }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18

经过测试,输出如下所示,发现确实优先消费了hello rabbitmq 6消息

2022-08-24 01:10:07.455  INFO 8280 --- [ntContainer#4-1] com.lzj.consumer.PriorityConsumer        : 消费者消费消息hello rabbitmq 6
2022-08-24 01:10:07.463  INFO 8280 --- [ntContainer#4-1] com.lzj.consumer.PriorityConsumer        : 消费者消费消息hello rabbitmq 1
2022-08-24 01:10:07.464  INFO 8280 --- [ntContainer#4-1] com.lzj.consumer.PriorityConsumer        : 消费者消费消息hello rabbitmq 2
2022-08-24 01:10:07.466  INFO 8280 --- [ntContainer#4-1] com.lzj.consumer.PriorityConsumer        : 消费者消费消息hello rabbitmq 5
2022-08-24 01:10:07.467  INFO 8280 --- [ntContainer#4-1] com.lzj.consumer.PriorityConsumer        : 消费者消费消息hello rabbitmq 4
2022-08-24 01:10:07.470  INFO 8280 --- [ntContainer#4-1] com.lzj.consumer.PriorityConsumer        : 消费者消费消息hello rabbitmq 3
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6

2. 惰性队列

默认情况下,生产者发布消息到RabbitMQ队列时,队列中的消息都是存储到内存中的,因此消费者消费消息时才可以快速的进行消费。即使对于持久化的消息,消息在持久化到磁盘中的同时也会在内存中保留一份,消费者消费消息时才可以进行快速的修复,但对于持久化的消息,RabbitMq需要释放内存的时候,会将内存中的消息换页至磁盘中,该操作耗时较长,会阻塞队列的操作,从而无法接收生产者的消息,虽然RabbitMq开发者们一直在研究相关算法,但效果不太理想,尤其是当消息量非常大的时候。

惰性队列是RabbitMq在3.6.0版本中加入的,惰性队列会尽可能的将消息存入磁盘中,当消费者消费消息时才从把消息从磁盘加载到内存中,因此消费速度会相对较慢。但惰性队列的目标是将更多的消息存储到磁盘中,因此可以支持更长的队列。惰性队列主要应用场景,当消费者由于各种原因(比如消费者下线、宕机或者由于维护而关闭等)而导致长时间不能消费消息导致消息积压,此时应该把积压的消息存储到磁盘中,惰性队列就很有必要了。

惰性队列创建以及代码实现方式与优先级队列的完全一样,就不再写一遍了。创建懒惰队列也是2中方式,其一在管理端创建,参数要选择Lazy mode模式。
在这里插入图片描述
其二通过代码创建懒惰队列

HashMap map = new HashMap();
/*队列优先级最大设置10*/
map.put("x-queue-mode", "lazy);
QueueBuilder.durable("my_queue").withArguments(map).build();
  • 1
  • 2
  • 3
  • 4
声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/weixin_40725706/article/detail/930061
推荐阅读
相关标签
  

闽ICP备14008679号