延遲阻塞佇列 DelayQueue
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(); } } 複製程式碼