Springboot2.x集成Rabbitmq实现延时消费

柔光的暖阳◎ 2023-06-17 05:53 13阅读 0赞

### 前言 ###

比如我们在使用mq的时候并不希望立即收到消息，比如实现3分钟后再查询订单，这时候我们就需要使用延迟消息发送了。

# RabbitMQ如何实现迟队列？ #

AMQP协议和RabbitMQ队列本身没有直接支持延迟队列功能。但是我们可以通过RabbitMQ的两个特性来曲线实现延迟队列：

**特性1、[Time To Live(TTL)][Time To Live_TTL]**

**1.RabbitMQ可以针对Queue设置x-expires**

**2.针对Message设置 x-message-ttl**

通过这两种方式来控制消息的生存时间

*  如果通过队列属性设置，队列中所有消息都有相同的过期时间。
 *  如果通过对消息进行单独设置，每条消息TTL可以不同。
 *  如果超时(两者同时设置以最先到期的时间为准)，则消息变为dead letter(死信)

**特性2、[Dead Letter Exchanges（DLX）][Dead Letter Exchanges_DLX]**

RabbitMQ的Queue可以配置x-dead-letter-exchange 和x-dead-letter-routing-key（可选）两个参数，如果队列内出现了dead letter，则按照这两个参数重新路由转发到指定的队列。

*  x-dead-letter-exchange：出现dead letter之后将dead letter重新发送到指定exchange
 *  x-dead-letter-routing-key：出现dead letter之后将dead letter重新按照指定的routing-key发送

队列出现dead letter的情况有：

*  消息或者队列的TTL过期
 *  队列达到最大长度
 *  消息被消费端拒绝（basic.reject or basic.nack）并且requeue=false

综合上述两个特性，设置了TTL规则之后当消息在一个队列中变成死信时，利用DLX特性它能被重新转发到另一个Exchange或者Routing Key，这时候消息就可以重新被消费了。

方式一：

/死信队列的应用//
    
        /**
         * DLX测试队列
         */
        @Bean
        public Queue DLXTestQueue() {
            Map<String, Object> args = new HashMap<>(2);
            args.put("x-dead-letter-exchange", "DL_exchange");
            args.put("x-dead-letter-routing-key", "DL_queue");
            return QueueBuilder.nonDurable("DLX-test-queue").withArguments(args).build();
        }
    
        @Bean
        public Exchange DLXTestExchange() {
            return ExchangeBuilder.directExchange("DLX-test-exchange").durable(true).build();
        }
    
        @Bean
        public Binding DLXTestBinding() {
            return new Binding("DLX-test-queue", Binding.DestinationType.QUEUE, "DLX-test-exchange", "DLX-test-queue", null);
        }
    
    
        ///声明一个死信队列,交换机，绑定//
        @Bean
        public Queue dlQueue() {
            return QueueBuilder.nonDurable("DL_queue").build();
        }
    
        @Bean
        public Exchange dlExchange() {
            return ExchangeBuilder.directExchange("DL_exchange").durable(true).build();
        }
    
        /**
         * 死信路由通过 DL_queue 绑定键绑定到死信队列上.
         */
        @Bean
        public Binding deadLetterBinding() {
            return new Binding("DL_queue", Binding.DestinationType.QUEUE, "DL_exchange", "DL_queue", null);
        }

发送：

/**
         * 测试死信队列
         */
        @GetMapping("ttl")
        public Resp testTTL() {
            rabbitTemplate.convertAndSend("DLX-test-exchange", "DLX-test-queue", "去私信队列", messagePostProcessor -> {
                messagePostProcessor.getMessageProperties().setExpiration("10000");
                return messagePostProcessor;
            });
            return Resp.success("ok", null);
        }

消费者：

@RabbitListener(queues = {"DL_queue"})
        public void DLXTestQueue(Message message, Channel channel) throws IOException {
            log.info("死信队列  10s 后 消费消息 {}", new String(message.getBody()));
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }

方式二：在rabbitmq 3.5.7及以上的版本提供了一个插件（rabbitmq-delayed-message-exchange）来实现延迟队列功能。同时插件依赖Erlang/OPT 18.0及以上。

插件下载地址：[https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases][https_github.com_rabbitmq_rabbitmq-delayed-message-exchange_releases]，具体可参考我[上一篇文章][Link 1]。

配置：

package com.zoo.mq.config;
    
    import org.springframework.amqp.core.*;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    import java.util.HashMap;
    import java.util.Map;
    
    
    /**
     * @author: 谢飞
     */
    @Configuration
    public class RabbitMqConfig {
    
        /**
         * 定义延迟消息发送的队列.
         */
        @Bean
        public Queue delayQueue() {
            return QueueBuilder.nonDurable("delay_queue").build();
        }
    
    
        /交换机/
    
        /**
         * 定义一个用于延迟消息发送的交换机
         * 延时队列交换机 注意这里的交换机类型：CustomExchange
         */
        @Bean
        public CustomExchange delayExchange() {
            Map<String, Object> args = new HashMap<>();
            args.put("x-delayed-type", "direct");
            return new CustomExchange("delay_exchange", "x-delayed-message", true, false, args);
        }
    
    
        绑定/
    
        /**
         * 给延时队列绑定交换机
         */
        @Bean
        public Binding redirectBinding() {
            return BindingBuilder.bind(delayQueue()).to(delayExchange()).with("delay_key").noargs();
        }
    
    }

生产者：

@GetMapping("sendTTL")
        public Resp sendTTL() {
            rabbitTemplate.convertAndSend("delay_exchange", "delay_key", "hello", messagePostProcessor -> {
                messagePostProcessor.getMessageProperties().setDelay(10000);
                return messagePostProcessor;
            });
            return Resp.success("ok", null);
        }

消费者：

@RabbitListener(queues = {"delay_queue"})
        public void delayQueue(Message message, Channel channel) throws IOException {
            log.info("delay_queue  10s 后 消费消息 {}", new String(message.getBody()));
        }

测试ok。

**使用rabbitmq-delayed-message-exchange插件时发送到队列的消息数量在web管理界面可能不可见,不影响正常功能使用**

[Time To Live_TTL]: http://www.rabbitmq.com/ttl.html#per-queue-message-ttl
[Dead Letter Exchanges_DLX]: http://www.rabbitmq.com/dlx.html
[https_github.com_rabbitmq_rabbitmq-delayed-message-exchange_releases]: https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases
[Link 1]: https://blog.csdn.net/qq_36850813/article/details/103279548