rabbitmq的延时队列

清疚 2022-12-27 12:55 148阅读 0赞

## 延时队列的应用场景 ##

那么什么时候需要用延时队列呢？考虑一下以下场景：

订单在十分钟之内未支付则自动取消。  
新创建的店铺，如果在十天内都没有上传过商品，则自动发送消息提醒。  
账单在一周内未支付，则自动结算。  
用户注册成功后，如果三天内没有登陆则进行短信提醒。  
用户发起退款，如果三天内没有得到处理则通知相关运营人员。  
预定会议后，需要在预定的时间点前十分钟通知各个与会人员参加会议。  
这些场景都有一个特点，需要在某个事件发生之后或者之前的指定时间点完成某一项任务，如：发生订单生成事件，在十分钟之后检查该订单支付状态，然后将未支付的订单进行关闭；发生店铺创建事件，十天后检查该店铺上新商品数，然后通知上新数为0的商户；发生账单生成事件，检查账单支付状态，然后自动结算未支付的账单；发生新用户注册事件，三天后检查新注册用户的活动数据，然后通知没有任何活动记录的用户；发生退款事件，在三天之后检查该订单是否已被处理，如仍未被处理，则发送消息给相关运营人员；发生预定会议事件，判断离会议开始是否只有十分钟了，如果是，则通知各个与会人员。

看起来似乎使用定时任务，一直轮询数据，每秒查一次，取出需要被处理的数据，然后处理不就完事了吗？如果数据量比较少，确实可以这样做，比如：对于“如果账单一周内未支付则进行自动结算”这样的需求，如果对于时间不是严格限制，而是宽松意义上的一周，那么每天晚上跑个定时任务检查一下所有未支付的账单，确实也是一个可行的方案。但对于数据量比较大，并且时效性较强的场景，如：“订单十分钟内未支付则关闭“，短期内未支付的订单数据可能会有很多，活动期间甚至会达到百万甚至千万级别，对这么庞大的数据量仍旧使用轮询的方式显然是不可取的，很可能在一秒内无法完成所有订单的检查，同时会给数据库带来很大压力，无法满足业务要求而且性能低下。

## rabbitmq中的TTL ##

TTL是什么呢？TTL是RabbitMQ中一个消息或者队列的属性，表明一条消息或者该队列中的所有消息的最大存活时间，单位是毫秒。换句话说，如果一条消息设置了TTL属性或者进入了设置TTL属性的队列，那么这条消息如果在TTL设置的时间内没有被消费，则会成为“死信”（至于什么是死信，请翻看上一篇）。如果同时配置了队列的TTL和消息的TTL，那么较小的那个值将会被使用。  
那么，如何设置这个TTL值呢？有两种方式，第一种是在创建队列的时候设置队列的“x-message-ttl”属性，如下：

Map<String, Object> args = new HashMap<String, Object>();
    args.put("x-message-ttl", 6000);
    channel.queueDeclare(queueName, durable, exclusive, autoDelete, args);

这样所有被投递到该队列的消息都最多不会存活超过6s。

另一种方式便是针对每条消息设置TTL，代码如下：

MessageProperties messageProperties = new MessageProperties();
    messageProperties.setExpiration("5000");
    Message message = new Message(msg.getBytes(), messageProperties);
    rabbitTemplate.convertAndSend(DELAY_EXCHANGE_NAME, DELAY_QUEUEC_ROUTING_KEY, message);

但这两种方式是有区别的，如果设置了队列的TTL属性，那么一旦消息过期，就会被队列丢弃，而第二种方式，消息即使过期，也不一定会被马上丢弃，因为消息是否过期是在即将投递到消费者之前判定的，如果当前队列有严重的消息积压情况，则已过期的消息也许还能存活较长时间。

另外，还需要注意的一点是，如果不设置TTL，表示消息永远不会过期，如果将TTL设置为0，则表示除非此时可以直接投递该消息到消费者，否则该消息将会被丢弃。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xxcTQwNDI3MDIwMQ_size_16_color_FFFFFF_t_70]

## 声明交换机队列 ##

