Java DelayQueue

http://brokendreams.iteye.com/blog/2252538

An unbounded blocking queue of Delayed elements, in which an element can only be taken when its delay has expired. The head of the queue is that Delayed element whose delay expired furthest in the past. If no delay has expired there is no head and poll will return null. Expiration occurs when an element's getDelay(TimeUnit.NANOSECONDS) method returns a value less than or equal to zero. Even though unexpired elements cannot be removed using take or poll, they are otherwise treated as normal elements. For example, the size method returns the count of both expired and unexpired elements. This queue does not permit null elements.

This class and its iterator implement all of the optional methods of the Collection and Iterator interfaces. The Iterator provided in method iterator() is not guaranteed to traverse the elements of the DelayQueue in any particular order.

• DelayQueue是一种无界的阻塞队列，队列里只允许放入可以"延期"的元素，队列中列头的元素是最先"到期"的元素。如果队列中没有任何元素"到期"，尽管队列中有元素，也不能从队列头获取到任何元素。

1. public class DelayQueue<E extends Delayed> extends AbstractQueue<E>  
2.     implements BlockingQueue<E> {  
3.     private transient final ReentrantLock lock = new ReentrantLock();  
4.     private transient final Condition available = lock.newCondition();  
5.     private final PriorityQueue<E> q = new PriorityQueue<E>();  

       内部结构非常简单，一把锁，一个条件，一个优先队列。

       DelayQueue要求放入其中的元素必须实现Delayed接口

1. public boolean offer(E e) {  
2.     final ReentrantLock lock = this.lock;  
3.     lock.lock();  
4.     try {  
5.         //获取队头元素。  
6.         E first = q.peek();  
7.         //将元素放入内部队列。  
8.         q.offer(e);  
9.         if (first == null || e.compareTo(first) < 0)  
10.             available.signalAll(); //如果队头没有元素 或者 当前元素比队头元素的延时值小，那么唤醒available条件上的线程。  
11.         return true;  
12.     } finally {  
13.         lock.unlock();  
14.     }  
15. }  
16. /** 
17.  * 插入一个元素到延迟队列，由于队列是无界的，所以这个方法永远不会阻塞。 
18.  * 
19.  * @param e the element to add 
20.  * @throws NullPointerException {@inheritDoc} 
21.  */  
22. public void put(E e) {  
23.     offer(e);  
24. } 

1. public E take() throws InterruptedException {  
2.     final ReentrantLock lock = this.lock;  
3.     lock.lockInterruptibly();  
4.     try {  
5.         for (;;) {  
6.             //获取队头元素。  
7.             E first = q.peek();  
8.             if (first == null) {  
9.                 available.await();//如果队头没有元素，那么当前线程在available条件上等待。  
10.             } else {  
11.                 //如果队头有元素，获取元素的延时值。  
12.                 long delay =  first.getDelay(TimeUnit.NANOSECONDS);  
13.                 if (delay > 0) {  
14.                     //如果延时值大于0，那么等待一下。  
15.                     long tl = available.awaitNanos(delay);   
16.                 } else {  
17.                     //否则获取并移除队列列头元素。  
18.                     E x = q.poll();  
19.                     assert x != null;  
20.                     if (q.size() != 0)  
21.                         available.signalAll(); // 如果内部队列中还有元素，那么唤醒其他在available条件上等待着的take线程。  
22.                     return x;  
23.                 }  
24.             }  
25.         }  
26.     } finally {  
27.         lock.unlock();  
28.     }  
29. }  

http://www.cnblogs.com/jobs/archive/2007/04/27/730255.html
a) 关闭空闲连接。服务器中，有很多客户端的连接，空闲一段时间之后需要关闭之。
b) 缓存。缓存中的对象，超过了空闲时间，需要从缓存中移出。
c) 任务超时处理。在网络协议滑动窗口请求应答式交互时，处理超时未响应的请求。

一种笨笨的办法就是，使用一个后台线程，遍历所有对象，挨个检查。这种笨笨的办法简单好用，但是对象数量过多时，可能存在性能问题，检查间隔时间不好设置，间隔时间过大，影响精确度，多小则存在效率问题。而且做不到按超时的时间顺序处理。 

