Jdk 1.6 JUCソース解析（19）-SchduledThreadPool Exector
Jdk 1.6 JUCソース解析（19）-SchduledThreadPool Exector
作者：大飛
機能概要： SchduledThreadPool ExectorはTimerのようなタイマーまたはスケジューラであり、Timerと比べて主にいくつかの利点があります。2.ThreadPoolExectorから継承するので、柔軟性と伸縮性があります。3.timerのようなスレッド漏れ問題がなく、timerのスケジュールのタスクが異常終了すると、timer全体がキャンセルされ、他のタスクを実行できなくなります。 ソース分析: SchduledThreadPool ExectorはThreadPool Exectorを継承し、ScheduledExect rServiceインターフェースを実現しました。ThreadPool Exectorは以前に分析しました。ここではSchduledExectorServiceインターフェースを簡単に見てください。
次にScheduledThreadPoolExector内部のいくつかの属性定義を参照してください。
引き続きScheduledThreadPoolExectorの構造方法を見てください。
以下はSchduledExectorServiceインターフェースの方法から始まります。コードの実現の詳細を確認して、まずschedule方法を見ます。
このScheduledFutureTaskクラスを見て、まずこのクラスはFutureTaskを継承して、Runnable ScheduledFutureインターフェースを実現しました。FutureTaskの前に記事を分析しましたが、ここでは後者を見ます。
内部構造を見てください。
ScheduledThreadPoolExector内部で遅延列を使用するため、遅延列に置く要素はDelayedインターフェースを実現しなければならない。ScheduledFutureTaskも必ずDelayedインターフェースを実現している。インターフェース方法の詳細を参照してください。
最後にScheduledFutureTask種類の最も重要な方法を見にきます。
ScheduledFutureTask種類の内容を見終わって、私達はschedule方法に帰って、最後のdelayedExecute方法を見ます。
まとめてみます
1.ScheduledFutureTaskはスケジューリング可能な非同期タスクを示し、ScheduledThreadPool Exectorに提出するタスクはいずれもこのクラスに包装されます。
2.SchduledThreadPoolExectorスケジューリングタスクは、遅延時間によって最初に期限が切れるタスクは、まずスケジュールされます。もし2つのタスクの遅延時間が同じであれば、先に入る順番を確保するために内部シーケンス番号があります。
3. ScheduledFutureTaskがスケジュールされた後に、具体的に実行する時、自分が周期的なタスクかどうかを判断します。もしそうでないならば、任務は一回実行します。もし、先にタスクを実行し、実行が成功すれば、次のトリガイベント（遅延時間）を算出し、再度ScheduledThreadPool Exectorのタスクキューに入れて、次回のスケジュール実行を待つことになります。
ScheduledFutureTask内部の実行ロジックを了解しました。私達はもう次の二つのschedule方法を振り返ってみます。それらの内部包装はすべて使い捨てのScheduledFutureTaskで、引き続きscheduleAtFixedRateとscheduleWixdelayを見ます。
最後にもう一度見てください。SchduledThreadPoolExector閉鎖に関するいくつかの特殊処理があります。
ScheduledThreadPoolExectorのコード解析は完了しました。
参照：Jdk 1.6 JUCソース解析（17）-ThreadPool Exector
Jdk 1.6 JUCソース解析（18）-DelayQue
作者：大飛
機能概要：
public interface ScheduledExecutorService extends ExecutorService {
/**
* ， 。
*/
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);
}次にScheduledThreadPoolExector内部のいくつかの属性定義を参照してください。
public class ScheduledThreadPoolExecutor
extends ThreadPoolExecutor
implements ScheduledExecutorService {
/**
* 。
*/
private volatile boolean continueExistingPeriodicTasksAfterShutdown;
/**
* 。
*/
private volatile boolean executeExistingDelayedTasksAfterShutdown = true;
/**
* ( ) 。
*/
private static final AtomicLong sequencer = new AtomicLong(0);
/** Base of nanosecond timings, to avoid wrapping */
private static final long NANO_ORIGIN = System.nanoTime();引き続きScheduledThreadPoolExectorの構造方法を見てください。
public ScheduledThreadPoolExecutor(int corePoolSize) {
super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
new DelayedWorkQueue());
}
public ScheduledThreadPoolExecutor(int corePoolSize,
ThreadFactory threadFactory) {
super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
new DelayedWorkQueue(), threadFactory);
}
public ScheduledThreadPoolExecutor(int corePoolSize,
RejectedExecutionHandler handler) {
super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
new DelayedWorkQueue(), handler);
}
public ScheduledThreadPoolExecutor(int corePoolSize,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
new DelayedWorkQueue(), threadFactory, handler);
} /**
* An annoying wrapper class to convince javac to use a
* DelayQueue<RunnableScheduledFuture> as a BlockingQueue<Runnable>
*/
private static class DelayedWorkQueue
extends AbstractCollection<Runnable>
implements BlockingQueue<Runnable> {
private final DelayQueue<RunnableScheduledFuture> dq = new DelayQueue<RunnableScheduledFuture>();
...
public boolean add(Runnable x) {
return dq.add((RunnableScheduledFuture)x);
}
public boolean offer(Runnable x) {
return dq.offer((RunnableScheduledFuture)x);
}
public void put(Runnable x) {
dq.put((RunnableScheduledFuture)x);
}
public boolean offer(Runnable x, long timeout, TimeUnit unit) {
return dq.offer((RunnableScheduledFuture)x, timeout, unit);
} public ScheduledFuture<?> schedule(Runnable command,
long delay,
TimeUnit unit) {
if (command == null || unit == null)
throw new NullPointerException();
// RunnableScheduledFuture
RunnableScheduledFuture<?> t = decorateTask(command,
new ScheduledFutureTask<Void>(command, null,
triggerTime(delay, unit)));
// RunnableScheduledFuture
delayedExecute(t);
return t;
}
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;
} /**
* 。
*/
private long triggerTime(long delay, TimeUnit unit) {
// 。
return triggerTime(unit.toNanos((delay < 0) ? 0 : delay));
}
/**
* 。
*/
long triggerTime(long delay) {
// now()，
// ， delay ， overflowFree 。
return now() +
((delay < (Long.MAX_VALUE >> 1)) ? delay : overflowFree(delay));
}
/**
* Returns nanosecond time offset by origin
*/
final long now() {
return System.nanoTime() - NANO_ORIGIN;
}
/**
* Long.MAX_VALUE ，
* 。
* ， ，
* Long.MAX_VALUE 。
*/
private long overflowFree(long delay) {
Delayed head = (Delayed) super.getQueue().peek();
if (head != null) {
long headDelay = head.getDelay(TimeUnit.NANOSECONDS);
if (headDelay < 0 && (delay - headDelay < 0))
delay = Long.MAX_VALUE + headDelay;
}
return delay;
}
// ， 。
protected <V> RunnableScheduledFuture<V> decorateTask(
Runnable runnable, RunnableScheduledFuture<V> task) {
return task;
}
protected <V> RunnableScheduledFuture<V> decorateTask(
Callable<V> callable, RunnableScheduledFuture<V> task) {
return task;
}このScheduledFutureTaskクラスを見て、まずこのクラスはFutureTaskを継承して、Runnable ScheduledFutureインターフェースを実現しました。FutureTaskの前に記事を分析しましたが、ここでは後者を見ます。
public interface RunnableScheduledFuture<V> extends RunnableFuture<V>, ScheduledFuture<V> {
/**
* 。
*/
boolean isPeriodic();
}内部構造を見てください。
private class ScheduledFutureTask<V>
extends FutureTask<V> implements RunnableScheduledFuture<V> {
/** */
private final long sequenceNumber;
/** ， */
private long time;
/**
* ， 。
* ， ，0 。
*/
private final long period;
/**
* Creates a one-shot action with given nanoTime-based trigger time.
*/
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();
}ScheduledThreadPoolExector内部で遅延列を使用するため、遅延列に置く要素はDelayedインターフェースを実現しなければならない。ScheduledFutureTaskも必ずDelayedインターフェースを実現している。インターフェース方法の詳細を参照してください。
public long getDelay(TimeUnit unit) {
// 2 ：1. 。2. now()， now()。
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;
}
// ScheduledFutureTask， 。
long d = (getDelay(TimeUnit.NANOSECONDS) -
other.getDelay(TimeUnit.NANOSECONDS));
return (d == 0)? 0 : ((d < 0)? -1 : 1);
}最後にScheduledFutureTask種類の最も重要な方法を見にきます。
public void run() {
//
if (isPeriodic())
runPeriodic(); // ， 。
else
ScheduledFutureTask.super.run(); // ， run 。
}
public boolean isPeriodic() {
return period != 0;
}
private void runPeriodic() {
// 。
boolean ok = ScheduledFutureTask.super.runAndReset();
// ScheduledThreadPoolExecutor 。
boolean down = isShutdown();
// ，
// ScheduledThreadPoolExecutor ,
// ScheduledThreadPoolExecutor ，
// 。
if (ok && (!down ||
(getContinueExistingPeriodicTasksAfterShutdownPolicy() &&
!isStopped()))) {
// 。
long p = period;
if (p > 0)
time += p; // ， 。
else
time = triggerTime(-p); // ， 。
// ， 。
ScheduledThreadPoolExecutor.super.getQueue().add(this);
}
// 。 ，
// ScheduledThreadPoolExecutor ， 。
else if (down)
interruptIdleWorkers();
}
public boolean getContinueExistingPeriodicTasksAfterShutdownPolicy() {
return continueExistingPeriodicTasksAfterShutdown;
}ScheduledFutureTask種類の内容を見終わって、私達はschedule方法に帰って、最後のdelayedExecute方法を見ます。
private void delayedExecute(Runnable command) {
if (isShutdown()) {
reject(command); // ScheduledThreadPoolExecutor ， 。
return;
}
// ， 。
if (getPoolSize() < getCorePoolSize())
prestartCoreThread();
// 。
super.getQueue().add(command);
}まとめてみます
1.ScheduledFutureTaskはスケジューリング可能な非同期タスクを示し、ScheduledThreadPool Exectorに提出するタスクはいずれもこのクラスに包装されます。
2.SchduledThreadPoolExectorスケジューリングタスクは、遅延時間によって最初に期限が切れるタスクは、まずスケジュールされます。もし2つのタスクの遅延時間が同じであれば、先に入る順番を確保するために内部シーケンス番号があります。
3. ScheduledFutureTaskがスケジュールされた後に、具体的に実行する時、自分が周期的なタスクかどうかを判断します。もしそうでないならば、任務は一回実行します。もし、先にタスクを実行し、実行が成功すれば、次のトリガイベント（遅延時間）を算出し、再度ScheduledThreadPool Exectorのタスクキューに入れて、次回のスケジュール実行を待つことになります。
ScheduledFutureTask内部の実行ロジックを了解しました。私達はもう次の二つのschedule方法を振り返ってみます。それらの内部包装はすべて使い捨てのScheduledFutureTaskで、引き続きscheduleAtFixedRateとscheduleWixdelayを見ます。
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;
} public void setContinueExistingPeriodicTasksAfterShutdownPolicy(boolean value) {
continueExistingPeriodicTasksAfterShutdown = value;
if (!value && isShutdown())
cancelUnwantedTasks();
}
public void setExecuteExistingDelayedTasksAfterShutdownPolicy(boolean value) {
executeExistingDelayedTasksAfterShutdown = value;
if (!value && isShutdown())
cancelUnwantedTasks();
}
public void shutdown() {
cancelUnwantedTasks();
super.shutdown();
}
private void cancelUnwantedTasks() {
// 。
boolean keepDelayed = getExecuteExistingDelayedTasksAfterShutdownPolicy();
// 。
boolean keepPeriodic = getContinueExistingPeriodicTasksAfterShutdownPolicy();
if (!keepDelayed && !keepPeriodic)
super.getQueue().clear(); // ， 。
else if (keepDelayed || keepPeriodic) {
// 。
Object[] entries = super.getQueue().toArray();
for (int i = 0; i < entries.length; ++i) {
Object e = entries[i];
if (e instanceof RunnableScheduledFuture) {
RunnableScheduledFuture<?> t = (RunnableScheduledFuture<?>)e;
if (t.isPeriodic()? !keepPeriodic : !keepDelayed)
t.cancel(false);
}
}
entries = null;
// 。
purge();
}
}ScheduledThreadPoolExectorのコード解析は完了しました。
参照：Jdk 1.6 JUCソース解析（17）-ThreadPool Exector
Jdk 1.6 JUCソース解析（18）-DelayQue