Springboot利用Rabbitmq的延时队列+死信队列实现消息延期消费 ...

简介

RabbitMQ 的延时队列（Delayed Queue）是指消息在发送到队列后，会在一定的时间内被耽误处理处罚，直到预设的耽误时间结束，消息才会被消费者消费。RabbitMQ 自己并不原生支持延时队列，但我们可以通过一些插件或特定设置来实现这种功能。
延时队列的应用场景

• 定时任务：好比定时发送邮件、推送通知等利用。
  
• 重试机制：当某些任务失败时，可以让任务在一段时间后再次尝试。
  
• 耽误消费：在特定时间后消费消息，而不是立即消费。
  

死信队列（Dead Letter Queue，简称 DLQ）是消息队列中的一个特别队列，用于存放那些因某些原因无法被正常消费的消息。死信队列在很多消息队列体系（如 RabbitMQ、Kafka 等）中都有应用。通过将无法消费的消息转移到死信队列中，可以资助开发者发现和处理处罚消息消费过程中的异常或错误。
死信队列的主要用途是：

• 错误处理处罚：用于捕捉无法消费的消息，避免这些消息丢失，并提供后续人工干预或主动重试的机会。
  
• 监控和诊断：通过查看死信队列，可以资助开发者快速发现问题，例如队列被溢出、消息格式不正确等。
  
• 防止数据丢失：将那些由于超时、队列溢出、消费失败等原因无法处理处罚的消息放入死信队列，可以保证它们不会丢失，后续可以进一步观察和处理处罚。
  

本文联合延时队列和死信队列，延时队列的消息设置逾期时间，在时间逾期之后，将消息内容推送到死信队列中举行处理处罚。
整合逻辑

依赖

1. <dependency>
2.             <groupId>org.springframework.boot</groupId>
3.             <artifactId>spring-boot-starter</artifactId>
4.         </dependency>
5.         <dependency>
6.             <groupId>org.springframework.boot</groupId>
7.             <artifactId>spring-boot-starter-web</artifactId>
8.         </dependency>
9.         <dependency>
10.             <groupId>org.projectlombok</groupId>
11.             <artifactId>lombok</artifactId>
12.             <version>1.18.30</version>
13.         </dependency>
15.         <dependency>
16.             <groupId>cn.hutool</groupId>
17.             <artifactId>hutool-all</artifactId>
18.             <version>5.8.27</version>
19.         </dependency>
21.         <dependency>
22.             <groupId>org.springframework.boot</groupId>
23.             <artifactId>spring-boot-starter-amqp</artifactId>
24.             <version>2.7.7</version>
25.         </dependency>

复制代码

设置

1. server:
2.   port: 8081
3. spring:
4.   application:
5.     name: walker-rabbitmq
6.   rabbitmq:
7.     host: 127.0.0.1
8.     port: 5672
9.     username: guest_walker
10.     password: guest

复制代码

创建交换机|队列

延时队列

1. package com.walker.rabbitmq.delay.config;
3. import com.walker.rabbitmq.delay.constants.BaseConstant;
4. import com.walker.rabbitmq.delay.enums.RabbitMqConfigEnum;
5. import org.springframework.amqp.core.*;
6. import org.springframework.context.annotation.Bean;
7. import org.springframework.context.annotation.Configuration;
9. import java.util.HashMap;
10. import java.util.Map;
12. @Configuration
13. public class OrderDelayRabbitConfig {
15. //    交换机
16.     @Bean
17.     public DirectExchange orderDelayExchange(){
18.         return new DirectExchange(RabbitMqConfigEnum.ORDER_DELAY.getExchange());
19.     }
21. //    队列
22.     @Bean
23.     public Queue orderDelayQueue(){
24.         Map<String,Object> args = new HashMap<>();
25. //        延时队列，需要绑定死信队列，在ttl到期之后，将队列转到死信队列
26.         args.put(BaseConstant.xDeadLetterExchange,RabbitMqConfigEnum.ORDER_DEAD.getExchange());
27.         args.put(BaseConstant.xDeadLetterRoutingKey,RabbitMqConfigEnum.ORDER_DEAD.getRoutingKey());
28.         return QueueBuilder.durable(RabbitMqConfigEnum.ORDER_DELAY.getQueue()).autoDelete().withArguments(args).build();
29.     }
31. //
32.     @Bean
33.     public Binding orderDelayQueueBinding(){
34.         return BindingBuilder.bind(orderDelayQueue()).to(orderDelayExchange()).with(RabbitMqConfigEnum.ORDER_DELAY.getRoutingKey());
35.     }
36. }

