RabbitMQ学习笔记 - 优先级队列

比眉伴天荒 2022-01-21 11:03 286阅读 0赞

**参考：<<RabbitMQ实战指南>>**

优先级队列，也就是具有高优先级的队列具有高的优先权，优先级高的消息具备优先被消费的特权。

注：

*  优先级默认最低为0，最高为队列设置的最大优先级
 *  对于单条消息来谈优先级是没有什么意义的。假如消费者的消费速度大于生产者的速度且Broker中没有消息堆积的情况下，对发送的消息设置优先级就没有什么意义，因为生产者刚发完一个消息就被消费者消费了，相当于Broker中至多只有一条消息

通过队列的 **x-max-priority** 参数设置队列的最大优先级，之后在发送消息时通过 **priority** 属性再设置当前消息的优先级。

示例：

#### 一、原生api ####

...
    // 声明交换器
    String exchangeName = "direct.exchange.test.priority";
    AMQP.Exchange.DeclareOk declareOk = channel.exchangeDeclare(exchangeName, "direct", true);
    // 声明队列
    String queueName = "direct.queue.test.priority";
    Map<String, Object> arguments = new HashMap<>(4);
    // 定义队列优先的最大优先级最大为10
    arguments.put("x-max-priority", 10);
    channel.queueDeclare(queueName, true, false, false, arguments);
    // 绑定交换器和队列
    String routingKey = "direct.routing-key.test.priority";
    channel.queueBind(queueName, exchangeName, routingKey);
    
    // 设置消息的优先级为8
    byte[] content = ("Test Msg").getBytes("UTF-8");
    AMQP.BasicProperties properties = new AMQP.BasicProperties
             .Builder()
             .priority(8)
             .build();
     channel.basicPublish(exchangeName, routingKey, false, properties, content);
    ...

查看RabbitMQ管理页面，可以到队列上有 **Pri** 的标识，如下图所示：  
![优先级队列][20190609225104326.png]

#### 二、springboot ####

和原生api没有什么区别，也是在声明队列时设置队列的最大优先级，在发送消息时候再设置消息的优先级。  
（1）声明交换器、队列

@Bean
    public DirectExchange priorityExchange() { 
        return new DirectExchange(exchangeName);
    }
    
    @Bean
    public Queue priorityQueue() { 
        // 定义队列优先的最大优先级最大为10
        Map<String, Object> args = new HashMap<>(4);
        args.put("x-max-priority", 10);
        return new Queue(queueName, true, false, false, args);
    }
    
    @Bean
    Binding bindingOriorityQueue(Queue priorityQueue, DirectExchange priorityExchange) { 
        return BindingBuilder.bind(priorityQueue).to(priorityExchange).with(routingKey);
    }

（2）发送消息  
仍然是通过实现`MessagePostProcessor`接口，进一步设置消息的属性。

public boolean sendMessage() { 
        String content = "test normal message";
        this.rabbitTemplate.convertAndSend(exchangeName, routingKey, content, message -> { 
            // 设置消息的优先级
            message.getMessageProperties().setPriority(8);
            return message;
        });
        return true;
    }

springboot-rabbitmq-demo测试代码：[https://github.com/mytt-10566/springboot-rabbitmq-demo][https_github.com_mytt-10566_springboot-rabbitmq-demo]

[20190609225104326.png]: /images/20220121/f4a43b4e2e3642d9917e3d805fdada23.png
[https_github.com_mytt-10566_springboot-rabbitmq-demo]: https://github.com/mytt-10566/springboot-rabbitmq-demo