Springboot整合RabbitMQ优先级队列

使用场景

在我们订单系统中有一个订单催付的场景，我们的客户在天猫下的订单,会及时将订单推送给我们，如果在用户设定的时间内未付款那么就会给用户推送一条短信提醒，很简单的一个功能对吧，但是，天猫商家对我们来说，肯定是要分大客户和小客户的对吧，比如像苹果，小米这样大商家一年起码能给我们创造很大的利润，所以理应当然，他们的订单必须得到优先处理，而曾经我们的后端系统是使用 redis 来存放的定时轮询，大家都知道 redis 只能用 List 做一个简简单单的消息队列，并不能实现一个优先级的场景，所以订单量大了后采用 RabbitMQ 进行改造和优化,如果发现是大客户的订单给一个相对比较高的优先级，否则就是默认优先级。

在后台管理页面添加

注意事项 要让队列实现优先级需要做的事情有如下事情:

队列需要设置为优先级队列 消息需要设置消息的优先级 消费者需要等待消息已经发送到队列中才去消费因为，这样才有机会对消息进行排序

配置代码

@Slf4j
@Component
public class PriorityConfig {
          
   

    public static final String EXCHANGE = "priority-exchange";

    public static final String QUEUE = "priority-queue";

    public static final String ROUTING_KEY = "priority_key";

    /**
     * 定义优先级队列
     */
    @Bean
    Queue queue() {
          
   
        Map<String, Object> args= new HashMap<>();
        args.put("x-max-priority", 10); //设置优先级，范围0-255，此处设为10，则允许优先级的范围为0-10
        return new Queue(QUEUE, false, false, false, args);
    }

    /**
     * 定义交换器
     */
    @Bean
    DirectExchange exchange() {
          
   
        return new DirectExchange(EXCHANGE);
    }


    @Bean
    Binding binding(Queue queue, DirectExchange exchange) {
          
   
        return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY);
    }

}

生产者

/**
     * 模拟发送多条数据
     */
    @GetMapping("/sendPriorityMessage")
    public void sendPriorityMessage(){
          
   
        String message = "";
        for (int i = 0; i < 10;i++){
          
   
            message = "info" + i;
            if (i == 5){
          
    // i= 5 设置优先级为10 ,优先级也可以作为形参接受
                rabbitTemplate.convertAndSend(PriorityConfig.EXCHANGE,PriorityConfig.ROUTING_KEY,message,msg -> {
          
   
                    msg.getMessageProperties().setPriority(10);
                    return msg;
                });
            }else {
          
   
                rabbitTemplate.convertAndSend(PriorityConfig.EXCHANGE,PriorityConfig.ROUTING_KEY,message,msg -> {
          
   
                    msg.getMessageProperties().setPriority(5); 
                    return msg;
                });
            }
        }

        log.info("发出消息success");
    }

配置文件

#配置MQ连接信息
spring.rabbitmq.addresses=192.168.235.128
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=123456
#spring.rabbitmq.publisher-confirm-type=correlated
spring.rabbitmq.publisher-returns=false
spring.rabbitmq.listener.direct.acknowledge-mode=manual
# # 每个消费者可能未完成的最大未确认消息数量
spring.rabbitmq.listener.simple.prefetch=1
# 最小线程数量
spring.rabbitmq.listener.simple.concurrency=10
# 最大线程数量
spring.rabbitmq.listener.simple.max-concurrency=10

优先队列的注意点: 只有当消费者不足,即消息无法被即使消费而阻塞的时候,优先队列才会根据优先级来分配任务的执行序

启动进程测试