这场景，使用DelayQueue最适合了。

DelayQueue是java.util.concurrent中提供的一个很有意思的类。很巧妙，非常棒！但是java doc和Java SE 5.0的source中都没有提供Sample。我最初在阅读ScheduledThreadPoolExecutor源码时，发现DelayQueue的妙用。随后在实际工作中，应用在session超时管理，网络应答通讯协议的请求超时处理。

本文将会对DelayQueue做一个介绍，然后列举应用场景。并且提供一个Delayed接口的实现和Sample代码。

DelayQueue是一个BlockingQueue，其特化的参数是Delayed。（不了解BlockingQueue的同学，先去了解BlockingQueue再看本文）
Delayed扩展了Comparable接口，比较的基准为延时的时间值，Delayed接口的实现类getDelay的返回值应为固定值（final）。DelayQueue内部是使用PriorityQueue实现的。DelayQueue = BlockingQueue + PriorityQueue + DelayedDelayQueue的关键元素BlockingQueue、PriorityQueue、Delayed。可以这么说，DelayQueue是一个使用优先队列（PriorityQueue）实现的BlockingQueue，优先队列的比较基准值是时间。

他们的基本定义如下

public interface Comparable<T> {
    public int compareTo(T o);
}

public interface Delayed extends Comparable<Delayed> {
    long getDelay(TimeUnit unit);
}

public class DelayQueue<E extends Delayed> implements BlockingQueue<E> {
    private final PriorityQueue<E> q = new PriorityQueue<E>();
}

DelayQueue内部的实现使用了一个优先队列。当调用DelayQueue的offer方法时，把Delayed对象加入到优先队列q中。如下：

public boolean offer(E e) {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        E first = q.peek();
        q.offer(e);
        if (first == null || e.compareTo(first) < 0)
            available.signalAll();
        return true;
    } finally {
        lock.unlock();
    }
}

DelayQueue的take方法，把优先队列q的first拿出来（peek），如果没有达到延时阀值，则进行await处理。如下：

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        for (;;) {
            E first = q.peek();
            if (first == null) {
                available.await();
            } else {
                long delay =  first.getDelay(TimeUnit.NANOSECONDS);
                if (delay > 0) {
                    long tl = available.awaitNanos(delay);
                } else {
                    E x = q.poll();
                    assert x != null;
                    if (q.size() != 0)
                        available.signalAll(); // wake up other takers
                    return x;

                }
            }
        }
    } finally {
        lock.unlock();
    }
}

以下是Sample，是一个缓存的简单实现。共包括三个类Pair、DelayItem、Cache。如下：

public class Pair<K, V> {
    public K first;

    public V second;

    public Pair() {}

    public Pair(K first, V second) {
        this.first = first;
        this.second = second;
    }
}

--------------
以下是Delayed的实现

import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

public class DelayItem<T> implements Delayed {
    /** Base of nanosecond timings, to avoid wrapping */
    private static final long NANO_ORIGIN = System.nanoTime();

    /**
     * Returns nanosecond time offset by origin
     */
    final static long now() {
        return System.nanoTime() - NANO_ORIGIN;
    }

    /**
     * Sequence number to break scheduling ties, and in turn to guarantee FIFO order among tied
     * entries.
     */
    private static final AtomicLong sequencer = new AtomicLong(0);

    /** Sequence number to break ties FIFO */
    private final long sequenceNumber;

    /** The time the task is enabled to execute in nanoTime units */
    private final long time;

    private final T item;

    public DelayItem(T submit, long timeout) {
        this.time = now() + timeout;
        this.item = submit;
        this.sequenceNumber = sequencer.getAndIncrement();
    }

    public T getItem() {
        return this.item;
    }

    public long getDelay(TimeUnit unit) {
        long d = unit.convert(time - now(), TimeUnit.NANOSECONDS);
        return d;
    }

