延遲阻塞佇列 DelayQueue
摘要：
DelayQueue 是一個支援延時獲取元素的阻塞佇列， 內部採用優先佇列 PriorityQueue 儲存元素，同時元素必須實現 Delayed 介面；在建立元素時可以指定多久才可以從佇列中獲取當前元素，只有在延遲期滿時才能從佇列中提取元素。
使用場景
因延遲阻塞佇列的特性， 我們一...
DelayQueue 是一個支援延時獲取元素的阻塞佇列， 內部採用優先佇列 PriorityQueue 儲存元素，同時元素必須實現 Delayed 介面；在建立元素時可以指定多久才可以從佇列中獲取當前元素，只有在延遲期滿時才能從佇列中提取元素。
使用場景
因延遲阻塞佇列的特性， 我們一般將 DelayQueue 作用於以下場景 ：
- 快取系統 ： 當能夠從 DelayQueue 中獲取元素時，說該快取已過期
- 定時任務排程 ：
下面我們以快取系統的應用，看下 DelayQueue 的使用，程式碼如下：
public class DelayQueueDemo {
static class Cache implements Runnable {
private boolean stop = false;
private Map<String, String> itemMap = new HashMap<>();
private DelayQueue<CacheItem> delayQueue = new DelayQueue<>();
public Cache () {
// 開啟內部執行緒檢測是否過期
new Thread(this).start();
}
/**
* 新增快取
*
* @param key
* @param value
* @param exprieTime 過期時間,單位秒
*/
public void put (String key, String value, long exprieTime) {
CacheItem cacheItem = new CacheItem(key, exprieTime);
// 此處忽略新增重複 key 的處理
delayQueue.add(cacheItem);
itemMap.put(key, value);
}
public String get (String key) {
return itemMap.get(key);
}
public void shutdown () {
stop = true;
}
@Override
public void run() {
while (!stop) {
CacheItem cacheItem = delayQueue.poll();
if (cacheItem != null) {
// 元素過期, 從快取中移除
itemMap.remove(cacheItem.getKey());
System.out.println("key : " + cacheItem.getKey() + " 過期並移除");
}
}
System.out.println("cache stop");
}
}
static class CacheItem implements Delayed {
private String key;
/**
* 過期時間(單位秒)
*/
private long exprieTime;
private long currentTime;
public CacheItem(String key, long exprieTime) {
this.key = key;
this.exprieTime = exprieTime;
this.currentTime = System.currentTimeMillis();
}
@Override
public long getDelay(TimeUnit unit) {
// 計算剩餘的過期時間
// 大於 0 說明未過期
return exprieTime - TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis() - currentTime);
}
@Override
public int compareTo(Delayed o) {
// 過期時間長的放置在佇列尾部
if (this.getDelay(TimeUnit.MICROSECONDS) > o.getDelay(TimeUnit.MICROSECONDS)) {
return 1;
}
// 過期時間短的放置在佇列頭
if (this.getDelay(TimeUnit.MICROSECONDS) < o.getDelay(TimeUnit.MICROSECONDS)) {
return -1;
}
return 0;
}
public String getKey() {
return key;
}
}
public static void main(String[] args) throws InterruptedException {
Cache cache = new Cache();
// 新增快取元素
cache.put("a", "1", 5);
cache.put("b", "2", 4);
cache.put("c", "3", 3);
while (true) {
String a = cache.get("a");
String b = cache.get("b");
String c = cache.get("c");
System.out.println("a : " + a + ", b : " + b + ", c : " + c);
// 元素均過期後退出迴圈
if (StringUtils.isEmpty(a) && StringUtils.isEmpty(b) && StringUtils.isEmpty(c)) {
break;
}
TimeUnit.MILLISECONDS.sleep(1000);
}
cache.shutdown();
}
}
複製程式碼
執行結果如下：
a : 1, b : 2, c : 3 a : 1, b : 2, c : 3 a : 1, b : 2, c : 3 key : c 過期並移除 a : 1, b : 2, c : null key : b 過期並移除 a : 1, b : null, c : null key : a 過期並移除 a : null, b : null, c : null cache stop 複製程式碼
從執行結果可以看出，因迴圈內部每次停頓 1 秒，當等待 3 秒後，元素 c 過期並從快取中清除，等待 4 秒後，元素 b 過期並從快取中清除，等待 5 秒後，元素 a 過期並從快取中清除。
實現原理
變數
重入鎖
private final transient ReentrantLock lock = new ReentrantLock(); 複製程式碼
用於保證佇列操作的執行緒安全性
優先佇列
private final PriorityQueue<E> q = new PriorityQueue<E>(); 複製程式碼
儲存介質，用於保證延遲低的優先執行
leader
leader 指向的是第一個從佇列獲取元素阻塞等待的執行緒，其作用是減少其他執行緒不必要的等待時間。（這個地方我一直沒搞明白 怎麼就減少其他執行緒的等待時間了）
condition
private final Condition available = lock.newCondition(); 複製程式碼
條件物件，當新元素到達，或新執行緒可能需要成為leader時被通知
下面將主要對佇列的入隊，出隊動作進行分析 ：
入隊 - offer
public boolean offer(E e) {
final ReentrantLock lock = this.lock;
lock.lock();
try {
// 入隊
q.offer(e);
if (q.peek() == e) {
// 若入隊的元素位於佇列頭部，說明當前元素延遲最小
// 將 leader 置空
leader = null;
// 喚醒阻塞在等待佇列的執行緒
available.signal();
}
return true;
} finally {
lock.unlock();
}
}
複製程式碼
出隊 - poll
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
for (;;) {
E first = q.peek();
if (first == null)
// 等待 add 喚醒
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)
// 若 leader 不為空
// 說明已經有其他執行緒呼叫過 take 操作
// 當前呼叫執行緒 follower 掛起等待
available.await();
else {
// 若 leader 為空
// 將 leader 指向當前執行緒
Thread thisThread = Thread.currentThread();
leader = thisThread;
try {
// 當前呼叫執行緒在指定 delay 時間內掛起等待
available.awaitNanos(delay);
} finally {
if (leader == thisThread)
leader = null;
}
}
}
}
} finally {
if (leader == null && q.peek() != null)
// leader 處理完之後，喚醒 follower
available.signal();
lock.unlock();
}
}
複製程式碼