RabbitMQ 延迟任务（限时订单） 原理 以及代码 实战

Posted 2021-01-25 dbgzxx

• 一、场景

　　　　我们经常会碰见，一个需求就是，发送一条指令（消息），延迟一段时间执行，比如说常见的淘宝当下了一个订单后，订单支付时间为半个小时，如果半个小时没有支付，则关闭该订单。当然实现的方式有几种，今天来看看rabbitMQ实现的方式。

• 二、思路：rabbitMQ 如何实现 

1. rabbitMQ为每个队列设置消息的超时时间。只要给队列设置x-message-ttl 参数，就设定了该队列所有消息的存活时间，时间单位是毫秒。如果声明队列时指定了死信交换器，则过期消息会成为死信消息

2. 需要设置的参数为：

   

• 三、原理：上图

  

1. 将延迟队列（queue）在声明的时候设置参数 “ x-dead-letter-exchange ”，“ x-message-ttl “ 分别对应 死信路由器（dlx_exchange） 和 消息过期时间(比如说30分钟)。

2.  一个消息从生产者发送到延迟队列 ，在延迟队列里等待，等待30分钟后，会去绑定的死信路由（dlx_exchange）。通过死信路由的规则，走到死信队列。

3. 这时候监听死信队列的消费者就可以接收到消息，消费消息。比如说查看该订单是否支付，如果没有支付，则关闭该订单。

• 四、代码实战

1. 生产者代码：
   

   import com.dbg.example.connectionFactory.RQconnFactory;
   import com.rabbitmq.client.BuiltinExchangeType;
   import com.rabbitmq.client.Channel;
   import com.rabbitmq.client.Connection;
   import com.rabbitmq.client.ConnectionFactory;
   import org.slf4j.Logger;
   import org.slf4j.LoggerFactory;
   
   import java.io.IOException;
   import java.util.ArrayList;
   import java.util.HashMap;
   import java.util.List;
   import java.util.Map;
   import java.util.concurrent.TimeoutException;
   
   
   public class DelayProducer {
   
           public static Logger logger = LoggerFactory.getLogger(DelayProducer.class);
   
           // 交换机
           public static final String exchangeName = "delay_Task";
           // 声明路由键
           public static final String routekey = "delay_order";
   
           public static void  pend (List<String> orderNos) throws IOException, TimeoutException {
   
                   // 通过单例得到工厂
                   ConnectionFactory connectionFactory = RQconnFactory.getrQconnFactory();
                   Connection connection = connectionFactory.newConnection();
                   final Channel channel = connection.createChannel();
   
                   // 声明一个交换机
                   channel.exchangeDeclare(exchangeName , BuiltinExchangeType.DIRECT);
   
                   // 消息绑定
                   for (String orderNo :orderNos ) {
                       channel.basicPublish(exchangeName , routekey , null ,orderNo.getBytes());
                       logger.info("发送订单 : "+ orderNo);
                   }
                   // 关闭频道和连接
                   channel.close();
                   connection.close();
           }
   
           public static void main(String[] args) {
                   List orders = new ArrayList();
                   orders.add("order 0001");
                   orders.add("order 0002");
                   orders.add("order 0003");
                   try {
                           pend(orders);
                   } catch (IOException e) {
                           e.printStackTrace();
                   } catch (TimeoutException e) {
                           e.printStackTrace();
                   }
           }
   
   
   }

    

2. 消费者代码：
   

   import com.dbg.example.connectionFactory.RQconnFactory;
   import com.rabbitmq.client.*;
   
   import java.io.IOException;
   import java.util.HashMap;
   import java.util.Map;
   import java.util.concurrent.TimeoutException;
   
   
   public class DelayConsumer {
   
       public static void receive() throws IOException, TimeoutException {
   
           // 通过单例得到工厂
           ConnectionFactory connectionFactory = RQconnFactory.getrQconnFactory();
           Connection connection = connectionFactory.newConnection();
           final Channel channel = connection.createChannel();
   
           // 声明一个死信路由
           String dlxExchangeName = "dlx_delay";
           channel.exchangeDeclare(dlxExchangeName , BuiltinExchangeType.TOPIC);
           // 声明一个死信订单队列
           String dlxQueueName = "dlx_delay_order";
           channel.queueDeclare( dlxQueueName ,true , true ,false ,null);
           // 绑定
           channel.queueBind(dlxQueueName , dlxExchangeName , "#");
   
           // 声明一个交换机
           channel.exchangeDeclare(DelayProducer.exchangeName , BuiltinExchangeType.DIRECT);
           // 声明一个延迟订单队列 ,并绑定死信路由器
           String queueName = "delay_order";
           Map<String ,Object> arguments = new HashMap<>();
           arguments.put("x-dead-letter-exchange",dlxExchangeName);
           arguments.put("x-message-ttl", 5 * 1000);  // 5秒过期时间
           channel.queueDeclare(queueName,true , false ,false , arguments );
           // 绑定
           channel.queueBind(queueName , DelayProducer.exchangeName ,DelayProducer.routekey);
   
           // 声明一个消费者（关闭为支付订单的服务）
           final Consumer consumer = new DefaultConsumer(channel){
               @Override
               public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                                          byte[] body) throws IOException {
                   String message = new String(body, "UTF-8");
                   System.out.println("Received and close [" +envelope.getRoutingKey() +"]"+message);
               }
           };
   
           // 消费者消费--队列()
           channel.basicConsume(dlxQueueName,true , consumer);
   
       }
   
   
       public static void main(String[] args) {
           try {
               receive();
           } catch (IOException e) {
               e.printStackTrace();
           } catch (TimeoutException e) {
               e.printStackTrace();
           }
       }
   
   }

   3.得到单例连接工厂

   public class RQconnFactory {
   
       private volatile static ConnectionFactory rQconnFactory;
   
       private RQconnFactory(){
           
       }
   
       public static ConnectionFactory getrQconnFactory() {
           if (rQconnFactory == null) {
               synchronized (RQconnFactory.class){
                   if (rQconnFactory  == null ){
                       rQconnFactory = new ConnectionFactory();
                       rQconnFactory.setHost("192.168.31.220");
                       return rQconnFactory;
                   }
               }
           }
           return rQconnFactory;
       }
   }