RabbitMQ自学之路（九）——RabbitMQ实现延时队列的两种方式

清疚 2021-12-20 03:35 383阅读 0赞

## 一、什么是延时队列 ##

延时队列顾名思义，即放置在该队列里面的消息是不需要立即消费的，而是等待一段时间之后取出消费。

## 二、延时队列应用于什么场景 ##

场景一：在订单系统中，一个用户下单之后通常有30分钟的时间进行支付，如果30分钟之内没有支付成功，那么这个订单将进行一场处理。这是就可以使用延时队列将订单信息发送到延时队列。

场景二：用户希望通过手机远程遥控家里的智能设备在指定的时间进行工作。这时候就可以将用户指令发送到延时队列，当指令设定的时间到了再将指令推送到智能设备。

Rabbitmq实现延时队列一般而言有两种形式：  
第一种方式：利用两个特性： Time To Live(TTL)、Dead Letter Exchanges（DLX）  
第二种方式：利用rabbitmq中的插件x-delay-message

## 三、第一种：利用TTL DLX实现延时队列的方式 ##

> AMQP协议和RabbitMQ队列本身没有直接支持延迟队列功能，但是可以通过以下特性模拟出延迟队列的功能。

![在这里插入图片描述][20190629005119657.png]  
**1、Time To Live(TTL)**

RabbitMQ可以针对Queue设置x-expires 或者 针对Message设置 x-message-ttl，来控制消息的生存时间，如果超时(两者同时设置以最先到期的时间为准)，则消息变为dead letter(死信)

A: 通过队列属性设置，队列中所有消息都有相同的过期时间。  
B: 对消息进行单独设置，每条消息TTL可以不同。

**2、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发送

用一个具体案例来实现第一种方式：用户下订单后，如何在一分钟没有支付就取消订单  
![在这里插入图片描述][2019062901464233.png]

