java.util.concurrent.FutureTask 源码

线程池相关

源码：

package java.util.concurrent;

import java.util.concurrent.locks.LockSupport;

public class FutureTask<V> implements RunnableFuture<V> {

    private volatile int state;
    private static final int NEW = 0;
    private static final int COMPLETING = 1;
    private static final int NORMAL = 2;
    private static final int EXCEPTIONAL = 3;
    private static final int CANCELLED = 4;
    private static final int INTERRUPTING = 5;
    private static final int INTERRUPTED = 6;


    private Callable<V> callable;

    private Object outcome;

    private volatile Thread runner;

    private volatile WaitNode waiters;


    static final class WaitNode {
        volatile Thread thread;
        volatile WaitNode next;

        WaitNode() {
            thread = Thread.currentThread();
        }
    }

    public FutureTask(Callable<V> callable) {
        if (callable == null)
            throw new NullPointerException();
        this.callable = callable;
        this.state = NEW;
    }


    public FutureTask(Runnable runnable, V result) {
        this.callable = Executors.callable(runnable, result);
        this.state = NEW;
    }

    public boolean isCancelled() {
        return state >= CANCELLED;
    }

    public boolean isDone() {
        return state != NEW;
    }

    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;
    }


    public V get() throws InterruptedException, ExecutionException {
        int s = state;
        if (s <= COMPLETING)
            s = awaitDone(false, 0L);
        return report(s);
    }


    public V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
        if (unit == null)
            throw new NullPointerException();
        int s = state;
        if (s <= COMPLETING && (s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING)
            throw new TimeoutException();
        return report(s);
    }

    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);
        }
    }

    @SuppressWarnings("unchecked")
    private V report(int s) throws ExecutionException {
        Object x = outcome;
        if (s == NORMAL)
            return (V) x;
        if (s >= CANCELLED)
            throw new CancellationException();
        throw new ExecutionException((Throwable) x);
    }


    protected void done() {
    }


    protected void set(V v) {
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
            outcome = v;
            UNSAFE.putOrderedInt(this, stateOffset, NORMAL);
            finishCompletion();
        }
    }


    protected void setException(Throwable t) {
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
            outcome = t;
            UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL);
            finishCompletion();
        }
    }


    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();
                    ran = true;
                } catch (Throwable ex) {
                    setException(ex);
                }
            }
        } finally {

            runner = null;

            s = state;
            if (s >= INTERRUPTING)
                handlePossibleCancellationInterrupt(s);
        }
        return ran && s == NEW;
    }

    private void handlePossibleCancellationInterrupt(int s) {
        if (s == INTERRUPTING)
            while (state == INTERRUPTING)
                Thread.yield();
    }


    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;
    }


    private int awaitDone(boolean timed, long nanos) throws InterruptedException {
        final long deadline = timed ? System.nanoTime() + nanos : 0L;
        WaitNode q = null;
        boolean queued = false;
        for (; ; ) {
            if (Thread.interrupted()) {
                removeWaiter(q);
                throw new InterruptedException();
            }

            int s = state;
            if (s > COMPLETING) {
                if (q != null)
                    q.thread = null;
                return s;
            } else if (s == COMPLETING)
                Thread.yield();
            else if (q == null)
                q = new WaitNode();
            else if (!queued)
                queued = UNSAFE.compareAndSwapObject(this, waitersOffset, q.next = waiters, q);
            else if (timed) {
                nanos = deadline - System.nanoTime();
                if (nanos <= 0L) {
                    removeWaiter(q);
                    return state;
                }
                LockSupport.parkNanos(this, nanos);
            } else
                LockSupport.park(this);
        }
    }


    private void removeWaiter(WaitNode node) {
        if (node != null) {
            node.thread = null;
            retry:
            for (; ; ) {
                for (WaitNode pred = null, q = waiters, s; q != null; q = s) {
                    s = q.next;
                    if (q.thread != null)
                        pred = q;
                    else if (pred != null) {
                        pred.next = s;
                        if (pred.thread == null)
                            continue retry;
                    } else if (!UNSAFE.compareAndSwapObject(this, waitersOffset, q, s))
                        continue retry;
                }
                break;
            }
        }
    }

    // Unsafe方法
    private static final sun.misc.Unsafe UNSAFE;
    private static final long stateOffset;
    private static final long runnerOffset;
    private static final long waitersOffset;

    static {
        try {
            UNSAFE = sun.misc.Unsafe.getUnsafe();
            Class<?> k = FutureTask.class;
            stateOffset = UNSAFE.objectFieldOffset
                    (k.getDeclaredField("state"));
            runnerOffset = UNSAFE.objectFieldOffset
                    (k.getDeclaredField("runner"));
            waitersOffset = UNSAFE.objectFieldOffset
                    (k.getDeclaredField("waiters"));
        } catch (Exception e) {
            throw new Error(e);
        }
    }

}

类 FutureTask<V>

类型参数：

V - 此 FutureTask 的 get 方法所返回的结果类型。

所有已实现的接口：

Runnable, Future<V>, RunnableFuture<V>

可取消的异步计算。利用开始和取消计算的方法、查询计算是否完成的方法和获取计算结果的方法，此类提供了对 Future 的基本实现。

仅在计算完成时才能获取结果；如果计算尚未完成，则阻塞 get 方法。一旦计算完成，就不能再重新开始或取消计算。

可使用 FutureTask 包装 Callable 或 Runnable 对象。因为 FutureTask 实现了 Runnable，所以可将 FutureTask 提交给 Executor 执行。

除了作为一个独立的类外，此类还提供了 protected 功能，这在创建自定义任务类时可能很有用。

构造方法摘要

