“

在開發(fā)中，往往會(huì)遇到一些關(guān)于延時(shí)任務(wù)的需求。例如：生成訂單 30 分鐘未支付，則自動(dòng)取消；生成訂單 60 秒后，給用戶發(fā)短信。

對(duì)上述的任務(wù)，我們給一個(gè)專業(yè)的名字來(lái)形容，那就是延時(shí)任務(wù)。那么這里就會(huì)產(chǎn)生一個(gè)問題，這個(gè)延時(shí)任務(wù)和定時(shí)任務(wù)的區(qū)別究竟在哪里呢？

一共有如下幾點(diǎn)區(qū)別：

• 定時(shí)任務(wù)有明確的觸發(fā)時(shí)間，延時(shí)任務(wù)沒有

• 定時(shí)任務(wù)有執(zhí)行周期，而延時(shí)任務(wù)在某事件觸發(fā)后一段時(shí)間內(nèi)執(zhí)行，沒有執(zhí)行周期

• 定時(shí)任務(wù)一般執(zhí)行的是批處理操作是多個(gè)任務(wù)，而延時(shí)任務(wù)一般是單個(gè)任務(wù)

下面，我們以判斷訂單是否超時(shí)為例，進(jìn)行方案分析。

方案分析

①數(shù)據(jù)庫(kù)輪詢

思路：該方案通常是在小型項(xiàng)目中使用，即通過一個(gè)線程定時(shí)的去掃描數(shù)據(jù)庫(kù)，通過訂單時(shí)間來(lái)判斷是否有超時(shí)的訂單，然后進(jìn)行 update 或 delete 等操作。

實(shí)現(xiàn)：博主當(dāng)年早期是用 quartz 來(lái)實(shí)現(xiàn)的（實(shí)習(xí)那會(huì)的事），簡(jiǎn)單介紹一下。

maven 項(xiàng)目引入一個(gè)依賴，如下所示：

<dependency><groupId>org.quartz-scheduler</groupId><artifactId>quartz</artifactId><version>2.2.2</version></dependency>

調(diào)用 Demo 類 MyJob，如下所示：

