dslztx
0%

FutureTask和ScheduledFutureTask

发表于 更新于 专题于 ， ， Disqus：

在阅读和过程中，会碰到以下几个互相之间具有联系的类和接口：

  • Runnable
  • Future
  • RunnableFuture
  • FutureTask
  • Callable
  • ScheduledFuture
  • RunnableScheduledFuture
  • ScheduledFutureTask

相关继承类图如图1所示。

图1

本文接下来基于具体实现类“FutureTask”和“ScheduledFutureTask”梳理上述类和接口的脉络关系，以便于理解。

一、FutureTask

1.1、演进

需求：向线程池提交一个任务后，能够控制过程和获取结果。

演进：

  1. 设计Future接口作为向线程池提交任务后的返回对象，通过它控制过程和获取结果
  2. 线程池中执行任务，任务的最终载体必然是Runnable接口，其与返回的Future结果属于跨线程关系，因此两者产生关联的唯一途径是共享内存，自然而然设计RunnableFuture接口
  3. 执行RunnableFuture接口的run()方法，然后将执行结果赋值到RunnableFuture实例的某个成员变量
  4. FutureTask是RunnableFuture接口的实现类，引入一个Callable callable成员变量，在run()方法中执行该Callable接口的call()方法，然后call()方法的返回结果被赋值到Object outcome成员变量
  5. 由以上点可知，FutureTask的run()方法中调度执行Callable接口的call()方法：如果在调度任务时直接传入Callable接口，自然没啥问题；但是如果传入的是Runnable接口，则需要进行适配，具体是通过Executors.RunnableAdapter类，它实现了Callable接口，查看Executors.RunnableAdapter类的唯一构造方法，须传入两个参数，一个是Runnable task，另外一个是T result，后者表示预设的call()方法返回结果，因为Runnable实例的run()方法并没有执行结果

Future接口：

1
2
3
4
5
6
7
8
9
10
11
12
13
public interface Future<V> {

    boolean cancel(boolean mayInterruptIfRunning);

    boolean isCancelled();

    boolean isDone();

    V get() throws InterruptedException, ExecutionException;

    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

RunnableFuture接口：

1
2
3
4
5
6
7
public interface RunnableFuture<V> extends Runnable, Future<V> {
    /**
     * Sets this Future to the result of its computation
     * unless it has been cancelled.
     */
    void run();
}

FutureTask类：

1
2
3
4
5
public class FutureTask<V> implements RunnableFuture<V> {
    private Callable<V> callable;

    private Object outcome; // non-volatile, protected by state reads/writes
}

Callable接口：

1
2
3
4
5
6
7
8
9
public interface Callable<V> {
    /**
     * Computes a result, or throws an exception if unable to do so.
     *
     * @return computed result
     * @throws Exception if unable to compute a result
     */
    V call() throws Exception;
}

RunnableAdapter类：

1
2
3
4
5
6
7
8
9
10
11
12
static final class RunnableAdapter<T> implements Callable<T> {
    final Runnable task;
    final T result;
    RunnableAdapter(Runnable task, T result) {
        this.task = task;
        this.result = result;
    }
    public T call() {
        task.run();
        return result;
    }
}

1.2、源码阅读

带有笔者注释的FutureTask源码版本见链接。

1.2.1、内部状态

有成员变量volatile int state，表征内部状态。

其内部状态列表如下：

  • NEW=0，初始状态
  • COMPLETING=1，转向NORMAL或者EXCEPTIONAL时的临时中间状态
  • NORMAL=2，run()方法内调用call()方法未抛异常，调用正常退出
  • EXCEPTIONAL=3，run()方法内调用call()方法抛异常，调用非正常退出
  • CANCELLED=4，调用cancel()方法时，如果传入参数mayInterruptIfRunning=false，内部状态从NEW转为CANCELLED
  • INTERRUPTING=5，调用cancel()方法时，如果传入参数mayInterruptIfRunning=true，内部状态从NEW转为INTERRUPTING
  • INTERRUPTED=6，调用cancel()方法时，如果传入参数mayInterruptIfRunning=true，内部状态从NEW转为INTERRUPTING，等发出中断信号后，内部状态再转为INTERRUPTED

1.2.2、核心方法

1、run()
执行任务，具体逻辑是：

