“

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

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

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

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

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

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

下面，我們以判斷訂單是否超時為例，進行方案分析。

方案分析

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

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

實現(xiàn)：博主當年早期是用 quartz 來實現(xiàn)的（實習那會的事），簡單介紹一下。

maven 項目引入一個依賴，如下所示：

<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ù)庫掃描啦。。。");}public?static?void?main(String[]?args)?throws?Exception?{//?創(chuàng)建任務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();//?將任務及其觸發(fā)器放入調(diào)度器scheduler.scheduleJob(jobDetail,?trigger);//?調(diào)度器開始調(diào)度任務scheduler.start();}}

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

優(yōu)缺點：

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

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

②JDK 的延遲隊列

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

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

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

實現(xiàn)：定義一個類 OrderDelay 實現(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);}//?返回距離你自定義的超時時間還有多少public?long?getDelay(TimeUnit?unit)?{return?unit.convert(timeout?-?System.nanoTime(),TimeUnit.NANOSECONDS);}void?print()?{System.out.println(orderId+"編號的訂單要刪除啦。。。。");}}

運行的測試 Demo 為，我們設定延遲時間為 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編號的訂單要刪除啦。。。。After?3003?MilliSeconds00000002編號的訂單要刪除啦。。。。After?6006?MilliSeconds00000003編號的訂單要刪除啦。。。。After?9006?MilliSeconds00000004編號的訂單要刪除啦。。。。After?12008?MilliSeconds00000005編號的訂單要刪除啦。。。。After?15009?MilliSeconds

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

優(yōu)缺點：

• 優(yōu)點：效率高，任務觸發(fā)時間延遲低。

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

③時間輪算法

思路：先上一張時間輪的圖。

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

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

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

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

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

實現(xiàn)：我們用 Netty 的 HashedWheelTimer 來實現(xiàn)。

給 Pom 加上下面的依賴：

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

測試代碼 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ù)庫刪除訂單了。。。。");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ù)庫刪除訂單了。。。。6秒過去了

優(yōu)缺點：

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

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

④Redis 緩存

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

zset 常用命令：

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

• 按順序查詢元素：ZRANGE key start stop [WITHSCORES]

• 查詢元素 score：ZSCORE key member

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

測試如下：

添加單個元素redis>?ZADD?page_rank?10?google.com(integer)?1添加多個元素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"查詢元素的score值redis>?ZSCORE?page_rank?bing.com"8"移除單個元素redis>?ZREM?page_rank?google.com(integer)?1redis>?ZRANGE?page_rank?0?-1?WITHSCORES1)?"bing.com"2)?"8"3)?"baidu.com"4)?"9"

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

具體如下圖所示：

實現(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個訂單放進去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生成了一個訂單任務：訂單ID為"+"OID0000001"+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("當前沒有等待的任務");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消費了一個任務：消費的訂單OrderId為"+orderId);}}}public?static?void?main(String[]?args)?{AppTest?appTest?=new?AppTest();appTest.productionDelayMessage();appTest.consumerDelayMessage();}}

此時對應輸出如下：

可以看到，幾乎都是 3 秒之后，消費訂單。

然而，這一版存在一個致命的硬傷，在高并發(fā)條件下，多消費者會取到同一個訂單號，我們上測試代碼 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)了多個線程消費同一個資源的情況。

解決方案：

（1）用分布式鎖，但是用分布式鎖，性能下降了，該方案不細說。

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

if(nowSecond?>=?score){String?orderId?=?((Tuple)items.toArray()[0]).getElement();jedis.zrem("OrderId",?orderId);System.out.println(System.currentTimeMillis()+"ms:redis消費了一個任務：消費的訂單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消費了一個任務：消費的訂單OrderId為"+orderId);}}

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

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

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

notify-keyspace-events?Ex

運行代碼如下：

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 機制存在一個硬傷，官網(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）模式的，因此無法實現(xiàn)事件的可靠通知。也就是說，如果發(fā)布/訂閱的客戶端斷鏈之后又重連，則在客戶端斷鏈期間的所有事件都丟失了。因此，方案二不是太推薦。當然，如果你對可靠性要求不高，可以使用。

優(yōu)缺點：

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

• 缺點：需要額外進行 Redis 維護。

⑤使用消息隊列

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

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

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

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

優(yōu)缺點：

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

• 缺點：本身的易用度要依賴于 RabbitMQ 的運維，因為要引用 RabbitMQ，所以復雜度和成本變高。

作者：hjm4702192

編輯：陶家龍

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

精彩文章推薦：

美團技術：到店結(jié)算平臺實踐（膠片）

美團技術：到店供應鏈及商品平臺建設實踐（膠片）

美團技術：CRM平臺建設實踐（膠片）

美團技術：交易平臺建設實踐（膠片）

中生代技術

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

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

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

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

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

總結(jié)

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

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