package?com.rjzheng.delay1;import?org.quartz.JobBuilder;import?org.quartz.JobDetail;import?org.quartz.Scheduler;import?org.quartz.SchedulerException;import?org.quartz.SchedulerFactory;import?org.quartz.SimpleScheduleBuilder;import?org.quartz.Trigger;import?org.quartz.TriggerBuilder;import?org.quartz.impl.StdSchedulerFactory;import?org.quartz.Job;import?org.quartz.JobExecutionContext;import?org.quartz.JobExecutionException;public?class?MyJob?implements?Job?{public?void?execute(JobExecutionContext?context)throws?JobExecutionException?{System.out.println("要去數(shù)據(jù)庫(kù)掃描啦。。。");}public?static?void?main(String[]?args)?throws?Exception?{//?創(chuàng)建任務(wù)JobDetail?jobDetail?=?JobBuilder.newJob(MyJob.class).withIdentity("job1",?"group1").build();//?創(chuàng)建觸發(fā)器?每3秒鐘執(zhí)行一次Trigger?trigger?=?TriggerBuilder.newTrigger().withIdentity("trigger1",?"group3").withSchedule(SimpleScheduleBuilder.simpleSchedule().withIntervalInSeconds(3).repeatForever()).build();Scheduler?scheduler?=?new?StdSchedulerFactory().getScheduler();//?將任務(wù)及其觸發(fā)器放入調(diào)度器scheduler.scheduleJob(jobDetail,?trigger);//?調(diào)度器開始調(diào)度任務(wù)scheduler.start();}}

運(yùn)行代碼，可發(fā)現(xiàn)每隔 3 秒，輸出如下：要去數(shù)據(jù)庫(kù)掃描啦！

優(yōu)缺點(diǎn)：

• 優(yōu)點(diǎn)：簡(jiǎn)單易行，支持集群操作

• 缺點(diǎn)：對(duì)服務(wù)器內(nèi)存消耗大；存在延遲，比如你每隔 3 分鐘掃描一次，那最壞的延遲時(shí)間就是 3 分鐘；假設(shè)你的訂單有幾千萬(wàn)條，每隔幾分鐘這樣掃描一次，數(shù)據(jù)庫(kù)損耗極大。

②JDK 的延遲隊(duì)列

思路：該方案是利用 JDK 自帶的 DelayQueue 來(lái)實(shí)現(xiàn)，這是一個(gè)無(wú)界阻塞隊(duì)列，該隊(duì)列只有在延遲期滿的時(shí)候才能從中獲取元素，放入 DelayQueue 中的對(duì)象，是必須實(shí)現(xiàn) Delayed 接口的。

DelayedQueue 實(shí)現(xiàn)工作流程如下圖所示：

其中 Poll()：獲取并移除隊(duì)列的超時(shí)元素，沒有則返回空。take()：獲取并移除隊(duì)列的超時(shí)元素，如果沒有則 wait 當(dāng)前線程，直到有元素滿足超時(shí)條件，返回結(jié)果。

實(shí)現(xiàn)：定義一個(gè)類 OrderDelay 實(shí)現(xiàn) Delayed。

代碼如下：

package?com.rjzheng.delay2;import?java.util.concurrent.Delayed;import?java.util.concurrent.TimeUnit;public?class?OrderDelay?implements?Delayed?{private?String?orderId;private?long?timeout;OrderDelay(String?orderId,?long?timeout)?{this.orderId?=?orderId;this.timeout?=?timeout?+?System.nanoTime();}public?int?compareTo(Delayed?other)?{if?(other?==?this)return?0;OrderDelay?t?=?(OrderDelay)?other;long?d?=?(getDelay(TimeUnit.NANOSECONDS)?-?t.getDelay(TimeUnit.NANOSECONDS));return?(d?==?0)???0?:?((d?<?0)???-1?:?1);}//?返回距離你自定義的超時(shí)時(shí)間還有多少public?long?getDelay(TimeUnit?unit)?{return?unit.convert(timeout?-?System.nanoTime(),TimeUnit.NANOSECONDS);}void?print()?{System.out.println(orderId+"編號(hào)的訂單要?jiǎng)h除啦。。。。");}}

運(yùn)行的測(cè)試 Demo 為，我們?cè)O(shè)定延遲時(shí)間為 3 秒：

package?com.rjzheng.delay2;import?java.util.ArrayList;import?java.util.List;import?java.util.concurrent.DelayQueue;import?java.util.concurrent.TimeUnit;public?class?DelayQueueDemo?{public?static?void?main(String[]?args)?{??//?TODO?Auto-generated?method?stub??List<String>?list?=?new?ArrayList<String>();??list.add("00000001");??list.add("00000002");??list.add("00000003");??list.add("00000004");??list.add("00000005");??DelayQueue<OrderDelay>?queue?=?newDelayQueue<OrderDelay>();??long?start?=?System.currentTimeMillis();??for(int?i?=?0;i<5;i++){??//延遲三秒取出queue.put(new?OrderDelay(list.get(i),??TimeUnit.NANOSECONDS.convert(3,TimeUnit.SECONDS)));??try?{??queue.take().print();??System.out.println("After?"?+??(System.currentTimeMillis()-start)?+?"?MilliSeconds");??}?catch?(InterruptedException?e)?{??//?TODO?Auto-generated?catch?block??e.printStackTrace();??}??}??}??}

輸出如下：

00000001編號(hào)的訂單要?jiǎng)h除啦。。。。After?3003?MilliSeconds00000002編號(hào)的訂單要?jiǎng)h除啦。。。。After?6006?MilliSeconds00000003編號(hào)的訂單要?jiǎng)h除啦。。。。After?9006?MilliSeconds00000004編號(hào)的訂單要?jiǎng)h除啦。。。。After?12008?MilliSeconds00000005編號(hào)的訂單要?jiǎng)h除啦。。。。After?15009?MilliSeconds

可以看到都是延遲 3 秒，訂單被刪除。

優(yōu)缺點(diǎn)：

