集合-DelayQueue原始碼解析
問題
(1)DelayQueue是阻塞佇列嗎?
(2)DelayQueue的實現方式?
(3)DelayQueue主要用於什麼場景?
簡介
DelayQueue是java併發包下的延時阻塞佇列,常用於實現定時任務。
繼承體系
從繼承體系可以看到,DelayQueue實現了BlockingQueue,所以它是一個阻塞佇列。
另外,DelayQueue還組合了一個叫做Delayed的介面,DelayQueue中儲存的所有元素必須實現Delayed介面。
那麼,Delayed是什麼呢?
public interface Delayed extends Comparable<Delayed> {long getDelay(TimeUnit unit); }
Delayed是一個繼承自Comparable的介面,並且定義了一個getDelay()方法,用於表示還有多少時間到期,到期了應返回小於等於0的數值。
原始碼分析
主要屬性
// 用於控制併發的鎖 private final transient ReentrantLock lock = new ReentrantLock(); // 優先順序佇列 private final PriorityQueue<E> q = new PriorityQueue<E>(); // 用於標記當前是否有執行緒在排隊(僅用於取元素時)private Thread leader = null; // 條件,用於表示現在是否有可取的元素 private final Condition available = lock.newCondition();
從屬性我們可以知道,延時佇列主要使用優先順序佇列來實現,並輔以重入鎖和條件來控制併發安全。
因為優先順序佇列是無界的,所以這裡只需要一個條件就可以了。
還記得優先順序佇列嗎?點選連結直達【死磕 java集合之PriorityQueue原始碼分析】
主要構造方法
public DelayQueue() {} public DelayQueue(Collection<? extends E> c) {this.addAll(c); }
構造方法比較簡單,一個預設構造方法,一個初始化新增集合c中所有元素的構造方法。
入隊
因為DelayQueue是阻塞佇列,且優先順序佇列是無界的,所以入隊不會阻塞不會超時,因此它的四個入隊方法是一樣的。
public boolean add(E e) { return offer(e); } public void put(E e) { offer(e); } public boolean offer(E e, long timeout, TimeUnit unit) { return offer(e); } public boolean offer(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(); } }
入隊方法比較簡單:
(1)加鎖;
(2)新增元素到優先順序佇列中;
(3)如果新增的元素是堆頂元素,就把leader置為空,並喚醒等待在條件available上的執行緒;
(4)解鎖;
出隊
因為DelayQueue是阻塞佇列,所以它的出隊有四個不同的方法,有丟擲異常的,有阻塞的,有不阻塞的,有超時的。
我們這裡主要分析兩個,poll()和take()方法。
public E poll() { final ReentrantLock lock = this.lock; lock.lock(); try { E first = q.peek(); if (first == null || first.getDelay(NANOSECONDS) > 0) return null; else return q.poll(); } finally { lock.unlock(); } }
poll()方法比較簡單:
(1)加鎖;
(2)檢查第一個元素,如果為空或者還沒到期,就返回null;
(3)如果第一個元素到期了就呼叫優先順序佇列的poll()彈出第一個元素;
(4)解鎖。
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); // 如果小於0說明已到期,直接呼叫poll()方法彈出堆頂元素 if (delay <= 0) return q.poll(); // 如果delay大於0 ,則下面要阻塞了 // 將first置為空方便gc,因為有可能其它元素彈出了這個元素 // 這裡還持有著引用不會被清理 first = null; // don't retain ref while waiting // 如果前面有其它執行緒在等待,直接進入等待 if (leader != null) available.await(); else { // 如果leader為null,把當前執行緒賦值給它 Thread thisThread = Thread.currentThread(); leader = thisThread; try { // 等待delay時間後自動醒過來 // 醒過來後把leader置空並重新進入迴圈判斷堆頂元素是否到期 // 這裡即使醒過來後也不一定能獲取到元素 // 因為有可能其它執行緒先一步獲取了鎖並彈出了堆頂元素 // 條件鎖的喚醒分成兩步,先從Condition的佇列裡出隊 // 再入隊到AQS的佇列中,當其它執行緒呼叫LockSupport.unpark(t)的時候才會真正喚醒 // 關於AQS我們後面會講的^^ available.awaitNanos(delay); } finally { // 如果leader還是當前執行緒就把它置為空,讓其它執行緒有機會獲取元素 if (leader == thisThread) leader = null; } } } } } finally { // 成功出隊後,如果leader為空且堆頂還有元素,就喚醒下一個等待的執行緒 if (leader == null && q.peek() != null) // signal()只是把等待的執行緒放到AQS的佇列裡面,並不是真正的喚醒 available.signal(); // 解鎖,這才是真正的喚醒 lock.unlock(); } }
take()方法稍微要複雜一些:
(1)加鎖;
(2)判斷堆頂元素是否為空,為空的話直接阻塞等待;
(3)判斷堆頂元素是否到期,到期了直接呼叫優先順序佇列的poll()彈出元素;
(4)沒到期,再判斷前面是否有其它執行緒在等待,有則直接等待;
(5)前面沒有其它執行緒在等待,則把自己當作第一個執行緒等待delay時間後喚醒,再嘗試獲取元素;
(6)獲取到元素之後再喚醒下一個等待的執行緒;
(7)解鎖;
使用方法
說了那麼多,是不是還是不知道怎麼用呢?那怎麼能行,請看下面的案例:
public class DelayQueueTest { public static void main(String[] args) { DelayQueue<Message> queue = new DelayQueue<>(); long now = System.currentTimeMillis(); // 啟動一個執行緒從佇列中取元素 new Thread(()->{ while (true) { try { // 將依次列印1000,2000,5000,7000,8000 System.out.println(queue.take().deadline - now); } catch (InterruptedException e) { e.printStackTrace(); } } }).start(); // 新增5個元素到佇列中 queue.add(new Message(now + 5000)); queue.add(new Message(now + 8000)); queue.add(new Message(now + 2000)); queue.add(new Message(now + 1000)); queue.add(new Message(now + 7000)); } } class Message implements Delayed { long deadline; public Message(long deadline) { this.deadline = deadline; } @Override public long getDelay(TimeUnit unit) { return deadline - System.currentTimeMillis(); } @Override public int compareTo(Delayed o) { return (int) (getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS)); } @Override public String toString() { return String.valueOf(deadline); } }
是不是很簡單,越早到期的元素越先出隊。
總結
(1)DelayQueue是阻塞佇列;
(2)DelayQueue內部儲存結構使用優先順序佇列;
(3)DelayQueue使用重入鎖和條件來控制併發安全;
(4)DelayQueue常用於定時任務;
彩蛋
java中的執行緒池實現定時任務是直接用的DelayQueue嗎?
當然不是,ScheduledThreadPoolExecutor中使用的是它自己定義的內部類DelayedWorkQueue,其實裡面的實現邏輯基本都是一樣的,只不過DelayedWorkQueue裡面沒有使用現成的PriorityQueue,而是使用陣列又實現了一遍優先順序佇列,本質上沒有什麼區別。