赞
踩
当队列中堆积了很多消息时,比如对于重要客户的消息要优先消费处理时,此时优先级队列就比较适用,否则按照队列的先入先消费的话,可能重要客户要等很长时间。
创建优先级队列有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 创建生产者
本生产者发布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.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.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.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(); } }
经过测试,输出如下所示,发现确实优先消费了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
默认情况下,生产者发布消息到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();
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。