• 優(yōu)點(diǎn)：效率高，任務(wù)觸發(fā)時(shí)間延遲低。

• 缺點(diǎn)：服務(wù)器重啟后，數(shù)據(jù)全部消失，怕宕機(jī)；集群擴(kuò)展相當(dāng)麻煩；因?yàn)閮?nèi)存條件限制的原因，比如下單未付款的訂單數(shù)太多，那么很容易就出現(xiàn) OOM 異常；代碼復(fù)雜度較高。

③時(shí)間輪算法

思路：先上一張時(shí)間輪的圖。

時(shí)間輪算法可以類比于時(shí)鐘，如上圖箭頭（指針）按某一個(gè)方向按固定頻率輪動(dòng)，每一次跳動(dòng)稱為一個(gè) tick。

這樣可以看出定時(shí)輪由個(gè) 3 個(gè)重要的屬性參數(shù)，ticksPerWheel（一輪的 tick 數(shù)），tickDuration（一個(gè) tick 的持續(xù)時(shí)間）以及 timeUnit（時(shí)間單位）。

例如當(dāng) ticksPerWheel=60，tickDuration=1，timeUnit=秒，這就和現(xiàn)實(shí)中的始終的秒針走動(dòng)完全類似了。

如果當(dāng)前指針指在 1 上面，我有一個(gè)任務(wù)需要 4 秒以后執(zhí)行，那么這個(gè)執(zhí)行的線程回調(diào)或者消息將會(huì)被放在 5 上。

那如果需要在 20 秒之后執(zhí)行怎么辦，由于這個(gè)環(huán)形結(jié)構(gòu)槽數(shù)只到 8，如果要 20 秒，指針需要多轉(zhuǎn) 2 圈。位置是在 2 圈之后的 5 上面（20 % 8 + 1）。

實(shí)現(xiàn)：我們用 Netty 的 HashedWheelTimer 來(lái)實(shí)現(xiàn)。

給 Pom 加上下面的依賴：

<dependency><groupId>io.netty</groupId><artifactId>netty-all</artifactId><version>4.1.24.Final</version></dependency>

測(cè)試代碼 HashedWheelTimerTest，如下所示：

package?com.rjzheng.delay3;import?io.netty.util.HashedWheelTimer;import?io.netty.util.Timeout;import?io.netty.util.Timer;import?io.netty.util.TimerTask;import?java.util.concurrent.TimeUnit;public?class?HashedWheelTimerTest?{static?class?MyTimerTask?implements?TimerTask{boolean?flag;public?MyTimerTask(boolean?flag){this.flag?=?flag;}public?void?run(Timeout?timeout)?throws?Exception?{//?TODO?Auto-generated?method?stubSystem.out.println("要去數(shù)據(jù)庫(kù)刪除訂單了。。。。");this.flag?=false;}}public?static?void?main(String[]?argv)?{MyTimerTask?timerTask?=?new?MyTimerTask(true);Timer?timer?=?new?HashedWheelTimer();timer.newTimeout(timerTask,?5,?TimeUnit.SECONDS);int?i?=?1;while(timerTask.flag){try?{Thread.sleep(1000);}?catch?(InterruptedException?e)?{//?TODO?Auto-generated?catch?blocke.printStackTrace();}System.out.println(i+"秒過去了");i++;}}}

輸出如下：

1秒過去了2秒過去了3秒過去了4秒過去了5秒過去了要去數(shù)據(jù)庫(kù)刪除訂單了。。。。6秒過去了

優(yōu)缺點(diǎn)：

• 優(yōu)點(diǎn)：效率高，任務(wù)觸發(fā)時(shí)間延遲時(shí)間比 delayQueue 低，代碼復(fù)雜度比 delayQueue 低。

• 缺點(diǎn)：服務(wù)器重啟后，數(shù)據(jù)全部消失，怕宕機(jī)；集群擴(kuò)展相當(dāng)麻煩；因?yàn)閮?nèi)存條件限制的原因，比如下單未付款的訂單數(shù)太多，那么很容易就出現(xiàn) OOM 異常。

④Redis 緩存

思路一：利用 Redis 的 zset。zset 是一個(gè)有序集合，每一個(gè)元素（member）都關(guān)聯(lián)了一個(gè) score，通過 score 排序來(lái)取集合中的值。

zset 常用命令：

• 添加元素：ZADD key score member [[score member] [score member] …]

• 按順序查詢?cè)?#xff1a;ZRANGE key start stop [WITHSCORES]