FutureTask(Callable<V> callable)
创建一个 FutureTask，一旦运行就执行给定的 Callable。

FutureTask(Runnable runnable, V result)
创建一个 FutureTask，一旦运行就执行给定的 Runnable，并安排成功完成时 get 返回给定的结果。

方法摘要

`boolean`	`cancel(boolean mayInterruptIfRunning)` 试图取消对此任务的执行。
`protected void`	`done()` 当此任务转换到状态 isDone（不管是正常地还是通过取消）时，调用受保护的方法。
`V`	`get()` 如有必要，等待计算完成，然后获取其结果。
`V`	`get(long timeout, TimeUnit unit)` 如有必要，最多等待为使计算完成所给定的时间之后，获取其结果（如果结果可用）。
`boolean`	`isCancelled()` 如果在任务正常完成前将其取消，则返回 true。
`boolean`	`isDone()` 如果任务已完成，则返回 true。
`void`	`run()` 除非已将此 Future 取消，否则将其设置为其计算的结果。
`protected boolean`	`runAndReset()` 执行计算而不设置其结果，然后将此 Future 重置为初始状态，如果计算遇到异常或已取消，则该操作失败。
`protected void`	`set(V v)` 除非已经设置了此 Future 或已将其取消，否则将其结果设置为给定的值。
`protected void`	`setException(Throwable t)` 除非已经设置了此 Future 或已将其取消，否则它将报告一个 ExecutionException，并将给定的 throwable 作为其原因。

从类 java.lang.Object 继承的方法

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

FutureTask

public FutureTask(Callable<V> callable)

创建一个 FutureTask，一旦运行就执行给定的 Callable。

参数：

callable - 可调用的任务。

抛出：

NullPointerException - 如果 callable 为 null。

FutureTask

public FutureTask(Runnable runnable,
                  V result)

创建一个 FutureTask，一旦运行就执行给定的 Runnable，并安排成功完成时 get 返回给定的结果。

参数：

runnable - 可运行的任务。

result - 成功完成时要返回的结果。如果不需要特定的结果，则考虑使用下列形式的构造： Future<?> f = new FutureTask<Object>(runnable, null)

抛出：

NullPointerException - 如果 runnable 为 null。

isCancelled

public boolean isCancelled()

从接口 Future 复制的描述

如果在任务正常完成前将其取消，则返回 true。

指定者：

接口 Future<V> 中的 isCancelled

返回：

如果任务完成前将其取消，则返回 true

isDone

public boolean isDone()

从接口 Future 复制的描述

如果任务已完成，则返回 true。可能由于正常终止、异常或取消而完成，在所有这些情况中，此方法都将返回 true。

指定者：

接口 Future<V> 中的 isDone

返回：

如果任务已完成，则返回 true

cancel

public boolean cancel(boolean mayInterruptIfRunning)

从接口 Future 复制的描述

试图取消对此任务的执行。如果任务已完成、或已取消，或者由于某些其他原因而无法取消，则此尝试将失败。当调用 cancel 时，如果调用成功，而此任务尚未启动，则此任务将永不运行。如果任务已经启动，则 mayInterruptIfRunning 参数确定是否应该以试图停止任务的方式来中断执行此任务的线程。

此方法返回后，对 Future.isDone() 的后续调用将始终返回 true。如果此方法返回 true，则对 Future.isCancelled() 的后续调用将始终返回 true。

指定者：

接口 Future<V> 中的 cancel

参数：

mayInterruptIfRunning - 如果应该中断执行此任务的线程，则为 true；否则允许正在运行的任务运行完成

返回：

如果无法取消任务，则返回 false，这通常是由于它已经正常完成；否则返回 true

get

public V get()
      throws InterruptedException,
             ExecutionException

从接口 Future 复制的描述

如有必要，等待计算完成，然后获取其结果。

指定者：

接口 Future<V> 中的 get

返回：

计算的结果

抛出：

CancellationException - 如果计算被取消

InterruptedException - 如果当前的线程在等待时被中断

ExecutionException - 如果计算抛出异常

get

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

从接口 Future 复制的描述

如有必要，最多等待为使计算完成所给定的时间之后，获取其结果（如果结果可用）。

指定者：

接口 Future<V> 中的 get

参数：

timeout - 等待的最大时间

unit - timeout 参数的时间单位

返回：

计算的结果

抛出：

CancellationException - 如果计算被取消

InterruptedException - 如果当前的线程在等待时被中断

ExecutionException - 如果计算抛出异常

TimeoutException - 如果等待超时

done

protected void done()

当此任务转换到状态 isDone（不管是正常地还是通过取消）时，调用受保护的方法。默认实现不执行任何操作。子类可以重写此方法，以调用完成回调或执行簿记。注意，可以查询此方法的实现内的状态，从而确定是否已取消了此任务。

set

protected void set(V v)

除非已经设置了此 Future 或已将其取消，否则将其结果设置为给定的值。在计算成功完成时通过 run 方法内部调用此方法。

参数：

v - 值

setException

protected void setException(Throwable t)

除非已经设置了此 Future 或已将其取消，否则它将报告一个 ExecutionException，并将给定的 throwable 作为其原因。在计算失败时通过 run 方法内部调用此方法。

参数：

t - 失败的原因

run

public void run()

除非已将此 Future 取消，否则将其设置为其计算的结果。

指定者：

接口 Runnable 中的 run

指定者：

接口 RunnableFuture<V> 中的 run

另请参见：

Thread.run()

runAndReset

protected boolean runAndReset()

执行计算而不设置其结果，然后将此 Future 重置为初始状态，如果计算遇到异常或已取消，则该操作失败。本操作被设计用于那些本质上要执行多次的任务。

返回：

如果成功运行并重置，则返回 true。