訂單超時自動取消的技術方案解析及程式碼實現
2023-07-18 12:02:12
前言
訂單超時自動取消是電商平臺中常見的功能之一，例如在淘寶、京東、拼多多等商城下單後，如果在一定的時間內沒有付款，那麼訂單會自動被取消，是怎麼做到的呢？作為技術人員我們應該瞭解自動取消的原理和實現邏輯，本文將介紹幾種常用的技術方案，幫助開發者實現訂單超時自動取消的功能。
通過以上圖我們可以看到其實超時自動取消的方案有很多，雖然方案多（大多數都是結合延遲佇列來實現的），但每個方案都有自己的優缺點，具體場景需要選用合適的方案。
本文我們主要講解以下幾種常用取消方案，其他方案可自行搜尋研究。
- 方案1：定時輪詢（quartz實現）
- 方案2：JDK延遲佇列DelayQueue
- 方案3：時間輪演演算法（netty的HashedWheelTimer）
- 方案4：Redis
- 方案5：RabbitMQ訊息佇列
方案 1：定時輪詢（quartz實現）
方案描述
通過定時任務的方式去輪詢掃描資料庫表，根據訂單有效期來判斷訂單是否到期，到期則更新訂單狀態。
這裡我們使用quartz作業排程框架來實現定時輪詢。
程式碼
需要新增maven依賴
<dependency>
<groupId>org.quartz-scheduler</groupId>
<artifactId>quartz</artifactId>
<version>2.3.1</version>
</dependency>
程式碼如下：
public class CancelOrderJob implements Job {
@Override
public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {
String time = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
System.out.println(time + ":掃描訂單表超時未付款訂單...");
}
}
public class QuartzJobTest {
public static void main(String[] args) throws Exception {
JobDetail jobDetail = JobBuilder.newJob(CancelOrderJob.class).build();
Trigger trigger = TriggerBuilder.newTrigger()
.withSchedule(SimpleScheduleBuilder.simpleSchedule().withIntervalInSeconds(1).repeatForever())
.build();
Scheduler scheduler = new StdSchedulerFactory().getScheduler();
scheduler.scheduleJob(jobDetail, trigger);
System.out.println("定時任務開啟，每隔1秒執行一次");
scheduler.start();
}
}
執行結果：
定時任務開啟，每隔1秒執行一次
2023-06-27 11:53:43:掃描訂單表超時未付款訂單...
2023-06-27 11:53:43:掃描訂單表超時未付款訂單...
2023-06-27 11:53:44:掃描訂單表超時未付款訂單...
2023-06-27 11:53:45:掃描訂單表超時未付款訂單...
2023-06-27 11:53:46:掃描訂單表超時未付款訂單...
2023-06-27 11:53:47:掃描訂單表超時未付款訂單...
...
優點
這種方案優點是實現簡單，通過quartz框架進行任務排程，無其他依賴，支援叢集部署。
缺點
簡單粗暴的全表掃描方式對資料庫效能影響特別大，可能影響其他正常的業務操作響應時效，另外設定掃描時間間隔也是個問題，設定大了，掃描延遲，影響取消訂單的精準時間，在資料量較大的情況下，設定小了影響資料庫效能，所以需要根據實際情況進行評估。
方案 2：JDK延遲佇列DelayQueue
方案描述
JDK中的DelayQueue可以實現延遲，是一個無界阻塞佇列，其實底層使用的是優先順序佇列PriorityQueue，可以對放入的物件進行排序，物件需要實現Delayed介面，採用阻塞的方式獲取資料，也就是相當於延遲時間到了就會獲取到資料。
程式碼
public class CancelOrder implements Delayed {
private String orderNo;
private long timeout;
CancelOrder(String orderNo, long timeout) {
this.orderNo = orderNo;
this.timeout = timeout + System.nanoTime();
}
public int compareTo(Delayed other) {
if (other == this) {
return 0;
}
CancelOrder t = (CancelOrder) other;
long d = (getDelay(TimeUnit.NANOSECONDS) - t.getDelay(TimeUnit.NANOSECONDS));
return (d == 0) ? 0 : ((d < 0) ? -1 : 1);
}
@Override
public long getDelay(TimeUnit unit) {
return unit.convert(timeout - System.nanoTime(), TimeUnit.NANOSECONDS);
}
@Override
public String toString() {
return "CancelOrder{" +
"orderNo='" + orderNo + '\'' +
", timeout=" + timeout +
'}';
}
}
public class DelayQueueTest {
public static void main(String[] args) throws Exception {
DelayQueue<CancelOrder> queue = new DelayQueue<>();
for (int i = 0; i < 5; i++) {
// 生成訂單，10秒超時
CancelOrder cancelOrder = new CancelOrder("orderNo100" + i, TimeUnit.NANOSECONDS.convert(10, TimeUnit.SECONDS));
String time = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
System.out.println(time + ":生成了訂單，10秒有效期，order：" + cancelOrder);
queue.put(cancelOrder);
// 每1秒生成一個訂單
Thread.sleep(1000);
}
try {
while (!queue.isEmpty()) {
CancelOrder order = queue.take();
String timeout = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
System.out.println(timeout + ":訂單超時，order：" + order);
}
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
}
執行結果：
2023-06-27 18:43:25:生成了訂單，10秒有效期，order：CancelOrder{orderNo='orderNo1000', timeout=1030377584852000}
2023-06-27 18:43:26:生成了訂單，10秒有效期，order：CancelOrder{orderNo='orderNo1001', timeout=1030378653717600}
2023-06-27 18:43:27:生成了訂單，10秒有效期，order：CancelOrder{orderNo='orderNo1002', timeout=1030379654276300}
2023-06-27 18:43:28:生成了訂單，10秒有效期，order：CancelOrder{orderNo='orderNo1003', timeout=1030380655228900}
2023-06-27 18:43:29:生成了訂單，10秒有效期，order：CancelOrder{orderNo='orderNo1004', timeout=1030381656177500}
2023-06-27 18:43:35:訂單超時，order：CancelOrder{orderNo='orderNo1000', timeout=1030377584852000}
2023-06-27 18:43:36:訂單超時，order：CancelOrder{orderNo='orderNo1001', timeout=1030378653717600}
2023-06-27 18:43:37:訂單超時，order：CancelOrder{orderNo='orderNo1002', timeout=1030379654276300}
2023-06-27 18:43:38:訂單超時，order：CancelOrder{orderNo='orderNo1003', timeout=1030380655228900}
2023-06-27 18:43:39:訂單超時，order：CancelOrder{orderNo='orderNo1004', timeout=1030381656177500}
例子中設定的訂單有效期是10秒中，每隔1秒生成一個訂單，目的是為了便於觀察不同的訂單到期時間，可以看到10s後各訂單相繼超時。
優點
不需要任何第三方依賴，實現非常簡單
缺點
資料全部儲存在JVM記憶體中，佔用記憶體，可能會引發記憶體溢位，另外宕機或重啟資料會全部丟失，無法做叢集。
方案 3：時間輪演演算法（netty的HashedWheelTimer）
方案描述
時間輪演演算法用的是一個環形的資料結構（使用陣列實現），每一輪相當於沿著環形走一圈，類似於鐘錶，可以分成很多格子（秒針一圈分成60格，演演算法中叫bucket，這個bucket裡可以存放任務），然後每個格子有持續的時間間隔（比如秒針一個格子是1秒，也就是走過這一格持續1秒的時間，演演算法中對應的是tickDuration）。
時間輪演演算法有多種實現，單輪演演算法，多輪演演算法（相當於在單輪上做了迴圈），分層時間輪演演算法（類似於水錶，有多個錶盤共同計算出總水量）
時間輪演演算法使用一個worker執行緒，將任務放到計算獲得的bucket裡，並按指定的時間間隔tickDuration去執行bucket的時間到期任務。
netty4版本中的時間輪結構如下：
圖中HashedWheelTimer內部儲存使用的是HashedWheelBucket陣列，形成一個環形結構，每一個HashedWheelBucket中儲存的是HashedWheelTimeout雙向連結串列，在HashedWheelTimeout中存的是TimerTask，就是具體要執行的任務。
假設當前指標指在3上，如有有一個任務2秒後執行，那麼會存在5的格子上，如果有一個任務8秒後執行，則會放到3上，轉了一圈，這是任務的輪數就加了1。
程式碼
需要新增maven依賴
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.24.Final</version>
</dependency>
public class CancelOrderTimerTask implements TimerTask {
private String orderNo;
public CancelOrderTimerTask(String orderNo) {
this.orderNo = orderNo;
}
@Override
public void run(Timeout timeout) throws Exception {
String time = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
System.out.println(time + "：處理訂單超時，orderNo：" + orderNo);
}
}
public class HashedWheelTimerTest {
public static void main(String[] argv) {
/*
此處使用的HashedWheelTimer構造方法引數解釋如下：
threadFactory：建立執行緒的工廠
tickDuration：時間間隔，這裡的1，結合後面的TimeUnit.SECONDS,就是走完一格需要1秒時間。
unit：時間單位，這是裡秒
ticksPerWheel：表示陣列的大小，也就是格子的多少
*/
HashedWheelTimer hashedWheelTimer = new HashedWheelTimer(new DefaultThreadFactory("test-thread"), 1, TimeUnit.SECONDS, 60);
hashedWheelTimer.start();
CancelOrderTimerTask timerTask0 = new CancelOrderTimerTask("orderNo1000");
CancelOrderTimerTask timerTask1 = new CancelOrderTimerTask("orderNo1001");
CancelOrderTimerTask timerTask2 = new CancelOrderTimerTask("orderNo1002");
CancelOrderTimerTask timerTask3 = new CancelOrderTimerTask("orderNo1003");
hashedWheelTimer.newTimeout(timerTask0, 0, TimeUnit.SECONDS);
hashedWheelTimer.newTimeout(timerTask1, 5, TimeUnit.SECONDS);
hashedWheelTimer.newTimeout(timerTask2, 30, TimeUnit.SECONDS);
hashedWheelTimer.newTimeout(timerTask3, 70, TimeUnit.SECONDS);
}
}
執行結果：
2023-06-28 11:43:42：處理訂單超時，orderNo：orderNo1000
2023-06-28 11:43:47：處理訂單超時，orderNo：orderNo1001
2023-06-28 11:44:12：處理訂單超時，orderNo：orderNo1002
2023-06-28 11:44:52：處理訂單超時，orderNo：orderNo1003
任務均是按指定時間間隔執行的。
優點
精度靈活可控制，執行效率高，延遲時間比DelayQueue佇列低。
缺點
同DelayQueue一樣，資料全部儲存在JVM記憶體中，佔用記憶體，可能會引發記憶體溢位，另外宕機或重啟資料會全部丟失，巨量資料量的情況下也會影響延遲精度。
方案 4：Redis
redis有兩種方案可以實現延遲，一種是採用輪詢有序集合zset，一種是採用key過期監聽
方案4.1：定時任務輪詢有序集合zset
方案描述
zset是一個有序集合，儲存的每個元素都有個score分值，可以把score當做過期時間，按照score排序（預設按照score從小到大排序，降序可使用zrerange命令），再結合使用一個執行緒輪詢該集合即可實現延遲功能。
程式碼
需要新增maven依賴
<dependency>
<groupId>redis.clients</groupId>
<artifactId>jedis</artifactId>
<version>3.2.0</version>
</dependency>
public class CancelOrderRedisTest {
private static JedisPool jedisPool = new JedisPool("127.0.0.1", 6379);
public static void main(String[] args) throws Exception {
// 放入幾個元素到zset有序集合裡
new Thread(() -> {
try {
for (int i = 0; i < 5; i++) {
String time = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
redisClient().zadd("cancel:order:list", System.currentTimeMillis() + (i + 1) * 1000, "orderNo100" + i);
System.out.println(time + ":生成訂單，訂單號：orderNo100" + i + ",有效期：" + (i + 1) + "秒");
Thread.sleep(1000);
}
} catch (Exception e) {
e.printStackTrace();
}
}).start();
// 開一個執行緒輪詢這個有序集合
new Thread(() -> {
Jedis jedis = redisClient();
while (true) {
Set<Tuple> items = jedis.zrangeWithScores("cancel:order:list", 0, 1);
if (items == null || items.isEmpty()) {
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
} else {
Tuple tuple = (Tuple) items.toArray()[0];
long score = (long) tuple.getScore();
if (System.currentTimeMillis() >= score) {
Long num = jedis.zrem("cancel:order:list", tuple.getElement());
if (num != null && num > 0) {
String time = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
System.out.println(time + ":訂單號：" + tuple.getElement() + "已到期");
}
}
}
}
}).start();
}
/**
* 獲取redis連線
*
* @return
*/
public static Jedis redisClient() {
return jedisPool.getResource();
}
}
執行結果：
2023-06-28 12:41:31:生成訂單，訂單號：orderNo1000,有效期：1秒
2023-06-28 12:41:32:訂單號：orderNo1000已到期
2023-06-28 12:41:32:生成訂單，訂單號：orderNo1001,有效期：2秒
2023-06-28 12:41:33:生成訂單，訂單號：orderNo1002,有效期：3秒
2023-06-28 12:41:34:訂單號：orderNo1001已到期
2023-06-28 12:41:34:生成訂單，訂單號：orderNo1003,有效期：4秒
2023-06-28 12:41:35:生成訂單，訂單號：orderNo1004,有效期：5秒
2023-06-28 12:41:36:訂單號：orderNo1002已到期
2023-06-28 12:41:38:訂單號：orderNo1003已到期
2023-06-28 12:41:40:訂單號：orderNo1004已到期
優點
實現簡單，redis記憶體操作，速度快，效能高，叢集擴充套件方便，可儲存大量訂單資料，持久化機制使得故障時通過AOF或RDB方式恢復，適合對延遲精度要求不高的業務場景
缺點
輪詢執行緒如果不帶休眠或休眠時間短，可能導致空輪詢，CPU飆高，帶休眠時間，休眠多久不好評估，休眠時間過長可能導致延遲不準確。另外處理訊息異常時可能要實現重試機制，還有一個就是可靠性問題，比如是先刪資料在處理訂單還是先處理訂單再刪除資料，處理異常時可能會導致資料丟失。
方案4.2：Redis key過期監聽
方案描述
過期監聽機制是redis在2.8版本以上提供的功能，如果key失效後，redis會給使用者端傳送訊息即pub/sub機制，從而實現延遲方案。
以windows系統的redis為例。
在redis安裝目錄的redis.windows.conf檔案中找到「notify-keyspace-events」如果被註釋則放開，將這行配成如下所示：
notify-keyspace-events Ex
然後再啟動（或重啟）redis。
注意：windows系統下，直接使用redis-server.exe，不會載入redis.windows.conf這個組態檔，需要用命令列啟動。命令列進入redis安裝目錄，執行命令：redis-server.exe redis.windows.conf
程式碼
public class RedisKeyExpireTest {
private static JedisPool jedisPool = new JedisPool("127.0.0.1", 6379);
public static void main(String[] args) throws InterruptedException {
// subscribe方法會阻塞等待，用非同步去初始化訂閱監聽事件
new Thread(() -> {
jedisPool.getResource().subscribe(new RedisSub(), "__keyevent@0__:expired");
}).start();
// 新增幾個帶過期時間的key
new Thread(() -> {
try {
for (int i = 0; i < 5; i++) {
String time = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
jedisPool.getResource().setex("orderNo100" + i, i + 1, "orderNo100" + i);
System.out.println(time + ":生成訂單，訂單號：orderNo100" + i + ",有效期：" + (i + 1) + "秒");
Thread.sleep(1000);
}
} catch (Exception e) {
e.printStackTrace();
}
}).start();
}
}
class RedisSub extends JedisPubSub {
@Override
public void onMessage(String channel, String message) {
String time = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
System.out.println(time + ":訂單號：" + message + "已到期");
}
}
執行結果：
2023-06-28 12:56:17:生成訂單，訂單號：orderNo1000,有效期：1秒
2023-06-28 12:56:18:生成訂單，訂單號：orderNo1001,有效期：2秒
2023-06-28 12:56:18:訂單號：orderNo1000已到期
2023-06-28 12:56:19:生成訂單，訂單號：orderNo1002,有效期：3秒
2023-06-28 12:56:20:生成訂單，訂單號：orderNo1003,有效期：4秒
2023-06-28 12:56:20:訂單號：orderNo1001已到期
2023-06-28 12:56:21:生成訂單，訂單號：orderNo1004,有效期：5秒
2023-06-28 12:56:22:訂單號：orderNo1002已到期
2023-06-28 12:56:24:訂單號：orderNo1003已到期
2023-06-28 12:56:26:訂單號：orderNo1004已到期
通過redis的key過期監聽實現了延遲功能，需要開啟redis伺服器的key過期監聽設定。
優點
實現簡單，redis記憶體操作，速度快，效能高，叢集擴充套件方便，可儲存大量訂單資料，持久化機制使得故障時通過AOF或RDB方式恢復，適合對延遲精度要求不高的業務場景
缺點
redis的key過期有惰性清除和定時清除兩種策略，可能會存在延遲時間不精確的問題，另外redis的pub/sub 機制是不可靠的，如果使用者端故障或重啟期間有key過期則過期通知事件的資料就丟失了，從而訂單無法過期，可以通過補償機制配合使用，定時任務去做輪詢補償。
方案 5：RabbitMQ訊息佇列
方案5.1：訊息TTL+死信佇列
RabbitMQ 可以針對 Queue 和 Message 設定 x-message-tt，來控制訊息的生存時間，如果超時，則訊息變為 dead letter
RabbitMQ 的 Queue 可以設定 x-dead-letter-exchange 和 x-dead-letter-routing-key（可選）兩個引數，用來控制佇列內出現了 deadletter，則按照這兩個引數重新路由。結合以上兩個特性，就可以模擬出延遲訊息的功能。
方案5.2：RabbitMQ延遲佇列外掛
在這裡下載RabbitMQ對應的外掛：https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases，本文使用的RabbitMQ版本是3.8.17，所以找到對應的版本下載的是ez結尾的檔案，直接放到RabbitMQ的外掛目錄plugins即可，位置：/usr/lib/rabbitmq/lib/rabbitmq_server-3.8.17/plugins，然後執行下面的命令使外掛生效（若不生效，需要重啟RabbitMQ）。
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
檢視是否生效（rabbitmq-plugins list命令是檢視所有的外掛）：
rabbitmq-plugins list | grep delayed
顯示如下表示已啟動。
程式碼
採用springboot整合rabbitMQ實現，程式碼如下：
- 設定類
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
import java.util.Map;
@Configuration
public class RabbitMQConfig {
/**
* 正常交換機名稱
*/
public static final String EXCHANGE_NAME_NORMAL_DIRECT = "exchange.normal.direct";
/**
* 訊息不帶ttl佇列名稱
*/
public static final String QUEUE_NAME_WITHOUT_TTL = "queue.without.ttl";
/**
* 訊息帶ttl佇列名稱
*/
public static final String QUEUE_NAME_WITH_TTL = "queue.with.ttl";
/**
* 訊息不帶ttl訊息路由
*/
public static final String ROUTING_KEY_WITHOUT_TTL = "routing.without.ttl.*";
/**
* 訊息帶ttl訊息路由
*/
public static final String ROUTING_KEY_WITH_TTL = "routing.with.ttl.*";
/**
* 死信交換機名稱
*/
public static final String EXCHANGE_NAME_DEADLETTER_DIRECT = "exchange.deadLetter.direct";
/**
* 死信佇列名稱
*/
public static final String QUEUE_NAME_DEAD_LETTER = "queue.deadLetter";
/**
* 死信佇列訊息路由
*/
public static final String ROUTING_KEY_DEAD_LETTER = "routing.deadletter.*";
/**
* rabbitMQ外掛實現的延遲佇列-佇列名稱
*/
public static final String QUEUE_NAME_PLUGIN = "queue.plugin";
/**
* rabbitMQ外掛實現的延遲佇列-交換機名稱
*/
public static final String EXCHANGE_NAME_PLUGIN = "exchange.customexchange.plugin";
/**
* rabbitMQ外掛實現的延遲佇列-路由名稱
*/
public static final String ROUTING_KEY_PLUGIN = "routing.plugin.*";
/**
* 正常交換機
*
* @return
*/
@Bean("normalExchange")
public DirectExchange normalExchange() {
return new DirectExchange(EXCHANGE_NAME_NORMAL_DIRECT);
}
/**
* 死信交換機
*
* @return
*/
@Bean("deadLetterExchange")
public DirectExchange deadLetterExchange() {
return new DirectExchange(EXCHANGE_NAME_DEADLETTER_DIRECT);
}
/**
* rabbitMQ外掛實現的延遲佇列-自定義的交換機
*
* @return
*/
@Bean
public CustomExchange customExchange() {
Map<String, Object> args = new HashMap<>();
args.put("x-delayed-type", "direct");
return new CustomExchange(EXCHANGE_NAME_PLUGIN, "x-delayed-message", true, false, args);
}
/**
* 訊息不帶ttl佇列並設定死信交換機
*
* @return
*/
@Bean("withOutttlQueue")
public Queue withOutttlQueue() {
Map<String, Object> args = new HashMap<>(3);
args.put("x-dead-letter-exchange", EXCHANGE_NAME_DEADLETTER_DIRECT);
args.put("x-dead-letter-routing-key", ROUTING_KEY_DEAD_LETTER);
args.put("x-message-ttl", 10000);
return QueueBuilder.durable(QUEUE_NAME_WITHOUT_TTL).withArguments(args).build();
}
/**
* 訊息帶ttl佇列並設定死信交換機
*
* @return
*/
@Bean("withttlQueue")
public Queue withttlQueue() {
Map<String, Object> args = new HashMap<>(3);
args.put("x-dead-letter-exchange", EXCHANGE_NAME_DEADLETTER_DIRECT);
args.put("x-dead-letter-routing-key", ROUTING_KEY_DEAD_LETTER);
return QueueBuilder.durable(QUEUE_NAME_WITH_TTL).withArguments(args).build();
}
/**
* 死信佇列
*
* @return
*/
@Bean("deadLetterQueue")
public Queue deadLetterQueue() {
return new Queue(QUEUE_NAME_DEAD_LETTER);
}
/**
* rabbitMQ外掛實現的延遲佇列-佇列
*
* @return
*/
@Bean
public Queue pluginQueue() {
return new Queue(QUEUE_NAME_PLUGIN);
}
/**
* 訊息不帶ttl佇列與正常交換機繫結
*
* @param queue
* @param exchange
* @return
*/
@Bean
public Binding withoutttlQueueBinding(@Qualifier("withOutttlQueue") Queue queue,
@Qualifier("normalExchange") DirectExchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY_WITHOUT_TTL);
}
/**
* 訊息帶ttl佇列與正常交換機繫結
*
* @param queue
* @param exchange
* @return
*/
@Bean
public Binding withttlQueueBinding(@Qualifier("withttlQueue") Queue queue,
@Qualifier("normalExchange") DirectExchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY_WITH_TTL);
}
/**
* 死信佇列與死信交換機繫結
*
* @param queue
* @param exchange
* @return
*/
@Bean
public Binding deadLetterBinding(@Qualifier("deadLetterQueue") Queue queue,
@Qualifier("deadLetterExchange") DirectExchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY_DEAD_LETTER);
}
/**
* rabbitMQ外掛實現的延遲佇列-佇列繫結交換機
*
* @param queue
* @param customExchange
* @return
*/
@Bean
public Binding pluginBinding(@Qualifier("pluginQueue") Queue queue,
@Qualifier("customExchange") CustomExchange customExchange) {
return BindingBuilder.bind(queue).to(customExchange).with(ROUTING_KEY_PLUGIN).noargs();
}
}
- 訊息生產者
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import static com.star95.springcloud.rabbitmq.cancelorder.RabbitMQConfig.*;
@Component
public class MessageSender {
@Autowired
private RabbitTemplate rabbitTemplate;
/**
* 傳送帶自定義ttl的訊息
*
* @param msg
* @param ttl
*/
public void sendMsgWithTtl(String msg, String ttl) {
MessagePostProcessor messagePostProcessor = message -> {
message.getMessageProperties().setExpiration(ttl);//設定訊息過期時間
return message;
};
rabbitTemplate.convertAndSend(EXCHANGE_NAME_NORMAL_DIRECT, ROUTING_KEY_WITH_TTL, msg, messagePostProcessor);
}
/**
* 傳送公共ttl的訊息
*
* @param msg
*/
public void sendMsgWithOutTtl(String msg) {
rabbitTemplate.convertAndSend(EXCHANGE_NAME_NORMAL_DIRECT, ROUTING_KEY_WITHOUT_TTL, msg);
}
/**
* 使用rabbitmq外掛實現的延遲佇列，傳送帶自定義ttl的訊息
* @param msg
* @param ttl
*/
public void sendMsgWithPlugin(String msg, Integer ttl) {
rabbitTemplate.convertAndSend(EXCHANGE_NAME_PLUGIN, ROUTING_KEY_PLUGIN, msg, a -> {
a.getMessageProperties().setDelay(ttl);
return a;
});
}
}
- 訊息消費者
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import static com.star95.springcloud.rabbitmq.cancelorder.RabbitMQConfig.QUEUE_NAME_DEAD_LETTER;
import static com.star95.springcloud.rabbitmq.cancelorder.RabbitMQConfig.QUEUE_NAME_PLUGIN;
@Slf4j
@Component
public class QueueConsumer {
/**
* 不帶ttl佇列消費訊息
*
* @param message
* @param channel
* @throws IOException
*/
// @RabbitListener(queues = QUEUE_NAME_WITHOUT_TTL)
public void withoutttlQueueReceive(Message message, Channel channel) throws IOException {
String time = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
String msg = new String(message.getBody());
log.info("當前時間：{}，公共ttl佇列消費的訊息內容：{}", time, msg);
channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
}
/**
* 帶ttl佇列消費訊息
*
* @param message
* @param channel
* @throws IOException
*/
// @RabbitListener(queues = QUEUE_NAME_WITH_TTL)
public void withttlQueueReceive(Message message, Channel channel) throws IOException {
String time = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
String msg = new String(message.getBody());
log.info("當前時間：{}，自定義ttl佇列消費的訊息內容：{}", time, msg);
channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
}
/**
* 死信佇列消費訊息
*
* @param message
* @param channel
* @throws IOException
*/
@RabbitListener(queues = QUEUE_NAME_DEAD_LETTER)
public void deadQueueReceive(Message message, Channel channel) throws IOException {
String time = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
String msg = new String(message.getBody());
log.info("當前時間：{}，死信佇列消費的訊息內容：{}", time, msg);
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
}
/**
* 使用rabbitmq外掛實現延遲佇列消費訊息
*
* @param message
* @param channel
* @throws IOException
*/
@RabbitListener(queues = QUEUE_NAME_PLUGIN)
public void pluginQueueReceive(Message message, Channel channel) throws IOException {
String time = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
String msg = new String(message.getBody());
log.info("當前時間：{}，使用rabbitmq外掛實現延遲佇列，消費的訊息內容：{}", time, msg);
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
}
}
- controller傳送訊息入口
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
@Slf4j
@RequestMapping("cancelorder")
@RestController
public class RabbitMQMsgController {
@Autowired
private MessageSender sender;
@RequestMapping("/msgwithttl")
public void msgWithttl(String msg, String ttl) {
String time = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
log.info("當前時間：{}，建立帶自定義ttl訊息，msg:{}，ttl:{}", time, msg, ttl);
sender.sendMsgWithTtl(msg, ttl);
}
@RequestMapping("/msgwithoutttl")
public void msgWithoutttl(String msg) {
String time = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
log.info("當前時間：{}，建立公共ttl訊息，msg:{}", time, msg);
sender.sendMsgWithOutTtl(msg);
}
@RequestMapping("msgwithplugin")
public void msgWithPlugin(String msg, Integer ttl) {
String time = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
log.info("當前時間：{}，使用rabbitmq外掛實現延遲佇列，傳送的訊息，msg:{}，ttl:{}", time, msg, ttl);
sender.sendMsgWithPlugin(msg, ttl);
}
}
注意：@RabbitListener(queues = QUEUE_NAME_WITHOUT_TTL)，@RabbitListener(queues = QUEUE_NAME_WITH_TTL)，這兩個是註釋掉的，如果不註釋訊息就被正常消費了，註釋掉就可以測試訊息到期會進入死信佇列實現延遲佇列的功能。
- 使用公共ttl的介面傳送兩條訊息
http://127.0.0.1:8080/cancelorder/msgwithoutttl?msg=order123
http://127.0.0.1:8080/cancelorder/msgwithoutttl?msg=order321
結果：
當前時間：2023-06-27 14:45:58，建立公共ttl訊息，msg:order123
當前時間：2023-06-27 14:45:00，建立公共ttl訊息，msg:order321
當前時間：2023-06-27 14:45:08，死信佇列消費的訊息內容：order123
當前時間：2023-06-27 14:45:10，死信佇列消費的訊息內容：order321
不管是兩條請求誰先執行，結果均是在預設的10秒後過期，結果正常，這種適用於訊息具有同一過期時間的場景。
- 使用自定義ttl的介面傳送兩條訊息
http://127.0.0.1:8080/cancelorder/msgwithttl?msg=order321&ttl=5000（先執行）
http://127.0.0.1:8080/cancelorder/msgwithttl?msg=order123&ttl=20000（後執行）
結果：
當前時間：2023-06-27 14:51:40，建立帶自定義ttl訊息，msg:order321，ttl:5000
當前時間：2023-06-27 14:51:45，死信佇列消費的訊息內容：order321
當前時間：2023-06-27 14:51:52，建立帶自定義ttl訊息，msg:order123，ttl:20000
當前時間：2023-06-27 14:52:12，死信佇列消費的訊息內容：order123
這裡可以看到先執行延遲5秒的請求，再執行延遲20秒的請求，結果是正常按指定時間消費的訊息。
如果我們按下面這個順序執行（先執行延遲20秒的請求，再執行延遲5秒的請求）
http://127.0.0.1:8080/cancelorder/msgwithttl?msg=order123&ttl=20000（先執行）
http://127.0.0.1:8080/cancelorder/msgwithttl?msg=order321&ttl=5000（後執行）
結果：
當前時間：2023-06-27 14:53:27，建立帶自定義ttl訊息，msg:order123，ttl:20000
當前時間：2023-06-27 14:53:31，建立帶自定義ttl訊息，msg:order321，ttl:5000
當前時間：2023-06-27 14:53:47，死信佇列消費的訊息內容：order123
當前時間：2023-06-27 14:53:47，死信佇列消費的訊息內容：order321
這裡兩條請求執行完後，結果卻是同一時間消費的訊息，這是因為RabbitMQ只會檢查第一個訊息是否過期，如果過期則丟到死信佇列，如果第一個訊息的延時時長很長，而第二個訊息的延時時長很短，則第二個訊息並不會優先得到執行，這個問題可以通過rabbMQ的延遲外掛來解決。
- 使用rabbitMQ外掛實現的延遲佇列介面傳送兩條訊息
http://127.0.0.1:8080/cancelorder/msgwithplugin?msg=order123&ttl=20000（先執行）
http://127.0.0.1:8080/cancelorder/msgwithplugin?msg=order321&ttl=5000（後執行）
結果：
當前時間：2023-06-27 15:08:04，使用rabbitmq外掛實現延遲佇列，傳送的訊息，msg:order123，ttl:20000
當前時間：2023-06-27 15:08:06，使用rabbitmq外掛實現延遲佇列，傳送的訊息，msg:order321，ttl:5000
當前時間：2023-06-27 15:08:11，使用rabbitmq外掛實現延遲佇列，消費的訊息內容：order321
當前時間：2023-06-27 15:08:24，使用rabbitmq外掛實現延遲佇列，消費的訊息內容：order123
這裡訊息過期正常被消費，解決了由於訊息過期時長不一致導致的不能及時消費的問題。
優點
RabbitMQ訊息服務可靠性高，訊息處理速度快，支援巨量資料量，並且支援分散式橫向擴充套件方便。
缺點
引入RabbitMQ中介軟體系統複雜度增高，運維成本增加，使用起來設定較複雜。
總結
訂單超時自動取消是電商平臺中非常重要的功能之一，通過合適的技術方案，可以實現自動化處理訂單超時的邏輯，提升使用者體驗和系統效率。本文介紹了幾種常用的技術方案，開發者可以根據具體的業務需求和技術棧選擇合適的方案，並結合相應的檔案和範例進行實現和設定。