package com.springboot.rabbitmq.example.demo5.config;
    
    import java.util.HashMap;
    import java.util.Map;
    
    import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.DirectExchange;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.core.Queue;
    import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
    import org.springframework.amqp.rabbit.connection.CorrelationData;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import lombok.extern.slf4j.Slf4j;
    
    /**
     * @method
     * @author Mr yi
     * @time 2019年6月23日
     */
    @Configuration
    @Slf4j
    public class RabbitConfigDemo5    {
    
     
    	//队列名称
    	final static String queue = "queue_demo5";
    
    	//交换机名称
    	final static String exchangeName = "deom5Exchange";
    	
    	// routingKey
    	final static String routingKey  = "keyDemo5";
    	
    	//死信消息队列名称
    	final static String deal_queue = "deal_queue_demo5";
    
    	//死信交换机名称
    	final static String deal_exchangeName = "deal_deom5Exchange";
    	
    	//死信 routingKey
    	final static String dead_RoutingKey  = "dead_routing_key";
    	
    	//死信队列 交换机标识符
        public static final String DEAD_LETTER_QUEUE_KEY = "x-dead-letter-exchange";
        
        //死信队列交换机绑定键标识符
        public static final String DEAD_LETTER_ROUTING_KEY = "x-dead-letter-routing-key";
    
        @Autowired
        private CachingConnectionFactory connectionFactory;
        
        /**
         * 
         * @method 定义队列(队列 绑定一个死信交换机,并指定routing_key)
         * @author Mr yi
         * @time 2019年6月29日
         * @return
         */
    	@Bean
    	public Queue queueDemo5() {
    		// 将普通队列绑定到死信队列交换机上
            Map<String, Object> args = new HashMap<>(2);
            //args.put("x-message-ttl", 5 * 1000);//直接设置 Queue 延迟时间 但如果直接给队列设置过期时间,这种做法不是很灵活
            //这里采用发送消息动态设置延迟时间,这样我们可以灵活控制
            args.put(DEAD_LETTER_QUEUE_KEY, deal_exchangeName);
            args.put(DEAD_LETTER_ROUTING_KEY, dead_RoutingKey);
            return new Queue(RabbitConfigDemo5.queue, true, false, false, args);
    	}
    
    	//声明一个direct类型的交换机
    	@Bean
    	DirectExchange exchangeDemo5() {
    		return new DirectExchange(RabbitConfigDemo5.exchangeName);
    	}
    
    	//绑定Queue队列到交换机,并且指定routingKey 
    	@Bean
    	Binding bindingDirectExchangeDemo5(   ) {
    		return BindingBuilder.bind(queueDemo5()).to(exchangeDemo5()).with(routingKey);
    	}
    	
    	//创建配置死信队列
        @Bean
        public Queue deadQueue5() {
            Queue queue = new Queue(deal_queue, true);
            return queue;
        }
    	
        //创建死信交换机
         @Bean
         public DirectExchange deadExchange5() {
             return new DirectExchange(deal_exchangeName);
         }
    	
         //死信队列与死信交换机绑定
          @Bean
          public Binding bindingDeadExchange5() {
              return BindingBuilder.bind(deadQueue5()).to(deadExchange5()).with(dead_RoutingKey);
          }
    
    /**      @Bean
          public RabbitTemplate rabbitTemplate(){
      		//若使用confirm-callback ，必须要配置publisherConfirms 为true
      		connectionFactory.setPublisherConfirms(true);
      		//若使用return-callback，必须要配置publisherReturns为true
              connectionFactory.setPublisherReturns(true);
              RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
              //使用return-callback时必须设置mandatory为true，或者在配置中设置mandatory-expression的值为true
             // rabbitTemplate.setMandatory(true);
       
              // 如果消息没有到exchange,则confirm回调,ack=false; 如果消息到达exchange,则confirm回调,ack=true
              rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
                  @Override
                  public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                      if(ack){
                          log.info("消息发送成功:correlationData({}),ack({}),cause({})",correlationData,ack,cause);
                      }else{
                          log.info("消息发送失败:correlationData({}),ack({}),cause({})",correlationData,ack,cause);
                      }
                  }
              });
              
              //如果exchange到queue成功,则不回调return;如果exchange到queue失败,则回调return(需设置mandatory=true,否则不回回调,消息就丢了)
              rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
                  @Override
                  public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                      log.info("消息丢失:exchange({}),route({}),replyCode({}),replyText({}),message:{}",exchange,routingKey,replyCode,replyText,message);
                  }
              });
              return rabbitTemplate;
          }
    
    **/
    	 
    
    }

生产者产生订单后，将订单信息发送到rabbitmq 服务段，设置TTL 时间，如果超过了这个时间，还没有消费这个消息，那么就变为死信，发送到死信队列中。

这里利用死信的机制来巧妙的实现延时，我这里没有设置正常消费者，即生产者发送消息后，消息不会被消费，那么在指定时间后，变为死信，有与死信队列绑定的消费者来消费消息（判断订单是否已经成功支付）

package com.springboot.rabbitmq.example.demo5.producers;
    
    import java.util.Date;
    import java.util.UUID;
    
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.core.MessageBuilder;
    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 com.alibaba.fastjson.JSONObject;
    
    import lombok.extern.slf4j.Slf4j;
    /**
     * 
     * @method 生产者
     * @author Mr yi
     * @time 2019年6月19日
     */
    @Component
    @Slf4j
    public class ProducersDemo5  {
    
     
        @Autowired
        private AmqpTemplate rabbitTemplate;
        
    	/**
    	 * @method 生产者发送消息,direct模式下需要传递一个routingKey
    	 * @author Mr yi
    	 * @time 2019年6月19日
    	 * @throws Exception
    	 */
    	public void send( ) throws Exception {
    		
    		log.info("【订单生成时间】" + new Date().toString() +"【1分钟后检查订单是否已经支付】"  );
            
            this.rabbitTemplate.convertAndSend("deom5Exchange", "keyDemo5", "订单实体类对象信息", message -> {
                // 如果配置了 params.put("x-message-ttl", 5 * 1000); 那么这一句也可以省略,具体根据业务需要是声明 Queue 的时候就指定好延迟时间还是在发送自己控制时间
                message.getMessageProperties().setExpiration(1 * 1000 * 60 + "");
                return message;
            });
    		 
    	}
    	 
    }