    public int compareTo(Delayed other) {
        if (other == this) // compare zero ONLY if same object
            return 0;
        if (other instanceof DelayItem) {
            DelayItem x = (DelayItem) other;
            long diff = time - x.time;
            if (diff < 0)
                return -1;
            else if (diff > 0)
                return 1;
            else if (sequenceNumber < x.sequenceNumber)
                return -1;
            else
                return 1;
        }
        long d = (getDelay(TimeUnit.NANOSECONDS) - other.getDelay(TimeUnit.NANOSECONDS));
        return (d == 0) ? 0 : ((d < 0) ? -1 : 1);
    }
}

以下是Cache的实现，包括了put和get方法，还包括了可执行的main函数。

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;

public class Cache<K, V> {
    private static final Logger LOG = Logger.getLogger(Cache.class.getName());

    private ConcurrentMap<K, V> cacheObjMap = new ConcurrentHashMap<K, V>();

    private DelayQueue<DelayItem<Pair<K, V>>> q = new DelayQueue<DelayItem<Pair<K, V>>>();

    private Thread daemonThread;

    public Cache() {

        Runnable daemonTask = new Runnable() {
            public void run() {
                daemonCheck();
            }
        };

        daemonThread = new Thread(daemonTask);
        daemonThread.setDaemon(true);
        daemonThread.setName("Cache Daemon");
        daemonThread.start();
    }

    private void daemonCheck() {

        if (LOG.isLoggable(Level.INFO))
            LOG.info("cache service started.");

        for (;;) {
            try {
                DelayItem<Pair<K, V>> delayItem = q.take();
                if (delayItem != null) {
                    // 超时对象处理
                    Pair<K, V> pair = delayItem.getItem();
                    cacheObjMap.remove(pair.first, pair.second); // compare and remove
                }
            } catch (InterruptedException e) {
                if (LOG.isLoggable(Level.SEVERE))
                    LOG.log(Level.SEVERE, e.getMessage(), e);
                break;
            }
        }

        if (LOG.isLoggable(Level.INFO))
            LOG.info("cache service stopped.");
    }

    // 添加缓存对象
    public void put(K key, V value, long time, TimeUnit unit) {
        V oldValue = cacheObjMap.put(key, value);
        if (oldValue != null)
            q.remove(key); // o(n) - maybe no need to remove from queue

        long nanoTime = TimeUnit.NANOSECONDS.convert(time, unit);
        q.put(new DelayItem<Pair<K, V>>(new Pair<K, V>(key, value), nanoTime));
    }

    public V get(K key) {
        return cacheObjMap.get(key);
    }

    // 测试入口函数
    public static void main(String[] args) throws Exception {
        Cache<Integer, String> cache = new Cache<Integer, String>();
        cache.put(1, "aaaa", 3, TimeUnit.SECONDS);

        Thread.sleep(1000 * 2);
        {
            String str = cache.get(1);
            System.out.println(str);
        }

        Thread.sleep(1000 * 2);
        {
            String str = cache.get(1);
            System.out.println(str);
        }
    }
}

http://blog.nirav.name/2015/02/a-simple-rate-limiter-using-javas.html

http://javapapers.com/java/java-delayqueue/\
http://tutorials.jenkov.com/java-util-concurrent/delayqueue.html
DelayQueue class implements the BlockingQueue interface. 

The DelayQueue blocks the elements internally until a certain delay has expired. The elements must implement the interface java.util.concurrent.Delayed. Here is how the interface looks:

public interface Delayed extends Comparable<Delayed< {

 public long getDelay(TimeUnit timeUnit);

}

The value returned by the getDelay() method should be the delay remaining before this element can be released. If 0 or a negative value is returned, the delay will be considered expired, and the element released at the next take() etc. call on the DelayQueue.

Java DelayQueue is a collection and an implementation of BlockingQueue. Unlike ArrayBlockingQueue, DelayQueue is an unbounded collection. The BlockingQueue methods are implemented in such a way that only delay expired elements can be taken out of the queue. If the delay has not expired for any elements in the queue then the poll method will return null.