  1. 调用成员变量callable的call()方法
  2. 根据call()方法的调用是否抛出异常分为两种情况：
    • 不抛出异常：1）内部状态经历NEW -> COMPLETING -> NORMAL；2）call()方法的返回结果赋值给成员变量outcome
    • 抛出异常：1）内部状态经历NEW -> COMPLETING -> EXCEPTIONAL；2）抛出的Throwable对象赋值给成员变量outcome

2、runAndReset()
STPE中提交的“周期性任务”调度执行时执行本方法，它的具体逻辑是：

  1. 调用成员变量callable的call()方法
  2. 根据call()方法的调用是否抛出异常分为两种情况：
    • 不抛出异常：1）内部状态仍为NEW，否则下一个周期不能执行；2）call()方法的返回结果不赋值给成员变量outcome，即使赋值了，在NEW内部状态下，调用get()方法获取结果恒挂起，调用get(long timeout, TimeUnit unit)方法获取结果超时后抛出TimeoutException异常；3）runAndReset()方法返回true值，表征该周期性任务会再被加入到workQueue队列进行后续调度[1]
    • 抛出异常：1）内部状态经历NEW -> COMPLETING -> EXCEPTIONAL；2）抛出的Throwable对象赋值给成员变量outcome；3）runAndReset()方法返回false值，表征该周期性任务不会再被加入到workQueue队列进行后续调度[1]

3、cancel(boolean mayInterruptIfRunning)
调用cancel(boolean mayInterruptIfRunning)方法尝试取消run()或者runAndReset()方法的运行。

其核心逻辑是：

  • 如果当前内部状态不为NEW，表明已经执行完成，不能再取消
  • 如果当前内部状态为NEW，表明未执行完成，尝试取消：
    • 如果传入参数mayInterruptIfRunning=false，则经历内部状态从NEW转为CANCELLED过程
    • 如果传入参数mayInterruptIfRunning=true，则经历内部状态从NEW转为INTERRUPTING -> 给正在执行当前任务的线程发中断信号 -> 内部状态转为INTERRUPTED过程

4、isCancelled()
当内部状态是CANCELLED、INTERRUPTING或者INTERRUPTED时（即成功调用cancel(boolean mayInterruptIfRunning)方法的可能结果），返回true，否则返回false。

5、isDone()
当内部状态不是NEW时，返回true，否则返回false。

6、get()
获取run()或者runAndReset()方法的执行结果：

  • 当内部状态<=COMPLETING时，恒挂起
  • 当内部状态>COMPLETING时，分为3种情况：
    • 内部状态为NORMAL，表示成员变量callable的call()方法执行未抛出异常，成功退出，这里返回成员变量outcome值（即call()方法的返回结果）
    • 内部状态为EXCEPTIONAL，表示成员变量callable的call()方法执行抛出异常，这里抛出ExecutionException异常
    • 内部状态>=CANCELLED，即为CANCELLED、INTERRUPTING或者INTERRUPTED，表示被取消执行，这里抛出CancellationException异常

7、get(long timeout, TimeUnit unit)
获取run()或者runAndReset()方法的执行结果：

  • 当内部状态<=COMPLETING时，挂起指定时间，如果到期仍是<=COMPLETING，则抛出TimeoutException异常
  • 当内部状态>COMPLETING时，分为3种情况：
    • 内部状态为NORMAL，表示成员变量callable的call()方法执行未抛出异常，成功退出，这里返回成员变量outcome值（即call()方法的返回结果）
    • 内部状态为EXCEPTIONAL，表示成员变量callable的call()方法执行抛出异常，这里抛出ExecutionException异常
    • 内部状态>=CANCELLED，即为CANCELLED、INTERRUPTING或者INTERRUPTED，表示被取消执行，这里抛出CancellationException异常

二、ScheduledFutureTask

相较于FutureTask，ScheduledFutureTask额外继承实现了3个接口的方法：