package com.qiangqiang.dalay_queue.config;
    
    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;
    import java.util.Map;
    
    /**
     * \* Created with IntelliJ IDEA.
     * \* @author: xiyue
     * \* Date: 2020/12/11
     * \* Time: 9:50
     * \* To change this template use File | Settings | File Templates.
     * \* Description:
     * \
     */
    @Configuration
    public class RabbitMQConfig {
        public static final String DELAY_EXCHANGE_NAME = "delay.queue.demo.business.exchange";
        public static final String DELAY_QUEUEA_NAME = "delay.queue.demo.business.queuea";
        public static final String DELAY_QUEUEB_NAME = "delay.queue.demo.business.queueb";
        public static final String DELAY_QUEUEA_ROUTING_KEY = "delay.queue.demo.business.queuea.routingkey";
        public static final String DELAY_QUEUEB_ROUTING_KEY = "delay.queue.demo.business.queueb.routingkey";
        public static final String DEAD_LETTER_EXCHANGE = "delay.queue.demo.deadletter.exchange";
        public static final String DEAD_LETTER_QUEUEA_ROUTING_KEY = "delay.queue.demo.deadletter.delay_10s.routingkey";
        public static final String DEAD_LETTER_QUEUEB_ROUTING_KEY = "delay.queue.demo.deadletter.delay_60s.routingkey";
        public static final String DEAD_LETTER_QUEUEA_NAME = "delay.queue.demo.deadletter.queuea";
        public static final String DEAD_LETTER_QUEUEB_NAME = "delay.queue.demo.deadletter.queueb";
    
        public static final String DEAD_LETTER_QUEUEC_ROUTING_KEY = "delay.queue.demo.deadletter.delay_anytime.routingkey";
        public static final String DEAD_LETTER_QUEUEC_NAME = "delay.queue.demo.deadletter.queuec";
        public static final String DELAY_QUEUEC_NAME = "delay.queue.demo.business.queuec";
        public static final String DELAY_QUEUEC_ROUTING_KEY = "delay.queue.demo.business.queuec.routingkey";
    
        //声明延时交换机
        @Bean("delayExchange")
        public DirectExchange delayExchange(){
            return new DirectExchange(DELAY_EXCHANGE_NAME);
        }
    
        //声明死信交换机
        @Bean("deadLetterExchange")
        public DirectExchange deadLetterExchange(){
            return new DirectExchange(DEAD_LETTER_EXCHANGE);
        }
    
        //声明延时队列A
        //并绑定到死信交换机上
        @Bean("delayQueueA")
        public Queue delayQueueA(){
            Map<String, Object> map = new HashMap<>(2);
            map.put("x-dead-letter-exchange", DEAD_LETTER_EXCHANGE);
            // x-dead-letter-routing-key  这里声明当前队列的死信路由key
            map.put("x-dead-letter-routing-key", DEAD_LETTER_QUEUEA_ROUTING_KEY);
            // x-message-ttl  声明队列的TTL
            map.put("x-message-ttl", 6000);
            return QueueBuilder.durable(DELAY_QUEUEA_NAME).withArguments(map).build();
        }
    
        //声明延时对别B
        //并绑定到死信交换机
        @Bean("delayQueueB")
        public Queue delayQueueB(){
            Map<String, Object> args = new HashMap<>(2);
            // x-dead-letter-exchange    这里声明当前队列绑定的死信交换机
            args.put("x-dead-letter-exchange", DEAD_LETTER_EXCHANGE);
            // x-dead-letter-routing-key  这里声明当前队列的死信路由key
            args.put("x-dead-letter-routing-key", DEAD_LETTER_QUEUEB_ROUTING_KEY);
            // x-message-ttl  声明队列的TTL
            args.put("x-message-ttl", 10000);
            return QueueBuilder.durable(DELAY_QUEUEB_NAME).withArguments(args).build();
        }
    
    
        // 声明延时队列C 不设置TTL
        // 并绑定到对应的死信交换机
        @Bean("delayQueueC")
        public Queue delayQueueC(){
            Map<String, Object> args = new HashMap<>(3);
            // x-dead-letter-exchange    这里声明当前队列绑定的死信交换机
            args.put("x-dead-letter-exchange", DEAD_LETTER_EXCHANGE);
            // x-dead-letter-routing-key  这里声明当前队列的死信路由key
            args.put("x-dead-letter-routing-key", DEAD_LETTER_QUEUEC_ROUTING_KEY);
            return QueueBuilder.durable(DELAY_QUEUEC_NAME).withArguments(args).build();
        }
    
        // 声明死信队列C 用于接收延时任意时长处理的消息
        @Bean("deadLetterQueueC")
        public Queue deadLetterQueueC(){
            return new Queue(DEAD_LETTER_QUEUEC_NAME);
        }
    
        // 声明延时列C绑定关系
        @Bean
        public Binding delayBindingC(@Qualifier("delayQueueC") Queue queue,
                                     @Qualifier("delayExchange") DirectExchange exchange){
            return BindingBuilder.bind(queue).to(exchange).with(DELAY_QUEUEC_ROUTING_KEY);
        }
    
        // 声明死信队列C绑定关系
        @Bean
        public Binding deadLetterBindingC(@Qualifier("deadLetterQueueC") Queue queue,
                                          @Qualifier("deadLetterExchange") DirectExchange exchange){
            return BindingBuilder.bind(queue).to(exchange).with(DEAD_LETTER_QUEUEC_ROUTING_KEY);
        }
    
    
        //声明死信队列A,用于接收延时的消息
        @Bean("deadLetterQueueA")
        public Queue deadLetterQueueA(){
            return new Queue(DEAD_LETTER_QUEUEA_NAME);
        }
    
        // 声明死信队列B 用于接收延时60s处理的消息
        @Bean("deadLetterQueueB")
        public Queue deadLetterQueueB(){
            return new Queue(DEAD_LETTER_QUEUEB_NAME);
        }
    
        //声明延时队列A与延时交换机的绑定关系
        @Bean
        public Binding delayBindingA(@Qualifier("delayQueueA") Queue queue ,
                                     @Qualifier("delayExchange")DirectExchange exchange){
            return BindingBuilder.bind(queue).to(exchange).with(DELAY_QUEUEA_ROUTING_KEY);
        }
    
        //声明延时队列B有延时交换机的绑定关系
        @Bean
        public Binding delayBindingB(@Qualifier("delayQueueB") Queue queue ,
                                     @Qualifier("delayExchange")DirectExchange exchange){
            return BindingBuilder.bind(queue).to(exchange).with(DELAY_QUEUEB_ROUTING_KEY);
        }
    
        //声明死信队列A与死信交换机的绑定
        @Bean
        public Binding deadLetterBindingA(@Qualifier("deadLetterQueueA") Queue queue,
                                          @Qualifier("deadLetterExchange") DirectExchange exchange){
            return BindingBuilder.bind(queue).to(exchange).with(DEAD_LETTER_QUEUEA_ROUTING_KEY);
        }
    
        // 声明死信队列B绑定关系
        @Bean
        public Binding deadLetterBindingB(@Qualifier("deadLetterQueueB") Queue queue,
                                          @Qualifier("deadLetterExchange") DirectExchange exchange){
            return BindingBuilder.bind(queue).to(exchange).with(DEAD_LETTER_QUEUEB_ROUTING_KEY);
        }
    }

