1. 程式人生 > >RabbitMQ 延遲任務(限時訂單) 原理 以及程式碼 實戰

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;
        }
    }