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

生成訂單30分鐘未支付，則自動(dòng)取消

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

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

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

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

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

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

方案分析

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

思路

該方案通常是在小型項(xiàng)目中使用，即通過(guò)一個(gè)線程定時(shí)的去掃描數(shù)據(jù)庫(kù)，通過(guò)訂單時(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)度器開(kāi)始調(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)：（1）對(duì)服務(wù)器內(nèi)存消耗大

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

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

（2）JDK的延遲隊(duì)列

思路

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

其中Poll（）：獲取并移除隊(duì)列的超時(shí)元素，沒(méi)有則返回空

take（）：獲取并移除隊(duì)列的超時(shí)元素，如果沒(méi)有則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 MilliSeconds

00000002編號(hào)的訂單要?jiǎng)h除啦。。。。

After 6006 MilliSeconds

00000003編號(hào)的訂單要?jiǎng)h除啦。。。。

After 9006 MilliSeconds

00000004編號(hào)的訂單要?jiǎng)h除啦。。。。

After 12008 MilliSeconds

00000005編號(hào)的訂單要?jiǎng)h除啦。。。。

After 15009 MilliSeconds

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

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

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

缺點(diǎn)：

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

（3）時(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 stub

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

e.printStackTrace（）;

}

System.out.println（i+“秒過(guò)去了”）;

i++;

}

}

}

輸出如下

1秒過(guò)去了

2秒過(guò)去了

3秒過(guò)去了

4秒過(guò)去了

5秒過(guò)去了

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

6秒過(guò)去了

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

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

缺點(diǎn)：

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

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

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

（4）redis緩存

思路一

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

添加元素：ZADD key score member ［［score member］ ［score member］ …］

按順序查詢?cè)兀篫RANGE key start stop ［WITHSCORES］

查詢?cè)豷core: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） 2

redis》 ZRANGE page_rank 0 -1 WITHSCORES

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

redis》 ZRANGE page_rank 0 -1 WITHSCORES

1） “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)前沒(méi)有等待的任務(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)了一個(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 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)了多個(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秒過(guò)后，訂單取消了

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

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

（5）使用消息隊(duì)列

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

RabbitMQ可以針對(duì)Queue和Message設(shè)置 x-message-tt，來(lái)控制消息的生存時(shí)間，如果超時(shí)，則消息變?yōu)閐ead 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è)特性，就可以模擬出延遲消息的功能，具體的，我改天再寫一篇文章，這里再講下去，篇幅太長(zhǎng)。

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

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

責(zé)任編輯：haq