死信消息消费者

package com.springboot.rabbitmq.example.demo5.consumers;
    
    import java.io.IOException;
    import java.util.Date;
    import java.util.Map;
    
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.amqp.support.AmqpHeaders;
    import org.springframework.messaging.handler.annotation.Header;
    import org.springframework.messaging.handler.annotation.Headers;
    import org.springframework.stereotype.Component;
    import com.alibaba.fastjson.JSONObject;
    import com.rabbitmq.client.Channel;
    
    import lombok.extern.slf4j.Slf4j;
    
    /**
     * 
     * @method  死信消费者，消费从死信队列传来的消息
     * @author Mr yi
     * @time 2019年6月19日
     */
    @Component
    @Slf4j
    public class ConsumersDemo5Deal {
    	
    	@RabbitListener(queues = "deal_queue_demo5")
        public void process(String order,  Message message, @Headers Map<String, Object> headers, Channel channel) throws IOException {
    	
    		log.info("【 监听到延时队列消息】 - 【消费时间】 - [{}]- 【订单内容】 - [{}]",  new Date(), order); 
    		// 判断订单是否已经支付，如果支付则；否则，取消订单（逻辑代码省略)
    		
            // 手动ack
            Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
            // 手动签收
            channel.basicAck(deliveryTag, false);
            System.out.println("执行结束....");
            
        }
    	
    }

测试

@Autowired
    	private ProducersDemo5 producers;
    	
    	@RequestMapping("/send")
        public String send() throws Exception {
        	producers.send();
        	return "success";
        }

启动程序，测试

