Java併發-ScheduledThreadPoolExecutor

方法註釋等完善

Overview

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();
    }
}
複製代碼