复制代码

死信队列

1. package com.walker.rabbitmq.delay.config;
3. import com.walker.rabbitmq.delay.enums.RabbitMqConfigEnum;
4. import org.springframework.amqp.core.*;
5. import org.springframework.context.annotation.Bean;
6. import org.springframework.context.annotation.Configuration;
8. @Configuration
9. public class OrderDeadRabbitConfig {
11. //    交换机
12.     @Bean
13.     public DirectExchange orderDeadExchange(){
14.         return new DirectExchange(RabbitMqConfigEnum.ORDER_DEAD.getExchange());
15.     }
17. //    队列
18.     @Bean
19.     public Queue orderDeadQueue(){
20.         return QueueBuilder.durable(RabbitMqConfigEnum.ORDER_DEAD.getQueue()).autoDelete().build();
21.     }
23. //   交换机、队列、路由键绑定
24.     @Bean
25.     public Binding orderDeadQueueBinding(){
26.         return BindingBuilder.bind(orderDeadQueue()).to(orderDeadExchange())
27.                 .with(RabbitMqConfigEnum.ORDER_DEAD.getRoutingKey());
28.     }
29. }

复制代码

常量

1. package com.walker.rabbitmq.delay.constants;
3. import com.walker.rabbitmq.delay.enums.RabbitMqConfigEnum;
5. public interface BaseConstant {
6.     String xDeadLetterExchange = "x-dead-letter-exchange";
7.     String xDeadLetterRoutingKey = "x-dead-letter-routing-key";
9. }

复制代码

干系枚举

1. package com.walker.rabbitmq.delay.enums;
3. import lombok.AllArgsConstructor;
4. import lombok.Getter;
6. @AllArgsConstructor
7. @Getter
8. public enum RabbitMqConfigEnum {
9.     ORDER_DELAY("direct","订单延时队列","order_delay_exchange", "order_delay_queue", "order_delay_routing_key",true),
10.     ORDER_DEAD("direct","订单死信队列","order_dead_exchange", "order_dead_queue", "order_dead_routing_key",true);
13.     private final String type;
14.     private final String title;
15.     private final String exchange;
16.     private final String queue;
17.     private final String routingKey;
18.     private final Boolean durable;
19. }

复制代码

rabbitmq工具类封装

1. package com.walker.rabbitmq.delay.service;
3. import cn.hutool.core.util.BooleanUtil;
4. import cn.hutool.json.JSONUtil;
5. import com.rabbitmq.client.Channel;
6. import com.walker.rabbitmq.delay.enums.RabbitMqConfigEnum;
7. import lombok.extern.slf4j.Slf4j;
8. import org.springframework.amqp.AmqpException;
9. import org.springframework.amqp.core.Message;
10. import org.springframework.amqp.core.MessagePostProcessor;
11. import org.springframework.amqp.rabbit.core.RabbitTemplate;
12. import org.springframework.stereotype.Component;
14. import javax.annotation.Resource;
15. import java.io.IOException;
16. import java.util.function.Consumer;
17. import java.util.function.Function;
19. @Component
20. @Slf4j
21. public class RabbitmqService {
23.     @Resource
24.     private RabbitTemplate rabbitTemplate;
27.     /**
28.     * author:walker
29.     * time: 2024/12/30
30.     * description:  无返回结果的手动处理方式
31.     */
32.     public void doWithManualAck(String event, Message message, Channel channel, Consumer<String> method) {
33.         String data = new String(message.getBody());
34.         try {
35.             method.accept(data);
36.             sendAck(message, channel);
37.         } catch (Exception e) {
38.             log.error("处理事件[{}]异常数据为：{},原因如下：",event,data, e);
39.             sendNack(message, channel);
40.         }
41.     }
43.     /**
44.      * author:walker
45.      * time: 2024/12/30
46.      * description:  根据返回的布尔类型，进行确认和拒绝
47.      */
48.     public void doWithManualAck(String event, Message message, Channel channel, Function<String,Boolean> method) {
49.         String data = new String(message.getBody());
50.         try {
51.             Boolean res = method.apply(data);
52.             if(BooleanUtil.isTrue(res)){
53.                 sendAck(message, channel);
54.             }else{
55.                 sendNack(message, channel);
56.             }
57.         } catch (Exception e) {
58.             log.error("处理事件[{}]异常数据为：{},原因如下：",event,data, e);
59.             sendNack(message, channel);
60.         }
63.     }
65.     //    确认消息
66.     public void sendAck(Message message, Channel channel) throws IOException {
67.         channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
68.     }
70.     //    拒绝消息
71.     public void sendNack(Message message, Channel channel) {
72.         try {
73.             channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
74.         } catch (IOException e) {
75.             log.error("消息消费失败，消息ID：{}", message.getMessageProperties().getMessageId(), e);
76.         }
77.     }
80.     public <T> void sendDelayMsg(RabbitMqConfigEnum configEnum, T data, Integer delayTime) {
81.         rabbitTemplate.convertAndSend(configEnum.getExchange(), configEnum.getRoutingKey(),
82.                 JSONUtil.toJsonStr(data), new MessagePostProcessor() {
83.                     @Override
84.                     public Message postProcessMessage(Message message) throws AmqpException {
85.                         message.getMessageProperties().setExpiration(String.valueOf(delayTime));
86.                         return message;
87.                     }
88.                 });
89.     }
92. }

