FutureTask详解

FutureTask详解

简介

• FutureTask为Future的实现类.
• 用以获取任务执行结果(get)和取消任务(cancel).
• 若任务未完成,则获取任务结果时会被阻塞.
• 若任务执行完成后,任务不能被重启或取消.
• 可用作一个任务提交到线程中执行.
• 线程安全由CAS保证.
• FutureTask实现RunnableFuture接口,该接口继承了Runnable接口和Future接口.FutureTask可以被Thread直接执行,也可以作为Future用作Callable的计算结果.

用法

Callable接口

有返回值,可以返回子任务执行的结果.

Future接口

表示异步计算的结果,可以:

• 查询异步计算任务是否执行完成
• 获取异步计算的结果
• 中断异步任务的执行

接口定义

boolean cancel(boolean mayInterruptIfRunning);
boolean isCancelled();
boolean isDone();
V get() throws InterruptedException, ExecutionException;
V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;

解释:

• cancel(): 取消子任务的执行.
  • 若子任务已执行结束/已被取消/不能取消,则方法执行失败,返回false.
  • 若子任务没有开始执行,子任务被取消,不再被执行.
  • 若子任务已开始执行,没有执行结束,则根据mayInterruptIfRunning判断:
    • 若为true,则段执行任务的线程,返回true.
    • 若为false,则返回true,不中断执行任务的线程.
• isCancelled(): 判断任务是否被取消.
  • 若任务执行结束(正常结束,发生异常结束)前被取消,返回true.(即调用cancel()返回true)
  • 否则返回false.
• idDone(): 判断任务是否执行结束.(以下情况返回true,否则返回false)
  • 正常执行结束.
  • 发生异常结束.
  • 被取消.
• get(): 获取结果.
  • 若任务没有执行结束,则调用线程进入阻塞状态.
  • 若任务已被取消,则抛出CancellationException异常.
  • 若执行过程中抛出异常,则抛出ExecutionException异常.
  • 若当前线程在阻塞时被中断,抛出InterruptedException异常.
• get(long timeout, TimeUnit unit):设定超时限制的get(),等待超时后,抛出TimeOutException异常.

FutureTask

成员变量

1. state: 用来表示当前任务的运行状态.
   • NEW(=0): 新任务/没有执行完的任务.
   • COMPLETING(=1): 任务执行结束(正常结束,发生异常结束),但结果没有保存到outcome中.
   • NORMAL: 任务正常执行结束,结果保存到outcome中.
   • CANCELLED: 任务执行结束前被取消,不要求中断正在执行的线程.
   • INTERRUPTING: 任务执行结束之前被取消,要求中断线程的执行.
   • INTERRUPTED: 调用cancel(true)取消异步任务,会调用interrupt()中断线程的执行.
2. callable: 封装计算任务,获取计算结果.
3. outcome: 保存计算任务的返回结果/执行过程中抛出的异常.
4. runner: 指向当前运行的callable任务的线程.
5. waiters: 任务没有结束,调用get()方法的线程会被阻塞,进入阻塞队列排队等待.

状态转移:

• NEW-->COMPLETING-->NORMAL : 正常执行结束
• NEW-->COMPLETING-->EXCEPTIONAL : 执行过程中发生异常
• NEW-->CANCELLED : 被取消,调用cancel(false)
• NEW-->INTERRUPTING-->INTERRUPTED : 被中断,调用cancel(false)

核心方法

任务执行 run()

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 = null;
        int s = state;
        if (s >= INTERRUPTING)
            handlePossibleCancellationInterrupt(s);
    }
}

流程:

1. 如果任务不是新建的任务或者无法运行该任务,则直接返回,不作处理.
2. 获取任务并检查任务是否是新建的,满足条件则试图执行任务.
3. 若执行任务过程中出现异常任务未完成,则清空任务,并设置任务状态为EXCEPTIONAL.
4. 检查任务是否完成,完成则设置状态为NORMAL.
5. 最后检查任务是否被中断过.若被中断过,则响应中断.

获取任务的执行结果 get()

public V get() throws InterruptedException, ExecutionException {
    int s = state;
    if (s <= COMPLETING)
        s = awaitDone(false, 0L);
    return report(s); // 若正常结束,则返回结果.否则抛出异常
}

流程:

1. 获取任务状态.
2. 若未完成,则等待完成后返回结果.
3. 若已完成,则直接返回结果.
4. 若被取消或执行过程中出现异常,则抛出异常.

取消任务 cancel(boolean mayInterruptIfRunning)

public boolean cancel(boolean mayInterruptIfRunning) {
    if (!(state == NEW &&
            UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
                mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
        return false;
    try {
        if (mayInterruptIfRunning) {
            try {
                Thread t = runner;
                if (t != null)
                    t.interrupt();
            } finally {
                UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
            }
        }
    } finally {
        finishCompletion();
    }
    return true;
}

流程:

1. 若任务已执行完成或者无法设置状态为中断或取消时,则直接返回.
2. 设置为要求中断,则试图中断线程,并设置任务状态为INTERRUPETED.
3. 唤醒等待队列的线程.

子线程返回结果的最后一步 finishCompletion()

private void finishCompletion() {
    for (WaitNode q; (q = waiters) != null;) {
        if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
            for (;;) {
                Thread t = q.thread;
                if (t != null) {
                    q.thread = null;
                    LockSupport.unpark(t);
                }
                WaitNode next = q.next;
                if (next == null)
                    break;
                q.next = null;
                q = next;
            }
            break;
        }
    }

    done();

    callable = null;
}

解释: 在子线程返回结果时,试图唤醒等待队列中的所有阻塞线程.
对于设置阻塞等待时间的线程,若阻塞时间未到但任务已经返回结果,则无需继续等待,直接返回结果.

小结

FutureTask主要采用CAS来替代锁,实现多线程并发.

参考:

• 可取消的异步任务——FutureTask用法及解析