在開發(fā)中，往往會遇到一些關(guān)于延時任務(wù)的需求。例如

生成訂單30分鐘未支付，則自動取消

生成訂單60秒后，給用戶發(fā)短信

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

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

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

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

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

方案分析

（1）數(shù)據(jù)庫輪詢

思路

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

實現(xiàn)

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

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)建任務(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ù)庫掃描啦。。。

優(yōu)缺點

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

缺點：（1）對服務(wù)器內(nèi)存消耗大

（2）存在延遲，比如你每隔3分鐘掃描一次，那最壞的延遲時間就是3分鐘

（3）假設(shè)你的訂單有幾千萬條，每隔幾分鐘這樣掃描一次，數(shù)據(jù)庫損耗極大

（2）JDK的延遲隊列

思路

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

其中Poll（）：獲取并移除隊列的超時元素，沒有則返回空

take（）：獲取并移除隊列的超時元素，如果沒有則wait當(dāng)前線程，直到有元素滿足超時條件，返回結(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+“編號的訂單要刪除啦。。。。”）;

}

}

運(yùn)行的測試Demo為，我們設(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編號的訂單要刪除啦。。。。

After 3003 MilliSeconds

00000002編號的訂單要刪除啦。。。。

After 6006 MilliSeconds

00000003編號的訂單要刪除啦。。。。

After 9006 MilliSeconds

00000004編號的訂單要刪除啦。。。。

After 12008 MilliSeconds

00000005編號的訂單要刪除啦。。。。

After 15009 MilliSeconds

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

優(yōu)缺點

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

缺點：

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

（3）時間輪算法

思路

先上一張時間輪的圖（這圖到處都是啦）

時間輪算法可以類比于時鐘，如上圖箭頭（指針）按某一個方向按固定頻率輪動，每一次跳動稱為一個 tick。這樣可以看出定時輪由個3個重要的屬性參數(shù)，ticksPerWheel（一輪的tick數(shù)），tickDuration（一個tick的持續(xù)時間）以及 timeUnit（時間單位），例如當(dāng)ticksPerWheel=60，tickDuration=1，timeUnit=秒，這就和現(xiàn)實中的始終的秒針走動完全類似了。

如果當(dāng)前指針指在1上面，我有一個任務(wù)需要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 stub

System.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 block

e.printStackTrace（）;

}

System.out.println（i+“秒過去了”）;

i++;

}

}

}

輸出如下

1秒過去了

2秒過去了

3秒過去了

4秒過去了

5秒過去了

要去數(shù)據(jù)庫刪除訂單了。。。。

6秒過去了

優(yōu)缺點

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

缺點：

（1）服務(wù)器重啟后，數(shù)據(jù)全部消失，怕宕機(jī)

（2）集群擴(kuò)展相當(dāng)麻煩

（3）因為內(nèi)存條件限制的原因，比如下單未付款的訂單數(shù)太多，那么很容易就出現(xiàn)OOM異常

（4）redis緩存

思路一

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

添加元素：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） 2

redis》 ZRANGE page_rank 0 -1 WITHSCORES

1） “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） 1

redis》 ZRANGE page_rank 0 -1 WITHSCORES

1） “bing.com”2） “8”3） “baidu.com”4） “9”

那么如何實現(xiàn)呢？我們將訂單超時時間戳與訂單號分別設(shè)置為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個訂單放進(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生成了一個訂單任務(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 block

e.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)了一個任務(wù)：消費(fèi)的訂單OrderId為”+orderId）;

}

}

}

public static void main（String［］ args） {

AppTest appTest =new AppTest（）;

appTest.productionDelayMessage（）;

appTest.consumerDelayMessage（）;

}

}

此時對應(yīng)輸出如下

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

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

e.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)了多個線程消費(fèi)同一個資源的情況。

解決方案

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

（2）對ZREM的返回值進(jìn)行判斷，只有大于0的時候，才消費(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)了一個任務(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)了一個任務(wù)：消費(fèi)的訂單OrderId為”+orderId）;

}

}

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

思路二

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

實現(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ī)制存在一個硬傷，官網(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ā)布/訂閱的客戶端斷鏈之后又重連，則在客戶端斷鏈期間的所有事件都丟失了。因此，方案二不是太推薦。當(dāng)然，如果你對可靠性要求不高，可以使用。

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

缺點：（1）需要額外進(jìn)行redis維護(hù)

（5）使用消息隊列

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

RabbitMQ可以針對Queue和Message設(shè)置 x-message-tt，來控制消息的生存時間，如果超時，則消息變?yōu)閐ead 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的分布式特性輕易的進(jìn)行橫向擴(kuò)展，消息支持持久化增加了可靠性。

缺點：本身的易用度要依賴于rabbitMq的運(yùn)維。因為要引用rabbitMq，所以復(fù)雜度和成本變高

責(zé)任編輯：haq