复制代码

案例

实体类

1. package com.walker.rabbitmq.delay.entity;
3. import lombok.Data;
5. @Data
6. public class OrderInfo {
7.     private String productName;
8.     private Integer num;
9.     private String userName;
10.     private String orderTime;
11. }

复制代码

controller 创建订单

1. package com.walker.rabbitmq.delay.controller;
3. import com.walker.rabbitmq.delay.enums.RabbitMqConfigEnum;
4. import com.walker.rabbitmq.delay.entity.OrderInfo;
5. import com.walker.rabbitmq.delay.service.RabbitmqService;
6. import lombok.extern.slf4j.Slf4j;
7. import org.springframework.web.bind.annotation.GetMapping;
8. import org.springframework.web.bind.annotation.RequestMapping;
9. import org.springframework.web.bind.annotation.RestController;
11. import javax.annotation.Resource;
12. import java.util.Date;
14. @Slf4j
15. @RestController
16. @RequestMapping("/order")
17. public class OrderController {
19.     @Resource
20.     private RabbitmqService rabbitmqService;
25.     /**
26.     * author:walker
27.     * time: 2025/1/2
28.     * description:  下单
29.     */
30.     @GetMapping("/create")
31.     public void createOrder() {
32.         OrderInfo orderInfo = new OrderInfo();
33.         orderInfo.setProductName("羽毛球");
34.         orderInfo.setUserName("张三");
35.         orderInfo.setOrderTime("2025-01-02 12:00:00");
36.         orderInfo.setNum(1);
37.         log.info("发送消息时间{}",new Date());
38.         rabbitmqService.sendDelayMsg(RabbitMqConfigEnum.ORDER_DELAY,orderInfo,10000);
39.     }
42. }

复制代码

消费者监听消息

消费者接口

1. package com.walker.rabbitmq.delay.service;
3. public interface IRabbitCosumerHandler {
5.     Boolean handle(String message);
6. }

复制代码

消费者监听方法

1. package com.walker.rabbitmq.delay.listener;
3. import com.rabbitmq.client.Channel;
4. import com.walker.rabbitmq.delay.enums.RabbitMqConfigEnum;
5. import com.walker.rabbitmq.delay.constants.BaseConstant;
6. import com.walker.rabbitmq.delay.service.IRabbitCosumerHandler;
7. import com.walker.rabbitmq.delay.service.RabbitmqService;
8. import lombok.extern.slf4j.Slf4j;
9. import org.springframework.amqp.core.Message;
10. import org.springframework.amqp.rabbit.annotation.RabbitListener;
11. import org.springframework.stereotype.Component;
13. import javax.annotation.Resource;
14. import java.io.IOException;
15. import java.util.Date;
18. @Slf4j
19. @Component
20. public class OrderDeadLetterQueueConsumer implements IRabbitCosumerHandler {
22.     @Resource
23.     private RabbitmqService rabbitmqService;
26.     @RabbitListener(queues = "#{T(com.walker.rabbitmq.delay.enums.RabbitMqConfigEnum).ORDER_DEAD.getQueue()}",ackMode = "MANUAL")
27.     public void receive(Message message, Channel channel) throws IOException {
28.         log.info("接受消息时间 {}",new Date());
29.         String msg = new String(message.getBody());
30.         log.info("当前时间：{},队列{}收到消息：{}", new Date().toString(),RabbitMqConfigEnum.ORDER_DEAD.getQueue(), msg);
31.         rabbitmqService.doWithManualAck(RabbitMqConfigEnum.ORDER_DEAD.getTitle(), message, channel, this::handle);
32.     }
34.     @Override
35.     public Boolean handle(String message) {
36. //        处理逻辑
37.         log.info("进行订单取消的操作");
38.         return true;
39.     }
40. }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。