  • Comparable接口的public int compareTo(T o)方法，ScheduledFutureTask被加入到STPE中属于DelayedWorkQueue类型的workQueue，DelayedWorkQueue是一个优先级队列，需要其中的元素实现Comparable接口的上述方法
  • Delay接口的long getDelay(TimeUnit unit)方法，获取任务计划执行时间与当前时间的时间差值，时间差值的单位由“TimeUnit unit”指定
  • RunnableScheduledFuture接口的boolean isPeriodic()方法，判断当前任务是否是周期性任务

三、关于源码的两个困惑

3.1、困惑1

困惑1描述：调用get()/get(long timeout, TimeUnit unit)方法获取set(V v)/setException(Throwable t)方法中设置的变量outcome结果时，如何证明设置的值对其是可见的。

1
2
3
4
5
6
7
8
9
protected void set(V v) {
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        outcome = v;                                                            //1

        //上面有介绍，取巧的做法，不会影响结果
        UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state         //2
        finishCompletion();
    }
}
1
2
3
4
5
6
7
8
9
10
protected void setException(Throwable t) {
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        outcome = t;                                                             //3


        //上面有介绍，取巧的做法，不会影响结果
        UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state     //4
        finishCompletion();
    }
}
1
2
3
4
5
6
public V get() throws InterruptedException, ExecutionException {
    int s = state;
    if (s <= COMPLETING)
        s = awaitDone(false, 0L);
    return report(s);
}
1
2
3
4
5
6
7
8
9
10
private V report(int s) throws ExecutionException {
                                                                                 //5，表示这里必有state>COMPLETING

    Object x = outcome;                                                          //6
    if (s == NORMAL)
        return (V)x;
    if (s >= CANCELLED)
        throw new CancellationException();
    throw new ExecutionException((Throwable)x);
}

以get()方法为例进行说明（get(long timeout, TimeUnit unit)方法的说明是类似的），相关源代码已贴在上面。

具体证明如下：根据get()方法的实现逻辑可知，在调用report()方法之前必有state>COMPLETING，本讨论中只关心两个状态NORMAL和EXCEPTIONAL，NORMAL状态通过set()方法中的UNSAFE.putOrderedInt(this, stateOffset, NORMAL)语句设置/EXCEPTIONAL状态通过setException(Throwable t)方法中的UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL)语句设置，state是volatile变量，在get()方法或者awaitDone(boolean timed, long nanos)方法中以“一般直白”形式读取故具有volatile变量读语义，根据《Unsafe类》中的“广义的volatile变量happens-before规则”，再结合happens-before规则——“程序顺序规则”和“传递性规则”，有“//1 -hb-> //2 -hb-> //5 -hb-> //6”或者“//3 -hb-> //4 -hb-> //5 -hb> //6”，因此//1处或者//3处对outcome的设置对//6处可见。

3.2、困惑2

困惑2描述：调用cancel(true)方法发送的中断信号可能会泄漏而递延到下一个任务的执行过程，而导致影响其执行吗？答案是否定的。

根据具体执行方法是run()还是runAndReset()（STPE中提交的“周期性任务”调度执行时执行该方法）分为两种情形进行讨论证明。

3.2.1、run()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
public boolean cancel(boolean mayInterruptIfRunning) {
    if (!(state == NEW &&
          UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
              mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))       //1
        return false;
    try {    // in case call to interrupt throws exception
        if (mayInterruptIfRunning) {
            try {
                Thread t = runner;
                if (t != null)
                    t.interrupt();                                      //2
            } finally { // final state
                //上面有介绍，取巧的做法，不会影响结果
                UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);   //3
            }
        }
    } finally {
        finishCompletion();
    }
    return true;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