## 接下来，创建两个消费者，分别对两个死信队列的消息进行消费： ##

package com.qiangqiang.dalay_queue.consumer;
    
    import com.rabbitmq.client.Channel;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    import java.io.IOException;
    import java.util.Date;
    
    import static com.qiangqiang.dalay_queue.config.RabbitMQConfig.DEAD_LETTER_QUEUEA_NAME;
    import static com.qiangqiang.dalay_queue.config.RabbitMQConfig.DEAD_LETTER_QUEUEB_NAME;
    import static com.qiangqiang.dalay_queue.config.RabbitMQConfig.DEAD_LETTER_QUEUEC_NAME;
    
    /**
     * \* Created with IntelliJ IDEA.
     * \* @author: xiyue
     * \* Date: 2020/12/11
     * \* Time: 10:20
     * \* To change this template use File | Settings | File Templates.
     * \* Description:
     * \
     */
    @Component
    public class DeadLetterQueueConsumer {
        @RabbitListener(queues = DEAD_LETTER_QUEUEA_NAME)
        public void receiveA(Message message, Channel channel) throws IOException {
            String msg = new String(message.getBody());
            System.out.println(new Date() + "死信队列A收到消息：{"+msg+"}" );
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }
        @RabbitListener(queues = DEAD_LETTER_QUEUEB_NAME)
        public void receiveB(Message message, Channel channel) throws IOException {
            String msg = new String(message.getBody());
            System.out.println(new Date() + "死信队列B收到消息：{"+msg+"}" );
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }
    
        @RabbitListener(queues = DEAD_LETTER_QUEUEC_NAME)
        public void receiveC(Message message, Channel channel) throws IOException {
            String msg = new String(message.getBody());
            System.out.println(new Date() + "死信队列C收到消息：{"+msg+"}" );
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }
    
    }

