RabbitMQ 延遲任務(限時訂單) 原理 以及程式碼 實戰
阿新 • • 發佈:2018-12-09
-
一、場景
我們經常會碰見,一個需求就是,傳送一條指令(訊息),延遲一段時間執行,比如說常見的淘寶當下了一個訂單後,訂單支付時間為半個小時,如果半個小時沒有支付,則關閉該訂單。當然實現的方式有幾種,今天來看看rabbitMQ實現的方式。
-
二、思路:rabbitMQ 如何實現
- rabbitMQ為每個佇列設定訊息的超時時間。只要給佇列設定x-message-ttl 引數,就設定了該佇列所有訊息的存活時間,時間單位是毫秒。如果宣告佇列時指定了死信交換器,則過期訊息會成為死信訊息
-
需要設定的引數為:
-
三、原理:上圖
-
將延遲佇列(queue)在宣告的時候設定引數 “ x-dead-letter-exchange ”,“ x-message-ttl “ 分別對應 死信路由器(dlx_exchange) 和 訊息過期時間(比如說30分鐘)。
-
一個訊息從生產者傳送到延遲佇列 ,在延遲佇列裡等待,等待30分鐘後,會去繫結的死信路由(dlx_exchange)。通過死信路由的規則,走到死信佇列。
- 這時候監聽死信佇列的消費者就可以接收到訊息,消費訊息。比如說檢視該訂單是否支付,如果沒有支付,則關閉該訂單。
-
四、程式碼實戰
- 生產者程式碼:
import
- 消費者程式碼:
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; } }