延遲阻塞佇列 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&emsp;過期時間,單位秒
         */
        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();
        }
    }
複製程式碼

Leader-follower 模式

該圖引用自 CSDN 《Leader/Follower多執行緒網路模型介紹》

小結

看了 DelayQueue 的實現 我們大概也明白 PriorityQueue 採用小頂堆的原因了。