然后是消息的生产者：

package com.qiangqiang.dalay_queue.producer;
    
    import org.springframework.amqp.AmqpException;
    import org.springframework.amqp.core.Message;
    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;
    
    import static com.qiangqiang.dalay_queue.config.RabbitMQConfig.*;
    
    /**
     * \* Created with IntelliJ IDEA.
     * \* @author: xiyue
     * \* Date: 2020/12/11
     * \* Time: 10:24
     * \* To change this template use File | Settings | File Templates.
     * \* Description:
     * \
     */
    @Component
    public class DelayMessageSender {
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        public void sendMsg(String msg, Integer type){
            if(type.equals(6000)){
                rabbitTemplate.convertAndSend(DELAY_EXCHANGE_NAME, DELAY_QUEUEA_ROUTING_KEY, msg);
            }else if(type.equals(10000)){
                rabbitTemplate.convertAndSend(DELAY_EXCHANGE_NAME, DELAY_QUEUEB_ROUTING_KEY, msg);
            }else{
                MessageProperties messageProperties = new MessageProperties();
                messageProperties.setExpiration(type.toString());
                Message message = new Message(msg.getBytes(), messageProperties);
                rabbitTemplate.convertAndSend(DELAY_EXCHANGE_NAME, DELAY_QUEUEC_ROUTING_KEY, message);
            }
        }
    
    }

## 接下来，我们暴露一个web接口来生产消息： ##

package com.qiangqiang.dalay_queue.controller;
    
    import com.qiangqiang.dalay_queue.producer.DelayMessageSender;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    import java.util.Date;
    import java.util.Objects;
    
    /**
     * \* Created with IntelliJ IDEA.
     * \* @author: xiyue
     * \* Date: 2020/12/11
     * \* Time: 10:29
     * \* To change this template use File | Settings | File Templates.
     * \* Description:
     * \
     */
    @RequestMapping("rabbitmq")
    @RestController
    public class RabbitMQMsgController {
        @Autowired
        private DelayMessageSender sender;
    
        @RequestMapping("sendmsg")
        public void sendMsg(String msg, Integer delayType){
            System.out.println(new Date() + "发送请求，msg:" + msg + "延时时间:"+delayType);
            sender.sendMsg(msg, delayType);
        }
    }

## application配置 ##

#配置rabbitmq服务器的ip地址
    spring.rabbitmq.host=192.168.200.106
    #rabbitmq的端口号
    spring.rabbitmq.port=5672
    #rabbitmq的账号
    spring.rabbitmq.username=root
    #rabbitmq的密码
    spring.rabbitmq.password=root
    #rabbitmq的虚拟机
    spring.rabbitmq.virtual-host=/
    #发送确认
    #NONE   禁用发布模式，是默认值
    #CORRELATED 发布消息成功到交换机后会触发回调方法，交互
    #SIMPLE
    spring.rabbitmq.publisher-confirm-type=correlated
    #路由失败回调
    spring.rabbitmq.publisher-returns=true
    #true消息路由失败通知监听者,false将消息丢弃
    spring.rabbitmq.template.mandatory=true
    
    spring.rabbitmq.listener.type=simple
    spring.rabbitmq.listener.simple.acknowledge-mode=manual
    spring.rabbitmq.listener.simple.default-requeue-rejected=false

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xxcTQwNDI3MDIwMQ_size_16_color_FFFFFF_t_70]: /images/20221120/1fe054ba332e461d84efa2ef7ef0c36c.png