SpringBoot整合RabbitMQ实现消息延迟队列(含源码)

淡淡的烟草味﹌ 2024-03-17 14:53 81阅读 0赞

## 环境依赖 ##

SpringBoot 3.1.0

JDK 17

## 前期准备 ##

安装MQ: [liunx+docker+rabbitmq安装延迟队列插件][liunx_docker_rabbitmq]

## 实例 ##

实现延迟队列的一种方式是在 RabbitMQ 中使用消息延迟插件，这个插件可以让你在消息发送时设置一个延迟时间，超过这个时间后消息才会被消费者接收到。下面是 SpringBoot 整合 RabbitMQ 实现延迟队列的简单步骤：

### 1.添加 RabbitMQ 的 Maven 依赖 ###

<dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>

### 2.配置 RabbitMQ ###

在 application.properties 配置文件中添加 RabbitMQ 的连接信息：

spring.rabbitmq.host=127.0.0.1
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=test
    spring.rabbitmq.password=test
    spring.rabbitmq.virtual-host=/
    # 手动应答
    #spring.rabbitmq.listener.simple.acknowledge-mode=manual
    #每次从队列中取一个,轮询分发，默认是公平分发
    spring.rabbitmq.listener.simple.prefetch=1
    # 开启重试
    spring.rabbitmq.listener.simple.retry.enabled=true
    # 重试次数
    spring.rabbitmq.listener.simple.retry.max-attempts=5

### 3.配置文件 ###

@Configuration
    public class RabbitMQOrderConfig {
    
        /**
         * 订单交换机
         */
        public static final String ORDER_EXCHANGE = "order_exchange";
        /**
         * 订单队列
         */
        public static final String ORDER_QUEUE = "order_queue";
        /**
         * 订单路由key
         */
        public static final String ORDER_QUEUE_ROUTING_KEY = "order.#";
    
        /**
         * 死信交换机
         */
        public static final String ORDER_DEAD_LETTER_EXCHANGE = "order_dead_letter_exchange";
        /**
         * 死信队列 routingKey
         */
        public static final String ORDER_DEAD_LETTER_QUEUE_ROUTING_KEY = "order_dead_letter_queue_routing_key";
    
        /**
         * 死信队列
         */
        public static final String ORDER_DEAD_LETTER_QUEUE = "order_dead_letter_queue";
    
        /**
         * 延迟时间 （单位：ms(毫秒)）
         */
        public  static final Integer DELAY_TIME = 10000;
    
    
        /**
         * 创建死信交换机
         */
        @Bean("orderDeadLetterExchange")
        public Exchange orderDeadLetterExchange() {
            return new TopicExchange(ORDER_DEAD_LETTER_EXCHANGE, true, false);
        }
    
        /**
         * 创建死信队列
         */
        @Bean("orderDeadLetterQueue")
        public Queue orderDeadLetterQueue() {
            return QueueBuilder.durable(ORDER_DEAD_LETTER_QUEUE).build();
        }
    
        /**
         * 绑定死信交换机和死信队列
         */
        @Bean("orderDeadLetterBinding")
        public Binding orderDeadLetterBinding(@Qualifier("orderDeadLetterQueue") Queue queue, @Qualifier("orderDeadLetterExchange")Exchange exchange) {
            return BindingBuilder.bind(queue).to(exchange).with(ORDER_DEAD_LETTER_QUEUE_ROUTING_KEY).noargs();
        }
    
    
        /**
         * 创建订单交换机
         */
        @Bean("orderExchange")
        public Exchange orderExchange() {
            return new TopicExchange(ORDER_EXCHANGE, true, false);
        }
    
        /**
         * 创建订单队列
         */
        @Bean("orderQueue")
        public Queue orderQueue() {
            Map<String, Object> args = new HashMap<>(3);
            //消息过期后，进入到死信交换机
            args.put("x-dead-letter-exchange", ORDER_DEAD_LETTER_EXCHANGE);
    
            //消息过期后，进入到死信交换机的路由key
            args.put("x-dead-letter-routing-key", ORDER_DEAD_LETTER_QUEUE_ROUTING_KEY);
    
            //过期时间，单位毫秒
            args.put("x-message-ttl", DELAY_TIME);
    
            return QueueBuilder.durable(ORDER_QUEUE).withArguments(args).build();
        }
    
        /**
         * 绑定订单交换机和队列
         */
        @Bean("orderBinding")
        public Binding orderBinding(@Qualifier("orderQueue") Queue queue, @Qualifier("orderExchange")Exchange exchange) {
            return BindingBuilder.bind(queue).to(exchange).with(ORDER_QUEUE_ROUTING_KEY).noargs();
        }
    }

### 4.定义消息实体类 ###

定义一个消息体类，用来存储需要发送的消息：

@Slf4j
    @Data
    @Builder
    public class OrderMessage implements Serializable {
        
    
        /**
         * 商户订单号
         */
        private String orderId;
    
        /**
         * 支付宝订单号
         */
        private String tradeNo;
    }

### 5.定义消息发送者 ###

定义一个 RabbitMQ 消息发送者类，用来发送消息到 RabbitMQ：

