Future Java

shysh95

发布于 2021-06-10 11:17:12

3850

发布于 2021-06-10 11:17:12

文章被收录于专栏：shysh95

摘要

什么是Future
为什么需要Future
Java中的Future模式
详解FutureTask

1. 什么是Future

Future是多线程开发中常见的一种设计模式。Future模式可以返回线程执行结果的契约，通过此契约程序可以选择在合适的时机取回执行的结果，如果取回结果时线程还没有执行完成，将会阻塞调用线程等待执行结果返回。

2. 为什么需要Future

在有些场景下，我们想使用另一个线程去执行复杂耗时的操作，此时又不想让主线程等待白白浪费CPU，此时可以让主线程先去做别的事，然后在合适的时机去通过Future契约取回线程执行的结果。

3. Java中的Future模式

Java中的Future模式主要由以上接口和类组成。

3.1 Callable & Runnable

这是我们普通的线程任务，其中Callable是带返回值（真实数据），Runnable是不带返回值的，因此在我们使用Runnable和Future时，必须传入一个Result对象，通过Future在获取结果时就是获取的该Result，核心代码如下：

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

public static <T> Callable<T> callable(Runnable task, T result) {
    if (task == null)
        throw new NullPointerException();
    return new RunnableAdapter<T>(task, result);
}

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

Callable目前只能搭配线程池或者Future来使用，不能直接和new Thread()搭配使用，Runnable可以搭配线程池和new Thread()使用，在配合Future使用时本质上是对其进行了适配，也就是上述代码中的RunnableAdapter。

3.2 Future

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

Future是线程的契约，通过其get()方法我们可以获取线程执行的结果，当然Future也提供了其他三个方法，分别是：

cancel：取消任务
isCancelled：任务是否已经取消
isDone：任务是否完成

3.3 RunnableFuture

public interface RunnableFuture<V> extends Runnable, Future<V> {

    void run();
}

RunnableFuture接口继承自Runnable和Future，表明RunnableFuture可以被线程执行并且可以通过契约获取到线程的执行结果。

4. FutureTask

4.1 属性

// 执行任务
private Callable<V> callable;
// 任务的实际执行结果
private Object outcome; 
// 执行任务的线程
private volatile Thread runner;
// 等待结果的线程栈
private volatile WaitNode waiters;

4.2 状态

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;

FutureTask除了4.1中的属性外，还有一个重要的属性就是state，FutureTask中的状态大约有7种：

NEW：任务的初始状态
COMPLETING：正在设置任务结果
NORMAL：任务执行完毕
EXCEPTIONAL：任务发行异常
CANCELLED：任务被取消
INTERRUPTING：正在中断任务
INTERRUPTED：任务被中断

4.3 run()方法

任务执行的时候实际就是执行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 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;
        if (s >= INTERRUPTING)
            handlePossibleCancellationInterrupt(s);
    }
}

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

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
}

run方法的大致流程如下：

校验任务的状态是否是NEW和当前是否无执行线程，如果校验通过，则获取任务执行
调用任务的call方法
如果执行异常，设置结果，状态修改为EXCEPTIONAL，并将任务结果设置为异常
如果正常执行，调用set(V v)设置结果，状态修改为NORMAL，结果设置为执行结果，并且唤醒等待结果的线程
最后在finally块中，我们将runner属性置为null，并且检查有没有遗漏的中断，如果发现s >= INTERRUPTING, 说明执行任务的线程有可能被中断了，因为s >= INTERRUPTING 只有两种可能，state状态为INTERRUPTING和INTERRUPTED。

4.3 get()方法

当我们需要去获取FutureTask的结果时，我们需要调用get方法获取结果。


public V get() throws InterruptedException, ExecutionException {
    int s = state;
    if (s <= COMPLETING)
        s = awaitDone(false, 0L);
    return report(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);
}

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) // cannot time out yet
            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);
    }
}

获取结果的大致步骤如下：