ScheduledThreadPoolExecutor/Timer最小堆原理

java中的ScheduledThreadPoolExecutor用於定時任務, 它使用最小堆來存放定時任務, Timer也是採用最小堆, 它們之間的區別:

1. ScheduledThreadPoolExecutor是多線程的, 多個任務能夠同時執行; 可是Timer是單線程的, 不會出現兩個任務同時執行的問題, 可是若是某個任務卡頓了太長時間, 會形成其它的任務延遲相應的時間。若是ScheduledThreadPoolExecutor的線程數設置成1，那麼和Timer是同樣的.
2. ScheduledThreadPoolExecutor取消任務以後, 會將定時任務從最小堆中移除; 可是Timer的最小堆實現, 沒有刪除節點的功能, 取消定時器時, 只是將這個任務標記爲CANCELLED狀態, 等觸發時間到達時, 纔會移除. 所以Timer的被取消的定時任務沒法及時被GC回收.

因此, 儘可能用ScheduledThreadPoolExecutor來代替Timer

定義

本文主要講ScheduledThreadPoolExecutor中的DelayedWorkQueue的最小堆實現。先看最小堆的含義: 一種徹底二叉樹, 父結點的值小於或等於它的左子節點和右子節點. 以下圖

存儲方式

DelayedWorkQueue使用數組來存儲定時任務, 按照從上到下, 從左到右的順序依次排列:

第K個節點的左子節點位於2K+1的位置, 右子節點位於2K+2的位置, 父結點位於(K-1)/2的位置

插入節點

依然使用上圖的例子, 如今要插入數據**0**
第1步, 插入末尾的位置

第2步, 0比19要小, 須要向上移動

第3步, 0比2要小, 再上移

第4步, 0比1要小, 再上移
至此結束.

刪除節點

以上面的最後一圖爲例, 假設刪除**1**

使用最後一個結點來填充被移除的位置

19與較小的子節點(2)交換位置
至此結束

算法複雜度

N個數據的最小堆, 共有logN層, 最壞的狀況下, 須要移動logN次.

源代碼解讀(jdk8)

ScheduledThreadPoolExecutor的最小堆實現爲DelayedWorkQueue

插入任務

public boolean offer(Runnable x) {
            if (x == null)
                throw new NullPointerException();
            RunnableScheduledFuture<?> e = (RunnableScheduledFuture<?>)x;
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                int i = size;
                if (i >= queue.length)
                    grow(); //數組長度不夠了, 增大一倍
                size = i + 1;
                if (i == 0) {
					//空列隊, 直接放在首位置
                    queue[0] = e;
					//index用於快速定位對象在數組中的位置
                    setIndex(e, 0);
                } else {
				    //i爲最末尾的位置, 從i開始上移
                    siftUp(i, e);
                }
                if (queue[0] == e) {
                    leader = null;
                    available.signal();
                }
            } finally {
                lock.unlock();
            }
            return true;
        }

上移

private void siftUp(int k, RunnableScheduledFuture<?> key) {
            while (k > 0) {
                int parent = (k - 1) >>> 1;//計算節點k的父節點
                RunnableScheduledFuture<?> e = queue[parent];
                if (key.compareTo(e) >= 0) //子節點大於等於父節點, 父節點不須要再移動, 結束
                    break;
                queue[k] = e;  //父節點大於子節點, 將父節點移到下面
                setIndex(e, k);
                k = parent;
            }
            queue[k] = key; //最終找到的合適位置
            setIndex(key, k);
        }

刪除

public boolean remove(Object x) {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                int i = indexOf(x); //索引已經存放在了x的成員變量中, 直接取出
                if (i < 0)  //若是定時任務已經被取消了, index會被置成-1
                    return false;

                setIndex(queue[i], -1);
                int s = --size;
                RunnableScheduledFuture<?> replacement = queue[s]; //最末尾的對象
                queue[s] = null;
                if (s != i) { //若是s == i, 則表示x是最末尾的對象, 不須要更多的操做
                    siftDown(i, replacement);  //將最末尾的對象, 從位置i開始下移.(由siftDown來判斷須要不須要下移)
                    if (queue[i] == replacement) //若是並無下移, 那確定就是須要上移了.
                        siftUp(i, replacement);
                }
                return true;
            } finally {
                lock.unlock();
            }
        }

下移

private void siftDown(int k, RunnableScheduledFuture<?> key) {
            int half = size >>> 1;
            while (k < half) {
                int child = (k << 1) + 1;  //左子節點位置
                RunnableScheduledFuture<?> c = queue[child]; //左子節點對象
                int right = child + 1; //右子節點位置
                if (right < size && c.compareTo(queue[right]) > 0) //左右子節點中, 選出較小的那個子節點
                    c = queue[child = right];
                if (key.compareTo(c) <= 0) //key比子節點都小, 不須要再移動
                    break;
                queue[k] = c; //較小的子節點上移
                setIndex(c, k);
                k = child;
            }
            queue[k] = key; //爲key找到的最終位置
            setIndex(key, k);
        }