public class DelayElement implements Delayed {
 private String element;
 private long expiryTime;

 public DelayElement(String element, long delay) {
  this.element = element;
  this.expiryTime = System.currentTimeMillis() + delay;
 }

 @Override
 public long getDelay(TimeUnit timeUnit) {
  long diff = expiryTime - System.currentTimeMillis();
  return timeUnit.convert(diff, TimeUnit.MILLISECONDS);
 }

 @Override
 public int compareTo(Delayed o) {
  if (this.expiryTime < ((DelayElement) o).expiryTime) {
   return -1;
  }
  if (this.expiryTime > ((DelayElement) o).expiryTime) {
   return 1;
  }
  return 0;
 }

 @Override
 public String toString() {
  return element + ": " + expiryTime;
 }
}

https://www.javacodegeeks.com/2012/09/changing-delay-and-hence-order-in.html

So the trick is when you know you are going to modify the delay is to remove and then re-add the element to the queue.

1	// Replacement swap loop

2	for (DelayedSwap d : delayed) {

3	if (dq.remove(d))

4

{

5	d.swap = true;

6	dq.add(d);

7

}

8

}

I don’t think this is a bug in the object itself, as you wouldn’t expect a HashTable to orders it’s self when the key changes

https://soulmachine.gitbooks.io/system-design/content/cn/task-scheduler.html

请实现一个定时任务调度器，有很多任务，每个任务都有一个时间戳，任务会在该时间点开始执行。

定时执行任务是一个很常见的需求，所以这题也是从实践中提炼出来的，做好了将来说不定能用上。

仔细分析一下，这题可以分为三个部分：

+

• 优先队列。因为多个任务需要按照时间从小到大排序，所以需要用优先队列。
• 生产者。不断往队列里塞任务。
• 消费者。如果队列里有任务过期了，则取出来执行该任务。

方案1: PriorityBlockingQueue + Polling

我们很快可以想到第一个办法：

• 用一个java.util.concurrent.PriorityBlockingQueue来作为优先队列。因为我们需要一个优先队列，又需要线程安全，用PriorityBlockingQueue再合适不过了。你也可以手工实现一个自己的PriorityBlockingQueue，用java.util.PriorityQueue + ReentrantLock，用一把锁把这个队列保护起来，就是线程安全的啦
• 对于生产者，可以用一个while(true)，造一些随机任务塞进去
• 对于消费者，起一个线程，在 while(true)里每隔几秒检查一下队列，如果有任务，则取出来执行。

这个方案的确可行，总结起来就是轮询(polling)。轮询通常有个很大的缺点，就是时间间隔不好设置，间隔太长，任务无法及时处理，间隔太短，会很耗CPU。

方案2: PriorityBlockingQueue + 时间差

可以把方案1改进一下，while(true)里的逻辑变成：

• 偷看一下堆顶的元素，但并不取出来，如果该任务过期了，则取出来
• 如果没过期，则计算一下时间差，然后 sleep()该时间差

不再是 sleep() 一个固定间隔了，消除了轮询的缺点。

方案3: DelayQueue

方案2虽然已经不错了，但是还可以优化一下，Java里有一个DelayQueue，完全符合题目的要求。DelayQueue 设计得非常巧妙，可以看做是一个特化版的PriorityBlockingQueue，它把计算时间差并让消费者等待该时间差的功能集成进了队列，消费者不需要关心时间差的事情了，直接在while(true)里不断take()就行了。

方案3: DelayQueue

方案2虽然已经不错了，但是还可以优化一下，Java里有一个DelayQueue，完全符合题目的要求。DelayQueue 设计得非常巧妙，可以看做是一个特化版的PriorityBlockingQueue，它把计算时间差并让消费者等待该时间差的功能集成进了队列，消费者不需要关心时间差的事情了，直接在while(true)里不断take()就行了。

DelayQueue这个方案，每个消费者线程只需要等待所需要的时间差，因此响应速度更快。它内部用了一个优先队列，所以插入和删除的时间复杂度都是$\log n$。

+