@Slf4j
    @Component
    public class MessageSender {
        
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        public void sendOrderMessage(OrderMessage message) {
        
            //为true,则交换机处理消息到路由失败，则会返回给生产者 配置文件指定，则这里不需指定
            rabbitTemplate.setMandatory(true);
            //开启强制消息投递（mandatory为设置为true），但消息未被路由至任何一个queue，则回退一条消息
            rabbitTemplate.setReturnsCallback(returned -> {
        
                int code = returned.getReplyCode();
                System.out.println("code=" + code);
                System.out.println("returned=" + returned);
            });
            rabbitTemplate.convertAndSend(RabbitMQConfig.ORDER_EXCHANGE, "order", message);
    
            log.info("===============延时队列生产消息====================");
            log.info("发送时间:{},发送内容:{}, {}ms后执行", LocalDateTime.now(), message, RabbitMQConfig.DELAY_TIME);
        }
    }

### 6.定义消息消费者 ###

定义一个 RabbitMQ 消息消费者类，用来接收并处理消息：

@Component
    @Slf4j
    @RabbitListener(queues = RabbitMQConfig.ORDER_DEAD_LETTER_QUEUE)
    public class OrderMQListener {
        
    
        @RabbitHandler
        public void consumer(OrderMessage orderMessage, Message message, Channel channel) throws IOException {
        
            log.info("收到消息：{}",new Date());
            log.info("msgTag：{}", message.getMessageProperties().getDeliveryTag());
            log.info("message：{}", message);
            log.info("content：{}", orderMessage);
        }
    }

这里使用了 @RabbitListener 注解来将一个方法标记为一个 RabbitMQ 消息监听器，通过设置 queues 属性来指定监听的队列名称。

### 7.定义一个controller ###

@Slf4j
    @Api(tags = "延迟消息接口")
    @RestController
    @RequestMapping("/rabbitmq_order_delay_message")
    public class RabbitMQDelayMessageController {
    
        @Autowired
        private MessageSender sender;
    
        /**
         * 发送消息
         * @return
         */
        @RequestMapping(value = "/sendMsg", method = RequestMethod.GET)
        @ResponseBody
        public void sendMsg() {
            OrderMessage orderMessage = OrderMessage.builder().orderId(UUID.randomUUID().toString()).tradeNo(UUID.randomUUID().toString()).build();
            sender.sendOrderMessage(orderMessage);
        }
    }

启动项目，请求运行结果：

![image-20230704171525938][]

总的xml：

<dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</artifactId>
    </dependency>
    
    <dependency>
        <groupId>com.xiaoleilu</groupId>
        <artifactId>hutool-all</artifactId>
        <version>3.0.7</version>
    </dependency>
    
    <dependency>
        <groupId>io.swagger</groupId>
        <artifactId>swagger-annotations</artifactId>
        <version>${swagger-annotations.version}</version>
    </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>
    </dependency>
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    
    <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>fastjson</artifactId>
        <version>1.2.73</version>
        <scope>compile</scope>
    </dependency>

## 问题总结 ##

### 1.Invalid argument, ‘x-delayed-type’ must be an existing exchange type ###

需要创建一个交换机

![image-20230703161834850][]

### 2.Connection refused: no further information ###

请检查配置 application.xml配置的rabbimq不生效，可以将配置放到application.properties

### 3.Channel shutdown: channel error; protocol method: \#method<channel.close>(reply-code=406, reply-text=PRECONDITION\_FAILED - unknown delivery tag 1, class-id=60, method-id=80) ###

这种情况：

1.消费者内部重复签收导致签收异常

解决方案:增加配置手动处理应答

1.  配置新增

spring.rabbitmq.listener.simple.acknowledge-mode=manual #手动签收

1.  代码里: 增加channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);

public void consumer(String body, Message message, Channel channel) throws IOException {
            long msgTag = message.getMessageProperties().getDeliveryTag();
            try {
                System.out.println("收到消息：" + new Date());
                System.out.println("msgTag=" + msgTag);
                System.out.println("message=" + message);
                System.out.println("body=" + body);
                channel.basicAck(msgTag, false);
            }catch (Exception e) {
                log.error("【订单延迟关闭处理异常】 接收到消息为：" + msgTag + " ，消息异常消费 ： ", e);
            } finally {
                // 处理完之后手动签收（这里再次签收）
                channel.basicAck(msgTag, false);
            }
        }

2.已经是自动处理了，然后代码里还有手动处理channel.basicAck(msgTag, false)

解决方案：去除channel.basicAck(msgTag, false)

### 4.Failed to convert message ###

消息发送和接收的方式不对 比如发送的是对象，则接收的也必须是对象，发送的是string ，接收的也必须是string

![image-20230704174011064][]

![image-20230704174057321][]

##### 如果需要完整源码请关注公众号"架构殿堂" ，回复 "SpringBoot+RabbitMQ实现消息延迟队列"获得 #####

## 写在最后 ##

**如果大家对相关文章感兴趣，可以关注公众号"架构殿堂"，会持续更新AIGC，java基础面试题, netty, spring boot,spring cloud等系列文章，一系列干货随时送达!**

[liunx_docker_rabbitmq]: https://blog.csdn.net/jinxinxin1314/article/details/131349218
[image-20230704171525938]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/c71ca424f763466fab5a380e0df124ba.png
[image-20230703161834850]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/51d54657623449b4a85be21fa333eb63.png
[image-20230704174011064]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/d706c2dd6a3148a9a1ab2ff8205b8439.png
[image-20230704174057321]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/b636beb5833a447c8d19963c984a988b.png