DeplayQueue延时无界阻塞队列

￡神魔★判官ぃ 2024-03-16 10:29 68阅读 0赞

在谈到 DelayQueue 的使用和原理的时候，我们首先介绍一下 DelayQueue，DelayQueue 是一个无界阻塞队列，只有在延迟期满时才能从中提取元素。该队列的头部是延迟期满后保存时间最长的Delayed 元素。

DelayQueue 阻塞队列在我们系统开发中也常常会用到，例如：缓存系统的设计，缓存中的对象，超过了空闲时间，需要从缓存中移出；任务调度系统，能够准确的把握任务的执行时间。我们可能需要通过线程处理很多时间上要求很严格的数据，如果使用普通的线程，我们就需要遍历所有的对象，一个一个的检 查看数据是否过期等，首先这样在执行上的效率不会太高，其次就是这种设计的风格也大大的影响了数据的精度。一个需要 12:00 点执行的任务可能12:01 才执行,这样对数据要求很高的系统有更大的弊端。由此我们可以使用DelayQueue。

下面将会对 DelayQueue 做一个介绍，然后举个例子。并且提供一个 Delayed 接口的实现和 Sample 代码。

DelayQueue 是一个 BlockingQueue，其特化的参数是 Delayed。（不了解 BlockingQueue 的同学，先去了解

BlockingQueue再看本文）

Delayed扩展了Comparable接口，比较的基准为延时的时间值，Delayed接口的实现类getDelay的返回值应

为固定值（final）。DelayQueue内部是使用PriorityQueue实现的。

**D****el****a****y****Q****ueu****e****=** **Bloc****k****i****ng****Q****ueue** **+****P****rio****r****ity****Q****ue****u****e****+****D****e****l****a****y****e****d**

**D****el****a****y****Q****ueu****e****定****义****和****原理**

DelayQueue的关键元素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();

**D****el****a****y****Q****ueu****e****实****例****应****用**

Ps：为了具有调用行为，存放到 DelayDeque 的元素必须继承 Delayed 接口。Delayed 接口使对象成为延迟对

象，它使存放在DelayQueue类中的对象具有了激活日期。该接口强制执行下列两个方法。

一下将使用Delay做一个缓存的实现。其中共包括三个类

n Pair

n DelayItem

n Cache

Pair类：

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;

一下是对Delay接口的实现：

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方法

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);

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);

测试main方法：

// 测试入口函数

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);

输出结果为：

aaaa

null

我们看到上面的结果，如果超过延时的时间，那么缓存中数据就会自动丢失，获得就为null。