JDK里还有一个ScheduledThreadPoolExecutor，原理跟DelayQueue类似，封装的更完善，平时工作中可以用它，不过面试中，还是拿DelayQueue来讲吧，它封装得比较薄，容易讲清楚原理。

public class DelayQueue<E extends Delayed> {
    private final transient ReentrantLock lock = new ReentrantLock();
    private final PriorityQueue<E> q = new PriorityQueue<E>();
    private final Condition available = lock.newCondition();
    private Thread leader = null;

    public boolean put(E e) {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            q.offer(e);
            if (q.peek() == e) {
                leader = null;
                available.signal();
            }
            return true;
        } finally {
            lock.unlock();
        }
    }

    public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            for (;;) {
                E first = q.peek();
                if (first == null)
                    available.await();
                else {
                    long delay = first.getDelay(NANOSECONDS);
                    if (delay <= 0)
                        return q.poll();
                    first = null; // don't retain ref while waiting
                    if (leader != null)
                        available.await();
                    else {
                        Thread thisThread = Thread.currentThread();
                        leader = thisThread;
                        try {
                            available.awaitNanos(delay);
                        } finally {
                            if (leader == thisThread)
                                leader = null;
                        }
                    }
                }
            }
        } finally {
            if (leader == null && q.peek() != null)
                available.signal();
            lock.unlock();
        }
    }

1. put()方法

if (q.peek() == e) {
    leader = null;
    available.signal();
}

如果第一个元素等于刚刚插入进去的元素，说明刚才队列是空的。现在队列里有了一个任务，那么就应该唤醒所有在等待的消费者线程。将leader重置为null，这些消费者之间互相竞争，自然有一个会被选为leader。

2. 线程leader的作用

leader这个成员有啥作用？它主要是为了减少不必要的等待时间。比如队列头部还有5秒就要开始了，那么就让消费者线程sleep 5秒，消费者不再需要等待固定的时间间隔了。

想象一下有个多个消费者线程用take方法去取任务,内部先加锁,然后每个线程都去peek头节点。如果leader不为空说明已经有线程在取了，让当前消费者无限等待。

if (leader != null)
   available.await();

如果为空说明没有其他消费者去取任务,设置leader为当前消费者，并让改消费者等待指定的时间，

else {
    Thread thisThread = Thread.currentThread();
    leader = thisThread;
    try {
         available.awaitNanos(delay);
    } finally {
         if (leader == thisThread)
             leader = null;
    }
}

下次循环会走如下分支，取到任务结束，

if (delay <= 0)
    return q.poll();

3. take()方法中为什么释放first

first = null; // don't retain ref while waiting

我们可以看到 Doug Lea 后面写的注释，那么这行代码有什么用呢？

如果删除这行代码，会发生什么呢？假设现在有3个消费者线程，

• 线程A进来获取first,然后进入 else 的 else ,设置了leader为当前线程A，并让A等待一段时间
• 线程B进来获取first, 进入else的阻塞操作,然后无限期等待，这时线程B是持有first引用的
• 线程A等待指定时间后被唤醒，获取对象成功，出队，这个对象理应被GC回收，但是它还被线程B持有着，GC链可达，所以不能回收这个first
• 只要线程B无限期的睡眠，那么这个本该被回收的对象就不能被GC销毁掉，那么就会造成内存泄露

JDK中有一个接口java.util.concurrent.Delayed，可以用于表示具有过期时间的元素，刚好可以拿来表示任务这个概念。

DelayQueue这个方案，每个消费者线程只需要等待所需要的时间差，因此响应速度更快。它内部用了一个优先队列，所以插入和删除的时间复杂度都是$\log n$。

JDK里还有一个ScheduledThreadPoolExecutor，原理跟DelayQueue类似，封装的更完善，平时工作中可以用它，不过面试中，还是拿DelayQueue来讲吧，它封装得比较薄，容易讲清楚原理。

http://www.cnblogs.com/haoxinyue/p/6663720.html

