361
技術社區[雲棲]
Java線程池架構2-多線程調度器（ScheduledThreadPoolExecutor）
在前麵介紹了java的多線程的基本原理信息：《Java線程池架構原理和源碼解析(ThreadPoolExecutor)》，本文對這個java本身的線程池的調度器做一個簡單擴展，如果還沒讀過上一篇文章，建議讀一下，因為這是調度器的核心組件部分。
我們如果要用java默認的線程池來做調度器，一種選擇就是Timer和TimerTask的結合，在以前的文章：《Timer與TimerTask的真正原理&使用介紹》中有明確的說明：一個Timer為一個單獨的線程，雖然一個Timer可以調度多個TimerTask，但是對於一個Timer來講是串行的，至於細節請參看對應的那篇文章的內容，本文介紹的多線程調度器，也就是定時任務，基於多線程調度完成，當然你可以為了完成多線程使用多個Timer，隻是這些Timer的管理需要你來完成，不是一個框架體係，而ScheduleThreadPoolExecutor提供了這個功能，所以我們第一要搞清楚是如何使用調度器的，其次是需要知道它的內部原理是什麼，也就是知其然，再知其所以然！
首先如果我們要創建一個基於java本身的調度池通常的方法是：
Executors.newScheduledThreadPool(int);
當有重載方法，我們最常用的是這個就從這個，看下定義：
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
return new ScheduledThreadPoolExecutor(corePoolSize);
}
其實內部是new了一個實例化對象出來，並傳入大小，此時就跟蹤到ScheduledThreadPoolExecutor的構造方法中：
public ScheduledThreadPoolExecutor(int corePoolSize) {
super(corePoolSize, Integer.MAX_VALUE, 0,TimeUnit.NANOSECONDS,
new DelayedWorkQueue());
}
你會發現調用了super，而super你跟蹤進去會發現，是ThreadPoolExecutor中，那麼ScheduledThreadPoolExecutor和ThreadPoolExecutor有何區別，就是本文要說得重點了，首先我們留下個引子，你發現在定義隊列的時候，不再是上文中提到的LinkedBlockingQueue，而是DelayedWorkQueue，那麼細節上我們接下來就是要講解的重點，既然他們又繼承關係，其實搞懂了不同點，就搞懂了共同點，而且有這樣的關係大多數應當是共同點，不同點的猜測：這個是要實現任務調度，任務調度不是立即的，需要延遲和定期做等情況，那麼是如何實現的呢？
這就是我們需要思考的了，通過源碼考察，我們發現，他們都有execute方法，隻是ScheduledThreadPoolExecutor將源碼進行了重寫，並且還有以下四個調度器的方法：
public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit); public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit); public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit); public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit);
那麼這四個方法有什麼區別呢？其實第一個和第二個區別不大，一個是Runnable、一個是Callable，內部包裝後是一樣的效果；所以把頭兩個方法幾乎當成一種調度，那麼三種情況分別是：
1、 進行一次延遲調度：延遲delay這麼長時間，單位為：TimeUnit傳入的的一個基本單位，例如：TimeUnit.SECONDS屬於提供好的枚舉信息；（適合於方法1和方法2）。
2、 多次調度，每次依照上一次預計調度時間進行調度，例如：延遲2s開始，5s一次，那麼就是2、7、12、17，如果中間由於某種原因導致線程不夠用，沒有得到調度機會，那麼接下來計算的時間會優先計算進去，因為他的排序會被排在前麵，有點類似Timer中的：scheduleAtFixedRate方法，隻是這裏是多線程的，它的方法名也叫：scheduleAtFixedRate，所以這個是比較好記憶的（適合方法3）
3、 多次調度，每次按照上一次實際執行的時間進行計算下一次時間，同上，如果在第7秒沒有被得到調度，而是第9s才得到調度，那麼計算下一次調度時間就不是12秒，而是9+5=14s，如果再次延遲，就會延遲一個周期以上，也就會出現少調用的情況（適合於方法3）；
4、 最後補充execute方法是一次調度，期望被立即調度，時間為空：
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
schedule(command, 0, TimeUnit.NANOSECONDS);
}
我們簡單看看scheduleAtFixedRate、scheduleWithFixedDelay對下麵的分析會更加有用途：
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
long initialDelay,
long period,
TimeUnit unit) {
if (command == null || unit == null)
throw new NullPointerException();
if (period <= 0)
throw new IllegalArgumentException();
RunnableScheduledFuture<?> t = decorateTask(command,
new ScheduledFutureTask<Object>(command,
null,
triggerTime(initialDelay, unit),
unit.toNanos(period)));
delayedExecute(t);
return t;
}
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
long initialDelay,
long delay,
TimeUnit unit) {
if (command == null || unit == null)
throw new NullPointerException();
if (delay <= 0)
throw new IllegalArgumentException();
RunnableScheduledFuture<?> t = decorateTask(command,
new ScheduledFutureTask<Boolean>(command,
null,
triggerTime(initialDelay, unit),
unit.toNanos(-delay)));
delayedExecute(t);
return t;
}
你是否發現，兩段源碼唯一的區別就是在unit.toNanos(int)這唯一一個地方，scheduleAtFixedRate裏麵是直接傳入值，而scheduleWithFixedDelay裏麵是取了相反數，也就是假如我們都傳入正數，scheduleWithFixedDelay其實就取反了，沒有任何區別，你是否聯想到前麵文章介紹Timer中類似的處理手段通過正負數區分時間間隔方法，為0代表僅僅調度一次，其實在這裏同樣是這樣的，他們也同樣有一個問題就是，如果你傳遞負數，方法的功能正好是相反的。
而你會發現，不論是那個schedule方法裏頭，都會創建一個ScheduledFutureTask類的實例，此類究竟是何方神聖呢，我們來看看。
ScheduledFutureTask的類（ScheduleThreadPoolExecutor的私有的內部類）來進行調度，那麼可以看看內部做了什麼操作，如下：
ScheduledFutureTask(Runnable r, V result, long ns) {
super(r, result);
this.time = ns;
this.period = 0;
this.sequenceNumber = sequencer.getAndIncrement();
}
/**
* Creates a periodic action with given nano time and period.
*/
ScheduledFutureTask(Runnable r, V result, long ns, long period) {
super(r, result);
this.time = ns;
this.period = period;
this.sequenceNumber = sequencer.getAndIncrement();
}
/**
* Creates a one-shot action with given nanoTime-based trigger.
*/
ScheduledFutureTask(Callable<V> callable, long ns) {
super(callable);
this.time = ns;
this.period = 0;
this.sequenceNumber = sequencer.getAndIncrement();
}
最核心的幾個參數正好對應了調度的延遲的構造方法，這些參數如何用起來的？那麼它還提供了什麼方法呢？
public long getDelay(TimeUnit unit) {
return unit.convert(time - now(), TimeUnit.NANOSECONDS);
}
public int compareTo(Delayed other) {
if (other == this) // compare zero ONLY if same object
return 0;
if (other instanceof ScheduledFutureTask) {
ScheduledFutureTask<?> x = (ScheduledFutureTask<?>)other;
long diff = time - x.time;
if (diff < 0)
return -1;
else if (diff > 0)
return 1;
else if (sequenceNumber < x.sequenceNumber)
return -1;
else
return 1;
}
long d = (getDelay(TimeUnit.NANOSECONDS) -
other.getDelay(TimeUnit.NANOSECONDS));
return (d == 0)? 0 : ((d < 0)? -1 : 1);
}
/**
* 返回是否為片段，也就是多次調度
*
*/
public boolean isPeriodic() {
return period != 0;
}
這裏發現了，他們可以運行，且判定時間的方法是getDelay方法我們知道了。
對比時間的方法是：compareTo，傳入了參數類型為：Delayed類型，不難猜測出，ScheduledFutureTask和Delayed有某種繼承關係，沒錯，ScheduledFutureTask實現了Delayed的接口，隻是它是間接實現的；並且Delayed接口繼承了Comparable接口，這個接口可用來幹什麼？看過我前麵寫的一篇文章關於中文和對象排序的應該知道，這個是用來自定義對比和排序的，我們的調度任務是一個對象，所以需要排序才行，接下來我們回溯到開始定義的代碼中，找一個實際調用的代碼來看看它是如何啟動到run方法的？如何排序的？如何調用延遲的？就是我們下文中會提到的，而這裏我們先提出問題，後文我們再來說明這些問題。
我們先來看下run方法的一些定義。
/**
* 時間片類型任務執行
*/
private void runPeriodic() {
//運行對應的程序，這個是具體的程序
boolean ok = ScheduledFutureTask.super.runAndReset();
boolean down = isShutdown();
// Reschedule if not cancelled and not shutdown or policy allows
if (ok && (!down ||
(getContinueExistingPeriodicTasksAfterShutdownPolicy() &&
!isStopped()))) {
long p = period;
if (p > 0)//規定時間間隔算出下一次時間
time += p;
else//用當前時間算出下一次時間，負負得正
time = triggerTime(-p);
//計算下一次時間，並資深再次放入等待隊列中
ScheduledThreadPoolExecutor.super.getQueue().add(this);
}
else if (down)
interruptIdleWorkers();
}
/**
* 是否為逐片段執行，如果不是，則調用父親類的run方法
*/
public void run() {
if (isPeriodic())//周期任務
runPeriodic();
else//隻執行一次的任務
ScheduledFutureTask.super.run();
}
可以看到run方法首先通過isPeriod()判定是否為時間片，判定的依據就是我們說的時間片是否“不為零”，如果不是周期任務，就直接運行一次，如果是周期任務，則除了運行還會計算下一次執行的時間，並將其再次放入等待隊列，這裏對應到scheduleAtFixedRate、scheduleWithFixedDelay這兩個方法一正一負，在這裏得到判定，並且將為負數的取反回來，負負得正，java就是這麼幹的，嗬嗬，所以不要認為什麼是不可能的，隻要好用什麼都是可以的，然後計算的時間一個是基於標準的time加上一個時間片，一個是根據當前時間計算一個時間片，在上文中我們已經明確說明了兩者的區別。
以：schedule方法為例：
public <V> ScheduledFuture<V> schedule(Callable<V> callable,
long delay,
TimeUnit unit) {
if (callable == null || unit == null)
throw new NullPointerException();
RunnableScheduledFuture<V> t = decorateTask(callable,
new ScheduledFutureTask<V>(callable,
triggerTime(delay, unit)));
delayedExecute(t);
return t;
}
其實這個方法內部創建的就是一個我們剛才提到的：ScheduledFutureTask，外麵又包裝了下叫做RunnableScheduledFuture，也就是適配了下而已，嗬嗬，代碼裏麵就是一個return操作，java這樣做的目的是方便子類去擴展。
關鍵是delayedExecute(t)方法中做了什麼？看名稱是延遲執行的意思，難道java的線程可以延遲執行，那所有的任務線程都在運行狀態？
它的源碼是這樣的：
private void delayedExecute(Runnable command) {
if (isShutdown()) {
reject(command);
return;
}
if (getPoolSize() < getCorePoolSize())
prestartCoreThread();
super.getQueue().add(command);
}
我們主要關心prestartCoreThread()和super.getQueue().add(command)，因為如果係統關閉，這些討論都沒有意義的，我們分別叫他們第二小段代碼和第三小段代碼。
第二個部分如果線程數小於核心線程數設置，那麼就調用一個prestartCoreThread()，看方法名應該是：預先啟動一個核心線程的意思，先看完第三個部分，再跟蹤進去看源碼。
第三個部分很明了，就是調用super.getQueue().add(command);也就是說直接將任務放入一個隊列中，其實super是什麼？super就是我們上一篇文章所提到的ThreadPoolExecutor，那麼這個Queue就是上一篇文章中提到的等待隊列，也就是任何schedule任務首先放入等待隊列，然後等待被調度的。
public boolean prestartCoreThread() {
return addIfUnderCorePoolSize(null);
}
private boolean addIfUnderCorePoolSize(Runnable firstTask) {
Thread t = null;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
if (poolSize < corePoolSize && runState == RUNNING)
t = addThread(firstTask);
} finally {
mainLock.unlock();
}
if (t == null)
return false;
t.start();
return true;
}
這個代碼是否似曾相似，沒錯，這個你在上一篇文章介紹ThreadPoolExecutor的時候就見到過，說明不論是ThreadPoolExecutor還是ScheduleThreadPoolExecutor他們的Thread都是由一個Worker來處理的（上一篇文章有介紹），而這個Worker處理的基本機製就是將當前任務執行後，不斷從線程等待隊列中獲取數據，然後用以執行，直到隊列為空為止。
那麼他們的區別在哪裏呢？延遲是如何實現的呢？和我們上麵介紹的ScheduledFutureTask又有何關係呢？
那麼我們回過頭來看看ScheduleThreadPool的定義是如何的。
public ScheduledThreadPoolExecutor(int corePoolSize) {
super(corePoolSize, Integer.MAX_VALUE, 0,TimeUnit.NANOSECONDS,
new DelayedWorkQueue());
}
發現它和ThreadPoolExecutor有個定義上很大的區別就是，ThreadPoolExecutor用的是LinkedBlockingQueue（當然可以修改），它用的是DelayedWeorkQueue，而這個DelayedWorkQueue裏麵你會發現它僅僅是對java.util.concurrent.DelayedQueue類一個簡單訪問包裝，這個隊列就是等待隊列，可以看到任務是被直接放到等待隊列中的，所以取數據必然從這裏獲取，而這個延遲的隊列有何神奇之處呢，它又是如何實現的呢，我們從什麼地方下手去看這個DelayWorkQueue？
我們還是回頭看看Worker裏麵的run方法（上一篇文章中已經講過）：
public void run() {
try {
Runnable task = firstTask;
firstTask = null;
while (task != null || (task = getTask()) != null) {
runTask(task);
task = null;
}
} finally {
workerDone(this);
}
}
這裏麵要調用等待隊列就是getTask()方法：
Runnable getTask() {
for (;;) {
try {
int state = runState;
if (state > SHUTDOWN)
return null;
Runnable r;
if (state == SHUTDOWN) // Help drain queue
r = workQueue.poll();
else if (poolSize > corePoolSize || allowCoreThreadTimeOut)
r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS);
else
r = workQueue.take();
if (r != null)
return r;
if (workerCanExit()) {
if (runState >= SHUTDOWN) // Wake up others
interruptIdleWorkers();
return null;
}
} catch (InterruptedException ie) {
}
}
}
你會發現，如果沒有設置超時，默認隻會通過workQueue.take()方法獲取數據，那麼我們就看take方法，而增加到隊列裏麵的方法自然看offer相關的方法。接下來我們來看下DelayQueue這個隊列的take方法：
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(TimeUnit.NANOSECONDS);
if (delay > 0) {
long tl = available.awaitNanos(delay);//最左等delay的時間段
} else {
E x = q.poll();//可以運行，取出一個
assert x != null;
if (q.size() != 0)
available.signalAll();
return x;
}
}
}
} finally {
lock.unlock();
}
}
這裏的for就是要找到數據為止，否則就等著，而這個“q”和“available”是什麼呢？
private transient final Condition available = lock.newCondition();
private final PriorityQueue<E> q = new PriorityQueue<E>();
怎麼裏麵還有一層隊列，不用怕，從這裏你貌似看出點名稱意味了，就是它是優先級隊列，而對於任務調度來講，優先級的方式就是時間，我們用這中猜測來繼續深入源碼。
上麵首先獲取這個隊列的第一個元素，若為空，就等待一個“available”發出的信號，我們可以猜測到這個offer的時候會發出的信號，一會來驗證即可；若不為空，則通過getDelay方法來獲取時間信息，這個getDelay方法就用上了我們開始說的ScheduledFutureTask了，如果是時間大於0，則也進入等待，因為還沒開始執行，等待也是“available”發出信號，但是有一個最長時間，為什麼還要等這個信號，是因為有可能進來一個新的任務，比這個等待的任務還要先執行，所以要等這個信號；而最多等這麼長時間，就是因為如果這段時間沒任務進來肯定就是它執行了。然後就返回的這個值，被Worker（上麵有提到）拿到後調用其run()方法進行運行。
那麼寫入隊列在那裏？他們是如何排序的？
我們看看隊列的寫入方法是這樣的：public boolean offer(E e) {
final ReentrantLock lock = this.lock;
lock.lock();
try {
E first = q.peek();
q.offer(e);
if (first == null || e.compareTo(first) < 0)
available.signalAll();
return true;
} finally {
lock.unlock();
}
}
隊列也是首先取出第一個（後麵會用來和當前任務做比較），而這裏“q”是上麵提到的“PriorityQueue”，看來offer的關鍵還在它的裏麵，我們看看調用過程：
public boolean offer(E e) {
if (e == null)
throw new NullPointerException();
modCount++;
int i = size;
if (i >= queue.length)
grow(i + 1);
size = i + 1;
if (i == 0)
queue[0] = e;
else
siftUp(i, e);//主要是這條代碼很關鍵
return true;
}
private void siftUp(int k, E x) {
if (comparator != null)
siftUpUsingComparator(k, x);
else
//我們默認走這裏，因為DelayQueue定義它的時候默認沒有給定義comparator
siftUpComparable(k, x);
}
/*
可以發現這個方法是將任務按照compareTo對比後，放在隊列的合適位置，但是它肯定不是絕對順序的，這一點和Timer的內部排序機製類似。
*/
private void siftUpComparable(int k, E x) {
Comparable<? super E> key = (Comparable<? super E>) x;
while (k > 0) {
int parent = (k - 1) >>> 1;
Object e = queue[parent];
if (key.compareTo((E) e) >= 0)
break;
queue[k] = e;
k = parent;
}
queue[k] = key;
}
你是否發現，compareTo也用上了，就是我們前麵描述一大堆的：ScheduledFutureTask類中的一個方法，那麼run方法也用上了，這個過程貌似完整了。
我們再來理一下思路：
1、調用的Thread的包裝，由在ThreadPoolExecutor中的Worker調用你傳入的Runnable的run方法，變成了Worker調用Runnable的run方法，由它來處理時間片的信息調用你傳入的線程。
2、ScheduledFutureTask類在整個過程中提供了基礎參考的方法，其中最為關鍵的就是實現了接口Comparable，實現內部的compareTo方法，也實現了Delayed接口中的getDelay方法用以判定時間（當然Delayed接口本身也是繼承於Comparable，我們不要糾結於細節概念就好）。
3、等待隊列由在ThreadPoolExecutor中默認使用的LinkedBlockingQueue換成了DelayQueue（它是被DelayWorkQueue包裝了一下子，沒多大區別），而DelayQueue主要提供了一個信號量“available”來作為寫入和讀取的信號控製開關，通過另一個優先級隊列“PriorityQueue”來控製實際的隊列順序，他們的順序就是基於上麵提到的ScheduledFutureTask類中的compareTo方法，而是否運行也是基於getDelay方法來實現的。
4、ScheduledFutureTask類的run方法會判定是否為時間片信息，如果為時間片，在執行完對應的方法後，開始計算下一次執行時間（注意判定時間片大於0，小於0，分別代表的是以當前執行完的時間為準計算下一次時間還是以當前時間為準），這個在前麵有提到。
5、它是支持多線程的，和Timer的機製最大的區別就在於多個線程會最征用這個隊列，隊裏的排序方式和Timer有很多相似之處，並非完全有序，而是通過位移動來盡量找到合適的位置，有點類似貪心的算法，嗬嗬。
最後更新：2017-04-03 19:06:50