详解RabbitMQ中延迟队列结合业务场景的使用
作者:小白的救赎
这篇文章主要介绍了详解RabbitMQ中延迟队列结合业务场景的使用,延迟队列中的元素都是带有时间属性的,延迟队列就是用来存放需要在指定时间被处理的元素的队列,需要的朋友可以参考下
消息进入队列后不会立即被消费,只有到达指定时间后才会被消费。业务场景就是支付时间内未支付就清除订单或者用户注册一段时间后发短信问候。在这里想说的是这只是一种思想,并不是真正的一种用法,这种思想所需要的用法就是用上消息TTL存活时间以及死信队列来实现。
生产者端
目录结构
导入依赖
<dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <scope>test</scope> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> <version>2.5.0</version> </dependency> <dependency> <groupId>junit</groupId> <artifactId>junit</artifactId> <version>4.12</version> <scope>test</scope> </dependency> </dependencies>
修改yml
spring:
rabbitmq:
host: localhost
port: 5672
username: guest
password: guest
publisher-returns: true # 开启退回回调
#三个类型:none默认不开启确认回调 correlated开启确认回调
#simple也会确认回调 还会调用waitForConfirms()方法或waitForConfirmsOrDie()方法
publisher-confirm-type: correlated # 开启确认回调
业务逻辑
@Component public class RabbitMQConfig { public static final String EXCHANGE_NAME = "order_exchange_name"; public static final String QUEUE_NAME = "order_queue_name"; public static final String DELAY_EXCHANGE_NAME = "delay_exchange_name"; public static final String DELAY_QUEUE_NAME = "delay_queue_name"; @Bean("orderExchange") public Exchange testExchange(){ return ExchangeBuilder.topicExchange(EXCHANGE_NAME).durable(true).build(); } @Bean("delayExchange") public Exchange deadExchange(){ return ExchangeBuilder.topicExchange(DELAY_EXCHANGE_NAME).durable(true).build(); } //订单队列绑定延迟交换机并且带上路由键 @Bean("orderQueue") public Queue testQueue(){ return QueueBuilder.durable(QUEUE_NAME).deadLetterExchange(DELAY_EXCHANGE_NAME) .deadLetterRoutingKey("order.delay.user").build(); } @Bean("delayQueue") public Queue deadQueue(){ return QueueBuilder.durable(DELAY_QUEUE_NAME).build(); } @Bean public Binding link(@Qualifier("orderExchange") Exchange exchange, @Qualifier("orderQueue") Queue queue){ return BindingBuilder.bind(queue).to(exchange).with("order.#").noargs(); } @Bean public Binding deadLink(@Qualifier("delayExchange") Exchange exchange, @Qualifier("delayQueue") Queue queue){ return BindingBuilder.bind(queue).to(exchange).with("order.delay.#").noargs(); } }
@SpringBootTest @RunWith(SpringRunner.class) class RabbitmqProducerApplicationTests { @Autowired private RabbitTemplate rabbitTemplate; @Test void testProducer() throws InterruptedException { rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() { @Override public void confirm(CorrelationData correlationData, boolean b, String s) { if(b) System.out.println("交换机成功接受到了消息"); else System.out.println("消息失败原因" + s); } }); // 设置交换机处理失败消息的模式 // true:消息到达不了队列时 会将消息重新返回给生产者 false:消息到达不了队列直接丢弃 rabbitTemplate.setMandatory(true); rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() { @Override public void returnedMessage(Message message, int i, String s, String s1, String s2) { System.out.println("队列接受不到交换机的消息进行了失败回调"); } }); // 以上代码均是为了保证消息的可靠性传递 // 对消息进行后置处理 设置其过期时间为10s MessagePostProcessor messagePostProcessor = new MessagePostProcessor() { @Override public Message postProcessMessage(Message message) throws AmqpException { message.getMessageProperties().setExpiration("10000"); return message; } }; // 下单成功发送消息 rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME,"order.delay.user","ikun书籍", messagePostProcessor); } }
测试结果
消费者端
目录结构
导入依赖
<dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <scope>test</scope> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> <version>2.5.0</version> </dependency> <dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> </dependency> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-api</artifactId> </dependency> </dependencies>
修改yml
spring:
rabbitmq:
host: localhost
port: 5672
username: guest
password: guest
listener:
simple:
acknowledge-mode: manual # 开启手动确认
业务逻辑
@Slf4j @Component public class OrderListener implements ChannelAwareMessageListener { @RabbitListener(queues = "delay_queue_name") // 监听的是死信队列 @Override public void onMessage(Message message, Channel channel) throws Exception { long deliveryTag = message.getMessageProperties().getDeliveryTag(); try { // 这里要注意在监听这条消息前 肯定会有接口点击支付去更改支付状态的业务逻辑 此处我是做了10s的订单业务而已 log.info("您在时间为:{},时有一条订单为:{}", LocalDateTime.now().minusSeconds(10), new String(message.getBody())); // 下面开始接受订单消息逻辑 log.info("将订单id传入数据库查询订单支付字段"); log.info("字段为支付成功状态就手动确认签收"); log.info("字段为未支付状态就取消订单并且回滚事务"); channel.basicAck(deliveryTag,false);// 仅确认本次消息 } catch (Exception e){ log.info("出现异常 拒绝签收消息 并且不重回队列"); channel.basicNack(deliveryTag,false,false); } } }
测试结果
到此这篇关于详解RabbitMQ中延迟队列结合业务场景的使用的文章就介绍到这了,更多相关RabbitMQ延迟队列内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!