DelayQueue是一种很好的实现方式，虽然是单机，但是可以多线程生产和消费，提高效率。拿到消息后也可以使用异步线程去执行下一步的任务。如果有分布式的需求可以使用Redis来实现消息的分发，如果对消息的可靠性有非常高的要求可以使用消息中间件

使用DelayQueue需要考虑程序挂掉之后，内存里面未处理消息的丢失带来的影响。

+

3. JDK ScheduledExecutorService

JDK自带的一种线程池，它能调度一些命令在一段时间之后执行，或者周期性的执行。文章开头的一些业务场景主要使用第一种方式，即，在一段时间之后执行某个操作

ScheduledExecutorService的实现类ScheduledThreadPoolExecutor提供了一种并行处理的模型，简化了线程的调度。DelayedWorkQueue是类似DelayQueue的实现，也是基于最小堆的、线程安全的数据结构，所以会有上例排序后输出的结果。

ScheduledExecutorService比上面一种DelayQueue更加实用。因为，一般来说，使用DelayQueue获取消息后触发事件都会实用多线程的方式执行，以保证其他事件能准时进行。而ScheduledThreadPoolExecutor就是对这个过程进行了封装，让大家更加方便的使用。同时在加强了部分功能，比如定时触发命令。

方案4: HashedWheelTimer

时间轮是一种非常惊艳的数据结构。其在Linux内核中使用广泛，是Linux内核定时器的实现方法和基础之一。按使用场景，大致可以分为两种时间轮：原始时间轮和分层时间轮。分层时间轮是原始时间轮的升级版本，来应对时间“槽”数量比较大的情况，对内存和精度都有很高要求的情况。我们延迟任务的场景一般只需要用到原始时间轮就可以了。

相比DelayQueue的数据结构，时间轮在算法复杂度上有一定优势。DelayQueue由于涉及到排序，需要调堆，插入和移除的复杂度是O(lgn)，而时间轮在插入和移除的复杂度都是O(1)。

https://my.oschina.net/haogrgr/blog/489320

时间轮(HashedWheelTimer)其实很简单，就是一个循环队列，如下图所示，

上图是一个长度为8的循环队列，假设该时间轮精度为秒，即每秒走一格，像手表那样，走完一圈就是8秒。每个格子指向一个任务集合，时间轮无限循环，每转到一个格子，就扫描该格子下面的所有任务，把时间到期的任务取出来执行。

举个例子，假设指针当前正指向格子0，来了一个任务需要4秒后执行，那么这个任务就会放在格子4下面，如果来了一个任务需要20秒后执行怎么？由于这个循环队列转一圈只需要8秒，这个任务需要多转2圈，所以这个任务的位置虽然依旧在格子4(20%8+0=4)下面，不过需要多转2圈后才执行。因此每个任务需要有一个字段记录需圈数，每转一圈就减1，减到0则立刻取出来执行。

怎么实现时间轮呢？Netty中已经有了一个时间轮的实现, HashedWheelTimer.java，可以参考它的源代码。

时间轮的优点是性能高，插入和删除的时间复杂度都是O(1)。Linux 内核中的定时器采用的就是这个方案。

Follow up: 如何设计一个分布式的定时任务调度器呢？ 答: Redis ZSet, RabbitMQ等

+

Quartz

quartz是一个企业级的开源的任务调度框架，quartz内部使用TreeSet来保存Trigger，如下图。Java中的TreeSet是使用TreeMap实现，TreeMap是一个红黑树实现。红黑树的插入和删除复杂度都是logN。和最小堆相比各有千秋。最小堆插入比红黑树快，删除顶层节点比红黑树慢。

相比上述的三种轻量级的实现功能丰富很多。有专门的任务调度线程，和任务执行线程池。quartz功能强大，主要是用来执行周期性的任务，当然也可以用来实现延迟任务。但是如果只是实现一个简单的基于内存的延时任务的话，quartz就稍显庞大。

Redis Zset

Redis中的ZSet是一个有序的Set，内部使用HashMap和跳表(SkipList)来保证数据的存储和有序，HashMap里放的是成员到score的映射，而跳跃表里存放的是所有的成员，排序依据是HashMap里存的score,使用跳跃表的结构可以获得比较高的查找效率，并且在实现上比较简单。

