RabbitMQ 延遲任務（限時訂單） 原理 以及代碼 實戰

• 一、場景

　　　　我們經常會碰見，一個需求就是，發送一條指令（消息），延遲一段時間執行，比如說常見的淘寶當下了一個訂單后，訂單支付時間為半個小時，如果半個小時沒有支付，則關閉該訂單。當然實現的方式有幾種，今天來看看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;
       }
   }