• 查詢?cè)?score：ZSCORE key member

• 移除元素：ZREM key member [member …]

測(cè)試如下：

添加單個(gè)元素redis>?ZADD?page_rank?10?google.com(integer)?1添加多個(gè)元素redis>?ZADD?page_rank?9?baidu.com?8?bing.com(integer)?2redis>?ZRANGE?page_rank?0?-1?WITHSCORES1)?"bing.com"2)?"8"3)?"baidu.com"4)?"9"5)?"google.com"6)?"10"查詢?cè)氐膕core值redis>?ZSCORE?page_rank?bing.com"8"移除單個(gè)元素redis>?ZREM?page_rank?google.com(integer)?1redis>?ZRANGE?page_rank?0?-1?WITHSCORES1)?"bing.com"2)?"8"3)?"baidu.com"4)?"9"

那么如何實(shí)現(xiàn)呢？我們將訂單超時(shí)時(shí)間戳與訂單號(hào)分別設(shè)置為 score 和 member，系統(tǒng)掃描第一個(gè)元素判斷是否超時(shí)。

具體如下圖所示：

實(shí)現(xiàn)一：

package?com.rjzheng.delay4;import?java.util.Calendar;import?java.util.Set;import?redis.clients.jedis.Jedis;import?redis.clients.jedis.JedisPool;import?redis.clients.jedis.Tuple;public?class?AppTest?{private?static?final?String?ADDR?=?"127.0.0.1";private?static?final?int?PORT?=?6379;private?static?JedisPool?jedisPool?=?new?JedisPool(ADDR,?PORT);public?static?Jedis?getJedis()?{return?jedisPool.getResource();}//生產(chǎn)者,生成5個(gè)訂單放進(jìn)去public?void?productionDelayMessage(){for(int?i=0;i<5;i++){//延遲3秒Calendar?cal1?=?Calendar.getInstance();cal1.add(Calendar.SECOND,?3);int?second3later?=?(int)?(cal1.getTimeInMillis()?/?1000);AppTest.getJedis().zadd("OrderId",second3later,"OID0000001"+i);System.out.println(System.currentTimeMillis()+"ms:redis生成了一個(gè)訂單任務(wù)：訂單ID為"+"OID0000001"+i);}}//消費(fèi)者，取訂單public?void?consumerDelayMessage(){Jedis?jedis?=?AppTest.getJedis();while(true){Set<Tuple>?items?=?jedis.zrangeWithScores("OrderId",?0,?1);if(items?==?null?||?items.isEmpty()){System.out.println("當(dāng)前沒有等待的任務(wù)");try?{Thread.sleep(500);}?catch?(InterruptedException?e)?{//?TODO?Auto-generated?catch?blocke.printStackTrace();}continue;}int??score?=?(int)?((Tuple)items.toArray()[0]).getScore();Calendar?cal?=?Calendar.getInstance();int?nowSecond?=?(int)?(cal.getTimeInMillis()?/?1000);if(nowSecond?>=?score){String?orderId?=?((Tuple)items.toArray()[0]).getElement();jedis.zrem("OrderId",?orderId);System.out.println(System.currentTimeMillis()?+"ms:redis消費(fèi)了一個(gè)任務(wù)：消費(fèi)的訂單OrderId為"+orderId);}}}public?static?void?main(String[]?args)?{AppTest?appTest?=new?AppTest();appTest.productionDelayMessage();appTest.consumerDelayMessage();}}

此時(shí)對(duì)應(yīng)輸出如下：

可以看到，幾乎都是 3 秒之后，消費(fèi)訂單。

然而，這一版存在一個(gè)致命的硬傷，在高并發(fā)條件下，多消費(fèi)者會(huì)取到同一個(gè)訂單號(hào)，我們上測(cè)試代碼 ThreadTest：

package?com.rjzheng.delay4;import?java.util.concurrent.CountDownLatch;public?class?ThreadTest?{private?static?final?int?threadNum?=?10;private?static?CountDownLatch?cdl?=?newCountDownLatch(threadNum);static?class?DelayMessage?implements?Runnable{public?void?run()?{try?{cdl.await();}?catch?(InterruptedException?e)?{//?TODO?Auto-generated?catch?blocke.printStackTrace();}AppTest?appTest?=new?AppTest();appTest.consumerDelayMessage();}}public?static?void?main(String[]?args)?{AppTest?appTest?=new?AppTest();appTest.productionDelayMessage();for(int?i=0;i<threadNum;i++){new?Thread(new?DelayMessage()).start();cdl.countDown();}}}

輸出如下所示：

顯然，出現(xiàn)了多個(gè)線程消費(fèi)同一個(gè)資源的情況。

解決方案：

（1）用分布式鎖，但是用分布式鎖，性能下降了，該方案不細(xì)說(shuō)。

（2）對(duì) ZREM 的返回值進(jìn)行判斷，只有大于 0 的時(shí)候，才消費(fèi)數(shù)據(jù)，于是將 consumerDelayMessage() 方法里的：

if(nowSecond?>=?score){String?orderId?=?((Tuple)items.toArray()[0]).getElement();jedis.zrem("OrderId",?orderId);System.out.println(System.currentTimeMillis()+"ms:redis消費(fèi)了一個(gè)任務(wù)：消費(fèi)的訂單OrderId為"+orderId);}

修改為：

if(nowSecond?>=?score){String?orderId?=?((Tuple)items.toArray()[0]).getElement();Long?num?=?jedis.zrem("OrderId",?orderId);if(?num?!=?null?&&?num>0){System.out.println(System.currentTimeMillis()+"ms:redis消費(fèi)了一個(gè)任務(wù)：消費(fèi)的訂單OrderId為"+orderId);}}

在這種修改后，重新運(yùn)行 ThreadTest 類，發(fā)現(xiàn)輸出正常了。

思路二：該方案使用 Redis 的 Keyspace Notifications。中文翻譯就是鍵空間機(jī)制，就是利用該機(jī)制可以在 key 失效之后，提供一個(gè)回調(diào)，實(shí)際上是 Redis 會(huì)給客戶端發(fā)送一個(gè)消息。是需要 Redis 版本 2.8 以上。

實(shí)現(xiàn)二：在 redis.conf 中，加入一條配置：

notify-keyspace-events?Ex

運(yùn)行代碼如下：

package?com.rjzheng.delay5;import?redis.clients.jedis.Jedis;import?redis.clients.jedis.JedisPool;import?redis.clients.jedis.JedisPubSub;public?class?RedisTest?{private?static?final?String?ADDR?=?"127.0.0.1";private?static?final?int?PORT?=?6379;private?static?JedisPool?jedis?=?new?JedisPool(ADDR,?PORT);private?static?RedisSub?sub?=?new?RedisSub();public?static?void?init()?{new?Thread(new?Runnable()?{public?void?run()?{jedis.getResource().subscribe(sub,?"__keyevent@0__:expired");}}).start();}public?static?void?main(String[]?args)?throws?InterruptedException?{init();for(int?i?=0;i<10;i++){String?orderId?=?"OID000000"+i;jedis.getResource().setex(orderId,?3,?orderId);System.out.println(System.currentTimeMillis()+"ms:"+orderId+"訂單生成");}}static?class?RedisSub?extends?JedisPubSub?{<ahref='http://www.jobbole.com/members/wx610506454'>@Override</a>public?void?onMessage(String?channel,?String?message)?{System.out.println(System.currentTimeMillis()+"ms:"+message+"訂單取消");}}}

輸出如下：

可以明顯看到 3 秒過后，訂單取消了。PS：Redis 的 pub/sub 機(jī)制存在一個(gè)硬傷，官網(wǎng)內(nèi)容如下：

原：Because Redis Pub/Sub is fire and forget currently there is no way to use this feature if your application demands reliable notification of events, that is, if your Pub/Sub client disconnects, and reconnects later, all the events delivered during the time the client was disconnected are lost.

翻：Redis 的發(fā)布/訂閱目前是即發(fā)即棄（fire and forget）模式的，因此無(wú)法實(shí)現(xiàn)事件的可靠通知。也就是說(shuō)，如果發(fā)布/訂閱的客戶端斷鏈之后又重連，則在客戶端斷鏈期間的所有事件都丟失了。因此，方案二不是太推薦。當(dāng)然，如果你對(duì)可靠性要求不高，可以使用。

優(yōu)缺點(diǎn)：

• 優(yōu)點(diǎn)：由于使用 Redis 作為消息通道，消息都存儲(chǔ)在 Redis 中。如果發(fā)送程序或者任務(wù)處理程序掛了，重啟之后，還有重新處理數(shù)據(jù)的可能性；做集群擴(kuò)展相當(dāng)方便；時(shí)間準(zhǔn)確度高。

• 缺點(diǎn)：需要額外進(jìn)行 Redis 維護(hù)。

⑤使用消息隊(duì)列

我們可以采用 RabbitMQ 的延時(shí)隊(duì)列。RabbitMQ 具有以下兩個(gè)特性，可以實(shí)現(xiàn)延遲隊(duì)列。

• RabbitMQ 可以針對(duì) Queue 和 Message 設(shè)置 x-message-tt，來(lái)控制消息的生存時(shí)間，如果超時(shí)，則消息變?yōu)?dead letter。

• lRabbitMQ的 Queue 可以配置 x-dead-letter-exchange 和 x-dead-letter-routing-key（可選）兩個(gè)參數(shù)，用來(lái)控制隊(duì)列內(nèi)出現(xiàn)了 deadletter，則按照這兩個(gè)參數(shù)重新路由。

結(jié)合以上兩個(gè)特性，就可以模擬出延遲消息的功能。

優(yōu)缺點(diǎn)：

• 優(yōu)點(diǎn)：高效，可以利用 RabbitMQ 的分布式特性輕易的進(jìn)行橫向擴(kuò)展,消息支持持久化增加了可靠性。

• 缺點(diǎn)：本身的易用度要依賴于 RabbitMQ 的運(yùn)維，因?yàn)橐?RabbitMQ，所以復(fù)雜度和成本變高。

作者：hjm4702192

編輯：陶家龍

出處：http://adkx.net/w73gf

精彩文章推薦：

美團(tuán)技術(shù)：到店結(jié)算平臺(tái)實(shí)踐（膠片）

美團(tuán)技術(shù)：到店供應(yīng)鏈及商品平臺(tái)建設(shè)實(shí)踐（膠片）

美團(tuán)技術(shù)：CRM平臺(tái)建設(shè)實(shí)踐（膠片）

美團(tuán)技術(shù)：交易平臺(tái)建設(shè)實(shí)踐（膠片）

中生代技術(shù)

推送有營(yíng)養(yǎng)的技術(shù)干貨文章；

總覆蓋會(huì)員100000+人；資深架構(gòu)、總監(jiān)等職位以上3000+人。

定期在線分享超過100期，線下技術(shù)沙龍超過70次、覆蓋20多個(gè)等城市！

關(guān)注技術(shù)架構(gòu)、研發(fā)管理、互聯(lián)網(wǎng)金融、電商、大數(shù)據(jù)、區(qū)塊鏈、人工智能等方向！

加入中生代技術(shù)群聊，投稿等請(qǐng)?zhí)砑影酌魑⑿?#xff1a;zsdwyq，注明姓名、職稱和技術(shù)方向，通過后加入中生代技術(shù)群，和群友們共同學(xué)習(xí)成長(zhǎng)！

總結(jié)

以上是生活随笔為你收集整理的拼夕夕订单超时未支付自动关闭实现方案！的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。