在用作延迟任务的时候，可以在添加数据的时候，使用zadd把score写成未来某个时刻的unix时间戳。消费者使用zrangeWithScores获取优先级最高的（最早开始的的）任务。注意，zrangeWithScores并不是取出来，只是看一下并不删除，类似于Queue的peek方法。程序对最早的这个消息进行验证，是否到达要运行的时间，如果是则执行，然后删除zset中的数据。如果不是，则继续等待。

由于zrangeWithScores 和 zrem是先后使用，所以有可能有并发问题，即两个线程或者两个进程都会拿到一样的一样的数据，然后重复执行，最后又都会删除。如果是单机多线程执行，或者分布式环境下，可以使用Redis事务，也可以使用由Redis实现的分布式锁，或者使用下例中Redis Script。你可以在Redis官方的Transaction章节找到事务的相关内容。

使用Redis的好处主要是：

1. 解耦：把任务、任务发起者、任务执行者的三者分开，逻辑更加清晰，程序强壮性提升，有利于任务发起者和执行者各自迭代，适合多人协作。

2. 异常恢复：由于使用Redis作为消息通道，消息都存储在Redis中。如果发送程序或者任务处理程序挂了，重启之后，还有重新处理数据的可能性。

3. 分布式：如果数据量较大，程序执行时间比较长，我们可以针对任务发起者和任务执行者进行分布式部署。特别注意任务的执行者，也就是Redis的接收方需要考虑分布式锁的问题。

        jedis.zadd("zset_test", second50later, "e");
        jedis.zadd("zset_test", second10later, "a");
        jedis.zadd("zset_test", second30later, "c");
        jedis.zadd("zset_test", second20later, "b");
        jedis.zadd("zset_test", second40later, "d");

        while (true) {
            try {

                Set<Tuple> set = jedis.zrangeWithScores("zset_test", 0, 0);

                String value = ((Tuple) set.toArray()[0]).getElement();
                int score = (int) ((Tuple) set.toArray()[0]).getScore();

                Calendar cal = Calendar.getInstance();
                int nowSecond = (int) (cal.getTimeInMillis() / 1000);

                if (nowSecond >= score) {
                    jedis.zrem("zset_test", value);
                    System.out.println(sdf.format(new Date()) + " removed value:" + value);
                }

                if (jedis.zcard("zset_test") <= 0)
                {
                    System.out.println(sdf.format(new Date()) + " zset empty ");
                    return;
                }
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    }

Jesque是Resque的java实现，Resque是一个基于Redis的Ruby项目，用于后台的定时任务。Jesque实现延迟任务的方法也是在Redis里面建立一个ZSet，和上例一样的处理方式。上例提到在使用ZSet作为优先级队列的时候，由于zrangeWithScores 和 zrem没法保证原子性，所有在分布式环境下会有问题。在Jesque中，它使用的Redis Script来解决这个问题。Redis Script可以保证操作的原子性，相比事务也减少了一些网络开销，性能更加出色。

 数据库轮询

这是比较常见的一种方式，所有的订单或者所有的命令一般都会存储在数据库中。我们会起一个线程去扫数据库或者一个数据库定时Job，找到那些超时的数据，直接更新状态，或者拿出来执行一些操作。这种方式很简单，不会引入其他的技术，开发周期短。

如果数据量比较大，千万级甚至更多，插入频率很高的话，上面的方式在性能上会出现一些问题，查找和更新对会占用很多时间，轮询频率高的话甚至会影响数据入库。一种可以尝试的方式就是使用类似TBSchedule或Elastic-Job这样的分布式的任务调度加上数据分片功能，把需要判断的数据分到不同的机器上执行。

如果数据量进一步增大，那扫数据库肯定就不行了。另一方面，对于订单这类数据，我们也许会遇到分库分表，那上述方案就会变得过于复杂，得不偿失。

http://www.cnblogs.com/haoxinyue/p/6613706.html