发现queue\_demo5正常队列有一条消息处于待续状态

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI5OTE0ODM3_size_16_color_FFFFFF_t_70]  
![!\[在这里插入图片描述\](https://img-blog.csd][https_img-blog.csd]  
等待一分钟后，控制台输出  
![在这里插入图片描述][20190629015554850.png]  
发现queue\_demo5 消息已经被消费（发送到deal\_queue\_demo5死信队列了）  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI5OTE0ODM3_size_16_color_FFFFFF_t_70 1]  
**使用死信队列实现延时消息的缺点**：

> 1） 如果统一用队列来设置消息的TTL，当延时时间梯度比较多的话，比如1分钟，2分钟，5分钟，10分钟，20分钟，30分钟……需要创建很多交换机和队列来路由消息。  
> 2） 如果单独设置消息的TTL，则可能会造成队列中的消息阻塞——前一条消息没有出队（没有被消费），后面的消息无法投递。  
> 3） 可能存在一定的时间误差。

## 四、第二种：利用rabbitmq-delayed-message-exchange插件来实现延迟队列功能 ##

插件下载地址：注意下载插件要和安装的rabbitmq版本一致，我这里下载的是3.7的  
[https://www.rabbitmq.com/community-plugins.html][https_www.rabbitmq.com_community-plugins.html]

![在这里插入图片描述][20190629125645327.png]

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

下载解压后，得到一个.ez的压缩文件，找到rabbitmq安装目录的plugins文件夹，将解压的文件复制进去

![在这里插入图片描述][20190629150632313.png]  
![在这里插入图片描述][20190629130011268.png]  
重新启动rabbitmq ，输入命令  
`rabbitmq-plugins enable rabbitmq_delayed_message_exchange`

停止：net stop RabbitMQ
    启动：net start RabbitMQ

![在这里插入图片描述][20190629151830302.png]

**配置类**

package com.springboot.rabbitmq.example.demo6.config;
    
    import java.util.HashMap;
    import java.util.Map;
    
    import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.CustomExchange;
    import org.springframework.amqp.core.DirectExchange;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.core.Queue;
    import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
    import org.springframework.amqp.rabbit.connection.CorrelationData;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import lombok.extern.slf4j.Slf4j;
    
    /**
     * @method
     * @author Mr yi
     * @time 2019年6月23日
     */
    @Configuration
    @Slf4j
    public class RabbitConfigDemo6    {
    
     
    	//队列名称
    	final static String queue = "queue_demo6";
    
    	//交换机名称
    	final static String exchangeName = "deom6Exchange";
    	
    	// routingKey
    	final static String routingKey  = "keyDemo6";
    	
    	 
    
        @Autowired
        private CachingConnectionFactory connectionFactory;
        
    	@Bean
    	public Queue queueDemo6() {
    		// 第一个参数是创建的queue的名字，第二个参数是是否支持持久化
            return new Queue(RabbitConfigDemo6.queue, true);
    	}
    
    	@Bean
        public CustomExchange delayExchange6() {
            Map<String, Object> args = new HashMap<String, Object>();
            args.put("x-delayed-type", "direct");
            return new CustomExchange(RabbitConfigDemo6.exchangeName, "x-delayed-message", true, false, args);
        }
     
    	@Bean
        public Binding bindingNotify6() {
            return BindingBuilder.bind(queueDemo6()).to(delayExchange6()).with(RabbitConfigDemo6.routingKey).noargs();
        }
     
    	 
    
    /**      @Bean
          public RabbitTemplate rabbitTemplate(){
      		//若使用confirm-callback ，必须要配置publisherConfirms 为true
      		connectionFactory.setPublisherConfirms(true);
      		//若使用return-callback，必须要配置publisherReturns为true
              connectionFactory.setPublisherReturns(true);
              RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
              //使用return-callback时必须设置mandatory为true，或者在配置中设置mandatory-expression的值为true
             // rabbitTemplate.setMandatory(true);
       
              // 如果消息没有到exchange,则confirm回调,ack=false; 如果消息到达exchange,则confirm回调,ack=true
              rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
                  @Override
                  public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                      if(ack){
                          log.info("消息发送成功:correlationData({}),ack({}),cause({})",correlationData,ack,cause);
                      }else{
                          log.info("消息发送失败:correlationData({}),ack({}),cause({})",correlationData,ack,cause);
                      }
                  }
              });
              
              //如果exchange到queue成功,则不回调return;如果exchange到queue失败,则回调return(需设置mandatory=true,否则不回回调,消息就丢了)
              rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
                  @Override
                  public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                      log.info("消息丢失:exchange({}),route({}),replyCode({}),replyText({}),message:{}",exchange,routingKey,replyCode,replyText,message);
                  }
              });
              return rabbitTemplate;
          }
    
    **/
    	 
    
    }

**生产者，设置setDelay(1 \* 1000 \* 60 ); 延时 1分钟**

package com.springboot.rabbitmq.example.demo6.producers;
    
    import java.text.SimpleDateFormat;
    import java.util.Date;
    import java.util.UUID;
    
    import org.springframework.amqp.AmqpException;
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.core.MessageBuilder;
    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 com.alibaba.fastjson.JSONObject;
    
    import lombok.extern.slf4j.Slf4j;
    /**
     * 
     * @method 生产者
     * @author Mr yi
     * @time 2019年6月19日
     */
    @Component
    @Slf4j
    public class ProducersDemo6 {
    
     
        @Autowired
        private AmqpTemplate rabbitTemplate;
        
    	/**
    	 * @method 生产者发送消息,direct模式下需要传递一个routingKey
    	 * @author Mr yi
    	 * @time 2019年6月19日
    	 * @throws Exception
    	 */
    	public void send( ) throws Exception {
    		
    		log.info("【订单生成时间】" + new Date().toString() +"【1分钟后检查订单是否已经支付】"  );
    		
    		this.rabbitTemplate.convertAndSend("deom6Exchange", "keyDemo6", "订单实体类对象信息", new MessagePostProcessor() {
                @Override
                public Message postProcessMessage(Message message) throws AmqpException {
                    message.getMessageProperties().setDelay(1 * 1000 * 60 );
                    return message;
                }
            });
     
    		 
    	}
    	 
    }

**消费者，消费者一分钟后得到生产者发送的消息**

package com.springboot.rabbitmq.example.demo6.consumers;
    
    import java.io.IOException;
    import java.util.Date;
    import java.util.Map;
    
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.amqp.support.AmqpHeaders;
    import org.springframework.messaging.handler.annotation.Header;
    import org.springframework.messaging.handler.annotation.Headers;
    import org.springframework.stereotype.Component;
    import com.alibaba.fastjson.JSONObject;
    import com.rabbitmq.client.Channel;
    
    import lombok.extern.slf4j.Slf4j;
    
    /**
     * 
     * @method  消费者
     * @author Mr yi
     * @time 2019年6月19日
     */
    @Component
    @Slf4j
    public class ConsumersDemo6 {
    	
    	@RabbitListener(queues = "queue_demo6")
        public void process(String order,  Message message, @Headers Map<String, Object> headers, Channel channel) throws IOException {
    	
    		log.info("【 监听到延时队列消息】 - 【消费时间】 - [{}]- 【订单内容】 - [{}]",  new Date(), order); 
    		// 判断订单是否已经支付，如果支付则；否则，取消订单（逻辑代码省略)
    		
            // 手动ack
            Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
            // 手动签收
            channel.basicAck(deliveryTag, false);
            System.out.println("执行结束....");
            
        }
    	
    }

测试

@Autowired
    	private ProducersDemo6 producers;
    	
    	@RequestMapping("/send")
        public String send() throws Exception {
        	producers.send();
        	return "success";
        }

**启动程序，执行方法**

控制台输出  
![在这里插入图片描述][20190629150847461.png]  
rabbitmq服务端，queue\_demo6 其中并没有消息进入就绪状态，这一点也是和第一种方式（使用死信）的区别优势所在。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI5OTE0ODM3_size_16_color_FFFFFF_t_70 3]

等待一分钟后，消费者接受到消息控制台  
![在这里插入图片描述][20190629150905444.png]  
![在这里插入图片描述][20190629150940841.png]

源码下载：[https://download.csdn.net/download/qq\_29914837/11264460][https_download.csdn.net_download_qq_29914837_11264460]

--------------------

如果你觉得本篇文章对你有所帮助的话，麻烦请点击头像右边的关注按钮，谢谢！

技术在交流中进步，知识在分享中传播

[20190629005119657.png]: /images/20211220/6a7ecb2288484cf0b5652ab50fa82f53.png
[2019062901464233.png]: /images/20211220/9cb3bf72074945e5991b41d6dddfc140.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI5OTE0ODM3_size_16_color_FFFFFF_t_70]: /images/20211220/8951f45c0e274b9da7d5e9b171d3e995.png
[https_img-blog.csd]: /images/20211220/6f0f6cc60705412581d7251fdeb390c3.png
[20190629015554850.png]: /images/20211220/39f0244373c84fa89385817932246c14.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI5OTE0ODM3_size_16_color_FFFFFF_t_70 1]: /images/20211220/71755c0068b747e6a4e6e1d40ba5505e.png
[https_www.rabbitmq.com_community-plugins.html]: https://www.rabbitmq.com/community-plugins.html
[20190629125645327.png]: /images/20211220/8e8746fd60dc4d9da4ca1f3cfaf2104a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI5OTE0ODM3_size_16_color_FFFFFF_t_70 2]: /images/20211220/3972462f2bca4f809a630b2eef5a93b3.png
[20190629150632313.png]: /images/20211220/be634eb91680449fb3b39197b7adca09.png
[20190629130011268.png]: /images/20211220/f4c4eaf5c81c44e4b90623b72daf52f8.png
[20190629151830302.png]: /images/20211220/20240830e5674c8ebe1ce00858332027.png
[20190629150847461.png]: /images/20211220/02b4b667185d4935bce9d334faad2567.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI5OTE0ODM3_size_16_color_FFFFFF_t_70 3]: /images/20211220/600ba77d1ac84cd6b3d9674b9601486b.png
[20190629150905444.png]: /images/20211220/75f750ff5d3a4f1881c8fe2209ec1f71.png
[20190629150940841.png]: /images/20211220/05e80bc6d05a411ea5dbc643bbf90bbe.png
[https_download.csdn.net_download_qq_29914837_11264460]: https://download.csdn.net/download/qq_29914837/11264460