Java併發-ScheduledThreadPoolExecutor

摘要： A ThreadPoolExecutor that can additionally schedule commands to run after a given delay, or to execute periodically（ 週期性地 ） . This class is prefer...

A ThreadPoolExecutor that can additionally schedule commands to run after a given delay, or to execute periodically（ 週期性地 ） . This class is preferable to（ 優先於…… ） Timer when multiple worker threads are needed, or when the additional flexibility（ 靈活性 ） or capabilities of ThreadPoolExecutor (which this class extends) are required.

Delayed tasks execute no sooner than they are enabled, but without any real-time guarantees about when, after they are enabled, they will commence. Tasks scheduled for exactly the same execution time are enabled in first-in-first-out (FIFO) order of submission.

When a submitted task is cancelled before it is run, execution is suppressed. By default, such a cancelled task is not automatically removed from the work queue until its delay elapses. While this enables further inspection and monitoring, it may also cause unbounded retention of cancelled tasks. To avoid this, set setRemoveOnCancelPolicy to true, which causes tasks to be immediately removed from the work queue at time of cancellation .

Extension notes : This class overrides the execute and submit methods to generate internal ScheduledFuture objects to control per-task delays and scheduling. To preserve functionality, any further overrides of these methods in subclasses must invoke superclass versions, which effectively disables additional task customization. However, this class provides alternative protected extension method decorateTask (one version each for Runnable and Callable) that can be used to customize the concrete task types used to execute commands entered via execute , submit , schedule , scheduleAtFixedRate , and scheduleWithFixedDelay . By default, a ScheduledThreadPoolExecutor uses a task type extending FutureTask .

類層次結構

構造方法

public ScheduledThreadPoolExecutor(int corePoolSize) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue());
}

public ScheduledThreadPoolExecutor(int corePoolSize,
ThreadFactory threadFactory) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue(), threadFactory);
}

public ScheduledThreadPoolExecutor(int corePoolSize,
RejectedExecutionHandler handler) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue(), handler);
}

public ScheduledThreadPoolExecutor(int corePoolSize,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue(), threadFactory, handler);
}
複製程式碼

執行和排程任務方法

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) 

public Future<?> submit(Runnable task)

public <T> Future<T> submit(Runnable task, T result)

public <T> Future<T> submit(Callable<T> task)

複製程式碼

其他方法

public void setContinueExistingPeriodicTasksAfterShutdownPolicy(boolean value)

public boolean getContinueExistingPeriodicTasksAfterShutdownPolicy()

public void setExecuteExistingDelayedTasksAfterShutdownPolicy(boolean value)

public boolean getExecuteExistingDelayedTasksAfterShutdownPolicy()

public void setRemoveOnCancelPolicy(boolean value)

public boolean getRemoveOnCancelPolicy()

public void shutdown()

public List<Runnable> shutdownNow()

public BlockingQueue<Runnable> getQueue()

複製程式碼

實現原理

ScheduledThreadPoolExecutor 排程和執行任務的過程可以抽象如下圖所示：

1. 將 Callable 或 Runnable 物件轉換 ScheduledFutureTask 物件；
2. 將轉換後的 ScheduledFutureTask 物件新增到延遲佇列並開啟執行緒執行任務；
3. 工作執行緒從佇列獲取 ScheduledFutureTask 任務執行任務。

通過上述描述可以發現， ScheduledFutureTask 是 ScheduledThreadPoolExecutor 實現的關鍵。

原始碼分析

public ScheduledFuture<?> schedule(Runnable command,
long delay,
TimeUnit unit) {
// 步驟1
RunnableScheduledFuture<?> t = decorateTask(command,
new ScheduledFutureTask<Void>(command, null,
triggerTime(delay, unit)));
// 步驟2
delayedExecute(t);
return t;
}

public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
long initialDelay,
long period,
TimeUnit unit) {
ScheduledFutureTask<Void> sft =
new ScheduledFutureTask<Void>(command,
null,
triggerTime(initialDelay, unit),
unit.toNanos(period));
RunnableScheduledFuture<Void> t = decorateTask(command, sft);
sft.outerTask = t;
delayedExecute(t);
return t;
}

// 將任務新增到延遲佇列
private void delayedExecute(RunnableScheduledFuture<?> task) {
if (isShutdown())
reject(task);
else {
super.getQueue().add(task);
if (isShutdown() &&
!canRunInCurrentRunState(task.isPeriodic()) &&
remove(task))
task.cancel(false);
else
ensurePrestart();
}
}
複製程式碼

ScheduledFutureTask

類層次結構

構造方法

ScheduledFutureTask(Runnable r, V result, long ns) {
super(r, result);
this.time = ns;
this.period = 0;
this.sequenceNumber = sequencer.getAndIncrement();
}

ScheduledFutureTask(Runnable r, V result, long ns, long period) {
super(r, result);
this.time = ns;
this.period = period;
this.sequenceNumber = sequencer.getAndIncrement();
}

ScheduledFutureTask(Callable<V> callable, long ns) {
super(callable);
this.time = ns;
this.period = 0;
this.sequenceNumber = sequencer.getAndIncrement();
}
}
複製程式碼

任務執行

public void run() {
boolean periodic = isPeriodic();
if (!canRunInCurrentRunState(periodic))
cancel(false);
else if (!periodic) // 非週期性任務，直接執行
ScheduledFutureTask.super.run();
else if (ScheduledFutureTask.super.runAndReset()) { // 執行任務，重置狀態
// 計算下一次執行時間
setNextRunTime();
// 重新新增到延遲佇列
reExecutePeriodic(outerTask);
}
}

void reExecutePeriodic(RunnableScheduledFuture<?> task) {
if (canRunInCurrentRunState(true)) {
super.getQueue().add(task);
if (!canRunInCurrentRunState(true) && remove(task))
task.cancel(false);
else
ensurePrestart();
}
}
複製程式碼