public void run() {
    if (state != NEW ||
        !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                     null, Thread.currentThread()))
        return;
    try {
        Callable<V> c = callable;
        if (c != null && state == NEW) {
            V result;
            boolean ran;
            try {
                result = c.call();
                ran = true;
            } catch (Throwable ex) {
                result = null;
                ran = false;
                setException(ex);
            }
            if (ran)
                set(result);
        }
    } finally {
        // runner must be non-null until state is settled to
        // prevent concurrent calls to run() 见最开始的判断语句
        runner = null;
        // state must be re-read after nulling runner to prevent
        // leaked interrupts
        int s = state;                                                  //4
        if (s >= INTERRUPTING)   
            handlePossibleCancellationInterrupt(s);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
private void handlePossibleCancellationInterrupt(int s) {
    // It is possible for our interrupter to stall before getting a
    // chance to interrupt us.  Let's spin-wait patiently.
    if (s == INTERRUPTING)
        while (state == INTERRUPTING)
            Thread.yield(); // wait out pending interrupt

    // assert state == INTERRUPTED;

    // We want to clear any interrupt we may have received from
    // cancel(true).  However, it is permissible to use interrupts
    // as an independent mechanism for a task to communicate with
    // its caller, and there is no way to clear only the
    // cancellation interrupt.
    //
    // Thread.interrupted();

    // 就是有可能c.call()中有发出有用的中断信号，不能区分cancel()方法发出的还是c.call()发出的，故不能复位中断信号
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;

    //针对的是构造时的`setState(-1)`语句，后续可以调用`interruptIfStarted()`方法了，否则是没有效果的
    w.unlock(); // allow interrupts

    boolean completedAbruptly = true;
    try {
        while (task != null || (task = getTask()) != null) {
            w.lock();

            // If pool is stopping, ensure thread is interrupted;
            // if not, ensure thread is not interrupted.  This
            // requires a recheck in second case to deal with
            // shutdownNow race while clearing interrupt

            //就是只有处于STOP状态，已经产生的interrupt信号才传递给任务，否则都清理掉；处于STOP状态，如果没有传递的interrupt信号，则自中断
            if ((runStateAtLeast(ctl.get(), STOP) ||
                 (Thread.interrupted() &&
                  runStateAtLeast(ctl.get(), STOP))) &&
                !wt.isInterrupted())                                        //5
                wt.interrupt();                                             
    ...
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
private Runnable getTask() {
    boolean timedOut = false; // Did the last poll() time out?

    for (;;) {
       
        ... 

        try {
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
            if (r != null)
                return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}

1、前导

  1. 在cancel(boolean mayInterruptIfRunning)方法中，Thread t = runner语句内隐含的加载runner操作不能被重排序到UNSAFE.compareAndSwapInt(this, stateOffset, NEW, mayInterruptIfRunning ? INTERRUPTING : CANCELLED)语句之前，因为结合《synchronized-volatile-final关键词》和《Unsafe类》可知，后者隐式带有StoreLoad绑定型内存屏障，故后续的Load操作不能被重排序到该Store操作之前
  2. 同理，在run()方法中，int s = state语句内隐含的加载state操作不能被重排序到会调用的set(V v)或者setException(Throwable t)方法内的UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)语句之前
  3. 同理，在run()方法中，int s = state语句内隐含的加载state操作不能被重排序到runner = null语句之前，因为后者是一个volatile变量写入，隐式带有StoreLoad内存屏障

2、具体证明
根据以上贴出的源码，进行具体证明，以“//4处加载到最新state时机”为讨论基点S，根据在S点时//1处CAS操作的执行情况分为3类完备情况（如果不是CAS操作，考虑到可见性问题，则有不只3类完备情况）：

  1. 在S点时，//1处操作已执行且返回true
    • 如果此时S点的state=CANCELLED，则不发出中断信号
    • 如果此时S点的state=INTERRUPTING | INTERRUPTED，需要注意的是，INTERRUPTED状态通过UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED)语句设置，并不立即可见。查看//4处后handlePossibleCancellationInterrupt(s)方法，存在一个确保逻辑：如果当前state=INTERRUPTING，则进入一个循环，直到state!=INTERRUPTING，根据可能的state状态转移，此时必有state=INTERRUPTED。根据《Unsafe类》中的“广义的volatile变量happens-before规则”，再结合happens-before规则——“程序顺序规则”和“传递性规则”，有//2 -hb-> //3 -hb-> TPE类中runWorker()方法内的下一轮getTask()语句执行 -hb-> TPE类中runWorker()方法内的下一轮//5语句执行，如果//2处实际上未执行到，自然不发出中断信号；如果//2处实际执行到了，且出现中断信号泄漏（如果该中断信号在真实任务执行逻辑内部被复位掉，即c.call();内部，则没有泄漏），根据分析可知，此时该泄漏的中断信号必然会在getTask()方法的workQueue.poll或者take语句处或者//5处（当执行器的状态为RUNNING）被复位，故不会递延到下一个任务的执行过程，得证
  2. 在S点时，//1处操作已执行且返回false，此时不发出中断信号
  3. 在S点时，//1处操作未执行，根据前导第2点，S点时必有state>=COMPLETING，此时执行//1处操作会返回false，故不发出中断信号

3、其他
run()方法中int s = state语句前有一个注释state must be re-read after nulling runner to prevent leaked interrupts，就是说在这里需要加载最新的state，避免中断信号泄漏。接下来给出一种不重新加载最新state导致中断信号泄漏的情形：线程T2执行cancel(true)方法，执行到T1.interrupt()语句时等待CPU资源，此时state的状态为INTERRUPTING，线程T1执行run()方法，执行到if (s >= INTERRUPTING)语句时，由于s没有重新加载最新的state值，因此还是NEW，条件判断失败，不会执行handlePossibleCancellationInterrupt(s)方法，继续执行直到下一个真实任务的执行逻辑，此时T2分配到CPU资源，执行T1.interrupt()语句，发出中断信号，即出现了“中断信号泄漏”。

3.2.2、runAndReset()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
protected boolean runAndReset() {
    if (state != NEW ||
        !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                     null, Thread.currentThread()))
        return false;
    boolean ran = false;
    int s = state;
    try {
        Callable<V> c = callable;
        if (c != null && s == NEW) {
            try {
                c.call(); // don't set result
                ran = true;
            } catch (Throwable ex) {
                setException(ex);
            }
            }
    } finally {
        // runner must be non-null until state is settled to
        // prevent concurrent calls to run()
        runner = null;
        // state must be re-read after nulling runner to prevent
        // leaked interrupts
        s = state;                                                //44
        if (s >= INTERRUPTING)
            handlePossibleCancellationInterrupt(s);
    }

    //只有成功执行，未抛出异常，才是一个合法的周期性任务
    return ran && s == NEW;
}

1、前导

  1. 在cancel(boolean mayInterruptIfRunning)方法中，Thread t = runner语句内隐含的加载runner操作不能被重排序到UNSAFE.compareAndSwapInt(this, stateOffset, NEW, mayInterruptIfRunning ? INTERRUPTING : CANCELLED)语句之前，因为结合《synchronized-volatile-final关键词》和《Unsafe类》可知，后者隐式带有StoreLoad绑定型内存屏障，故后续的Load操作不能被重排序到该Store操作之前
  2. 同理，在runAndReset()方法中，int s = state语句内隐含的加载state操作不能被重排序到会调用的setException(Throwable t)方法内的UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)语句之前
  3. 同理，在runAndReset()方法中，int s = state语句内隐含的加载state操作不能被重排序到runner = null语句之前，因为后者是一个volatile变量写入，隐式带有StoreLoad内存屏障

2、具体证明
根据以上贴出的源码和“3.2.1、run()”小节中贴出的源码，进行具体证明，以“//44处加载到最新state时机”为讨论基点S，根据在S点时//1处CAS操作的执行情况分为3类完备情况（如果不是CAS操作，考虑到可见性问题，则有不只3类完备情况）：

  1. 在S点时，//1处操作已执行且返回true
    • 如果此时S点的state=CANCELLED，则不发出中断信号
    • 如果此时S点的state=INTERRUPTING | INTERRUPTED，需要注意的是，INTERRUPTED状态通过UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED)语句设置，并不立即可见。查看//44处后handlePossibleCancellationInterrupt(s)方法，存在一个确保逻辑：如果当前state=INTERRUPTING，则进入一个循环，直到state!=INTERRUPTING，根据可能的state状态转移，此时必有state=INTERRUPTED。根据《Unsafe类》中的“广义的volatile变量happens-before规则”，再结合happens-before规则——“程序顺序规则”和“传递性规则”，有//2 -hb-> //3 -hb-> TPE类中runWorker()方法内的下一轮getTask()语句执行 -hb-> TPE类中runWorker()方法内的下一轮//5语句执行，如果//2处实际上未执行到，自然不发出中断信号；如果//2处实际执行到了，且出现中断信号泄漏（如果该中断信号在真实任务执行逻辑内部被复位掉，即c.call();内部，则没有泄漏），根据分析可知，此时该泄漏的中断信号必然会在getTask()方法的workQueue.poll或者take语句处或者//5处（当执行器的状态为RUNNING）被复位，故不会递延到下一个任务的执行过程，得证
  2. 在S点时，//1处操作已执行且返回false，此时不发出中断信号
  3. 在S点时，//1处操作未执行：
    • 如果有执行setException(ex)语句，根据前导第2点，S点时必有state>=COMPLETING，此时执行//1处操作会返回false，故不发出中断信号
    • 如果未有执行setException(ex)语句，根据前导第1和第3点，此时执行//1处操作，runner = null语句必然已经执行完且结果对其可见，Thread t = runner语句不可能重排序到//1处操作之前，故if (t != null)判断语句结果为false，不会发出中断信号

3、其他
针对“在S点时，//1处操作未执行，且未执行setException(ex)语句”情形，本来笔者还认为存在一种“中断信号泄漏递延给下一个任务”的可能，具体是“线程T1执行runAndReset()方法，S点时有state=NEW，runner = null;语句重排序到//44之后，故此时runner!=NULL，另外一个线程T2在S点时执行cancel(true)方法，执行到T1.interrupt()语句时等待CPU资源分配，T1继续执行直到下一个真实任务的执行逻辑（不会调用handlePossibleCancellationInterrupt(s)方法，因为if (s >= INTERRUPTING)判断语句返回false），然后T2此时发出中断信号，该中断信号对新任务的执行过程可见，即出现了‘中断信号泄漏’”。针对以上“中断信号泄漏”可能，还特地向OpenJDK反馈了一个Bug，后面根据Doug Lea的回复——The case presented here includes assumption of a reordering that should be precluded because fields state and runner are volatile.，才意识到了以上分析的错误在于没有考虑到前导第1和第3点。

四、其他

根据以下TPE类中runWorker(Worker w)方法源码可知，在TPE和STPE中调度执行的任务，必须在内部catch住自身过程抛出的Throwable，否则会导致执行该任务的Worker异常退出（虽然会被弥补）。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
final void runWorker(Worker w) {
    try {
        while (task != null || (task = getTask()) != null) {
            w.lock();

            ...

            try {
                beforeExecute(wt, task);
                Throwable thrown = null;
                try {
                    task.run();                    //1
                } catch (RuntimeException x) {
                    thrown = x; throw x;
                } catch (Error x) {
                    thrown = x; throw x;
                } catch (Throwable x) {
                    thrown = x; throw new Error(x);
                } finally {
                    afterExecute(task, thrown);
                }
            } finally {
                task = null;
                w.completedTasks++;
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        processWorkerExit(w, completedAbruptly);
    }
}

接下来针对TPE和STPE中的所有任务提交形式一一进行分析说明。

4.1、TPE

存在4种提交形式，分为两类：

  1. 第一类：//1处的task就是真实的任务实体，在该情形中，真实任务实体的执行方法必须在内部catch住自身过程抛出的Throwable
    • public void execute(Runnable command)，真实任务实体是command，必须在command.run()方法内部catch住自身过程抛出的Throwable
  2. 第二类：//1处的task指代FutureTask实例，查看FutureTask.run()方法源码，可知在调用真实任务实体执行方法的外围有catch住异常的逻辑，故不会导致Worker异常退出，但是为统一起见，建议还是在真实任务实体执行方法内部catch住自身过程抛出的Throwable
    • public <T> Future<T> submit(Callable<T> task)，真实任务实体是task，建议在task.call()方法内部catch住自身过程抛出的Throwable
    • public Future<?> submit(Runnable task)，真实任务实体是task，建议在task.run()方法内部catch住自身过程抛出的Throwable
    • public <T> Future<T> submit(Runnable task, T result)，真实任务实体是task，建议在task.run()方法内部catch住自身过程抛出的Throwable

4.2、STPE

存在8种提交形式，包括从TPE继承下来的4种提交形式（在STPE中覆盖实现），查看源码可知//1处的task指代ScheduledFutureTask实例，查看ScheduledFutureTask.run()方法源码，其转发调用FutureTask的run()或者runAndReset()方法，查看上述两个方法源码，可知在调用真实任务实体执行方法的外围有catch住异常的逻辑，故不会导致Worker异常退出，但是为统一起见，建议还是在真实任务实体执行方法内部catch住自身过程抛出的Throwable：

  • 继承下来的4个方法说明见“4.1、TPE”小节
  • public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit)，真实任务实体是command，建议在command.run()方法内部catch住自身过程抛出的Throwable
  • public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit)，真实任务实体是callable，建议在callable.call()方法内部catch住自身过程抛出的Throwable
  • public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit)，真实任务实体是command，建议在command.run()方法内部catch住自身过程抛出的Throwable
  • public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit)，真实任务实体是command，建议在command.run()方法内部catch住自身过程抛出的Throwable

参考文献

[1]《ScheduledThreadPoolExecutor》中的ScheduledFutureTask类的run()方法
[2]https://www.cnblogs.com/thisiswhy/p/13791966.html
[3]《Unsafe类》

您的支持将鼓励我继续分享！