FutureTask介绍及使用

太过爱你忘了你带给我的痛 2022-05-28 07:07 212阅读 0赞

# Future #

## Future是一个接口，它定义了5个方法： ##

*  boolean cancel(boolean mayInterruptIfRunning);
 *  boolean isCancelled();
 *  boolean isDone();
 *  V get() throws InterruptedException, ExecutionException;  
    **简单说明一下接口定义**
 *  boolean cancel(boolean mayInterruptInRunning) 取消一个任务，并返回取消结果。参数表示是否中断线程。
 *  boolean isCancelled() 判断任务是否被取消
 *  Boolean isDone() 　　 判断当前任务是否执行完毕，包括正常执行完毕、执行异常或者任务取消。
 *  V get() 获取任务执行结果，任务结束之前会阻塞。
 *  V get(long timeout, TimeUnit unit) 在指定时间内尝试获取执行结果。若超时则抛出超时异常

## 简单demo： ##

import java.util.concurrent.Callable;
 import java.util.concurrent.FutureTask;
 
 public class Demo { 
 
 
 
 /** * Callalbe和Runnable的区别 * * Runnable run方法是被线程调用的，在run方法是异步执行的 * * Callable的call方法，不是异步执行的，是由Future的run方法调用的 * * * * @param args * @throws Exception */
 public static void main(String[] args) throws Exception {
 
 Callable<Integer> call = new Callable<Integer>() {
 
 @Override
 public Integer call() throws Exception {
 System.out.println("正在计算结果...");
 Thread.sleep(3000);
 return 1;
 }
 };
 
 FutureTask<Integer> task = new FutureTask<>(call);
 
 Thread thread = new Thread(task);
 thread.start();
 
 // do something
 System.out.println(" 干点别的...");
 
 Integer result = task.get();
 
 System.out.println("拿到的结果为：" + result);
 
 }
 
 }

## FutureTask实现原理 ##

下面我们介绍一下FutureTask内部的一些实现机制。下文从以下几点叙述：

1、类继承结构  
2、核心成员变量  
3、内部状态转换  
4、核心方法解析

## 1 类继承结构 ##

**首先我们看一下FutureTask的继承结构：**  
![这里写图片描述][70]  
FutureTask实现了RunnableFuture接口，而RunnableFuture继承了Runnable和Future，也就是说FutureTask既是Runnable，也是Future。

## 2 核心成员变量 ##

**FutureTask内部定义了以下变量，以及它们的含义如下**

*  volatile int state:表示对象状态，volatile关键字保证了内存可见性。futureTask中定义了7种状态，代表了7种不同的执行状态

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; //任务线程已中断

*  Callable callable:被提交的任务
 *  Object outcome:任务执行结果或者任务异常
 *  volatile Thread runner:执行任务的线程
 *  volatile WaitNode waiters：等待节点，关联等待线程
 *  long stateOffset:state字段的内存偏移量
 *  long runnerOffset：runner字段的内存偏移量
 *  long waitersOffset：waiters字段的内存偏移量  
    后三个字段是配合Unsafe类做CAS操作使用的。

## 3 内部状态转换 ##

FutureTask中使用state表示任务状态，state值变更的由CAS操作保证原子性。

FutureTask对象初始化时，在构造器中把state置为为NEW，之后状态的变更依据具体执行情况来定。

例如任务执行正常结束前，state会被设置成COMPLETING，代表任务即将完成，接下来很快就会被设置为NARMAL或者EXCEPTIONAL，这取决于调用Runnable中的call()方法是否抛出了异常。有异常则后者，反之前者。

任务提交后、任务结束前取消任务，那么有可能变为CANCELLED或者INTERRUPTED。在调用cancel方法时，如果传入false表示不中断线程，state会被置为CANCELLED，反之state先被变为INTERRUPTING，后变为INTERRUPTED。

**总结下，FutureTask的状态流转过程，可以出现以下四种情况：**  
1. 任务正常执行并返回。 NEW -> COMPLETING -> NORMAL  
2. 执行中出现异常。NEW -> COMPLETING -> EXCEPTIONAL  
3. 任务执行过程中被取消，并且不响应中断。NEW -> CANCELLED  
4.任务执行过程中被取消，并且响应中断。 NEW -> INTERRUPTING -> INTERRUPTED

## 4 核心方法解析 ##

接下来我们一起扒一扒FutureTask的源码。我们先看一下任务线程是怎么执行的。当任务被提交到线程池后，会执行futureTask的run()方法。  
**1 public void run()**

public void run() { // 校验任务状态
 if (state != NEW || !UNSAFE.compareAndSwapObject(this, runnerOffset, null, Thread.currentThread()))
 return;
 try {
 Callable<V> c = callable; 　　　　　　　// double check
 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
 runner = null;
 int s = state;
 if (s >= INTERRUPTING)
 handlePossibleCancellationInterrupt(s);
 }
 }

翻译一下，这个方法经历了以下几步

1.  校验当前任务状态是否为NEW以及runner是否已赋值。这一步是防止任务被取消。
2.  double-check任务状态state
3.  执行业务逻辑，也就是c.call()方法被执行
4.  如果业务逻辑异常，则调用setException方法将异常对象赋给outcome，并且更新state值
5.  如果业务正常，则调用set方法将执行结果赋给outcome，并且更新state值  
    我们继续往下看，setException(Throwable t)和set(V v) 具体是怎么做的

protected void set(V v) {
 // state状态 NEW->COMPLETING
 if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
 outcome = v;
 // COMPLETING -> NORMAL 到达稳定状态
 UNSAFE.putOrderedInt(this, stateOffset, NORMAL);
 // 一些结束工作
 finishCompletion();
 }
 }

protected void setException(Throwable t) {
        // state状态 NEW->COMPLETING
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
            outcome = t;
            // COMPLETING -> EXCEPTIONAL 到达稳定状态
            UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL);
            // 一些结束工作
            finishCompletion();
        }
    }

code中的注释已经写的很清楚，故不翻译了。状态变更的原子性由unsafe对象提供的CAS操作保证。FutureTask的outcome变量存储执行结果或者异常对象，会由主线程返回。  
**2 get()和get(long timeout, TimeUnit unit)**

任务由线程池提供的线程执行，那么这时候主线程则会阻塞，直到任务线程唤醒它们。我们通过get(long timeout, TimeUnit unit)方法看看是怎么做的

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

get的源码很简洁，首先校验参数，然后根据state状态判断是否超时，如果超时则异常，不超时则调用report(s)去获取最终结果。

当 s<= COMPLETING时，表明任务仍然在执行且没有被取消。如果它为true，那么走到awaitDone方法。

awaitDone是futureTask实现阻塞的关键方法，我们重点关注一下它的实现原理。

/** 1. 等待任务执行完毕,如果任务取消或者超时则停止 2. @param timed 为true表示设置超时时间 3. @param nanos 超时时间 4. @return 任务完成时的状态 5. @throws InterruptedException */
 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)
 // 可能任务线程被阻塞了,主线程让出CPU
 Thread.yield();
 else if (q == null)
 // 等待线程节点为空,则初始化新节点并关联当前线程
 q = new WaitNode();
 else if (!queued)
 // 等待线程入队列,成功则queued=true
 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
 // timed=false时会走到这里,挂起当前线程
 LockSupport.park(this);
 }
 }

**注释里也很清楚的写明了每一步的作用，我们以设置超时时间为例，总结一下过程**

1、计算deadline，也就是到某个时间点后如果还没有返回结果，那么就超时了。  
2、进入自旋，也就是死循环。  
3、首先判断是否响应线程中断。对于线程中断的响应往往会放在线程进入阻塞之前，这里也印证了这一点。  
4、判断state值，如果>COMPLETING表明任务已经取消或者已经执行完毕，就可以直接返回了。  
5、如果任务还在执行，则为当前线程初始化一个等待节点WaitNode，入等待队列。这里和AQS的等待队列类似，只不过6、Node只关联线程，而没有状态。AQS里面的等待节点是有状态的。  
7、计算nanos，判断是否已经超时。如果已经超时，则移除所有等待节点，直接返回state。超时的话，state的值仍然还是COMPLETING。  
8、如果还未超时，就通过LockSupprot类提供的方法在指定时间内挂起当前线程，等待任务线程唤醒或者超时唤醒。

当线程被挂起之后，如果任务线程执行完毕，就会唤醒等待线程哦。这一步就是在finishCompletion里面做的，前面已经提到这个方法。我们再看看这个方法具体做了哪些事吧~

/** * 移除并唤醒所有等待线程,执行done,置空callable * nulls out callable. */
    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;
                    // unlink to help gc
                    q.next = null;
                    q = next;
                }
                break;
            }
        }
        //模板方法,可以被覆盖
        done();
        //清空callable
        callable = null;
    }

由代码和注释可以看出来，这个方法的作用主要在于唤醒等待线程。由前文可知，当任务正常结束或者异常时，都会调用finishCompletion去唤醒等待线程。这个时候，等待线程就可以醒来，开开心心的获得结果啦。

最后我们看一下任务取消　　  
3 public boolean cancel(boolean mayInterruptIfRunning)  
注意，取消操作不一定会起作用，这里我们先贴个demo

public class FutureDemo {
 public static void main(String[] args) {
 ThreadPoolExecutor executorService = (ThreadPoolExecutor) Executors.newFixedThreadPool(1);
 // 预创建线程
 executorService.prestartCoreThread();
 
 Future future = executorService.submit(new Callable<Object>() {
 @Override
 public Object call() {
 System.out.println("start to run callable");
 Long start = System.currentTimeMillis();
 while (true) {
 Long current = System.currentTimeMillis();
 if ((current - start) > 1000) {
 System.out.println("当前任务执行已经超过1s");
 return 1;
 }
 }
 }
 });
 
 System.out.println(future.cancel(false));
 
 try {
 Thread.currentThread().sleep(3000);
 executorService.shutdown();
 } catch (Exception e) {
 //NO OP
 }
 }
 }

我们多次测试后发现，出现了2种打印结果，如图  
**结果一**  
![这里写图片描述][70 1]  
**结果二**  
![这里写图片描述][70 2]  
第一种是任务压根没取消，第二种则是任务压根没提交成功。

方法签名注释告诉我们，取消操作是可能会失败的，如果当前任务已经结束或者已经取消，则当前取消操作会失败。如果任务尚未开始，那么任务不会被执行。这就解释了出现上图结果2的情况。我们还是从源码去分析cancel()究竟做了哪些事。

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

执行逻辑如下  
1、state不为NEW时，任务即将进入终态，直接返回false表明取消操作失败。  
2、state状态为NEW，任务可能已经开始执行，也可能还未开始。  
3、mayInterruptIfRunning表明是否中断线程。若是，则尝试将state设置为INTERRUPTING，并且中断线程，之后将state设置为终态INTERRUPTED。  
4、如果mayInterruptIfRunning=false，则不中断线程，把state设置为CANCELLED  
5、移除等待线程并唤醒。  
6、返回true

可见，cancel()方法改变了futureTask的状态位，如果传入的是false并且业务逻辑已经开始执行，当前任务是不会被终止的，而是会继续执行，直到异常或者执行完毕。如果传入的是true，会调用当前线程的interrupt()方法，把中断标志位设为true。
    
    事实上，除非线程自己停止自己的任务，或者退出JVM，是没有其他方法完全终止一个线程的任务的。mayInterruptIfRunning=true，通过希望当前线程可以响应中断的方式来结束任务。当任务被取消后，会被封装为CancellationException抛出。

出处：[http://www.cnblogs.com/maypattis/][http_www.cnblogs.com_maypattis]

[70]: /images/20220528/9c2f7d59c472411781f28a522cfae6fa.png
[70 1]: /images/20220528/b96ccee99ba04a149b17955bf63411c2.png
[70 2]: /images/20220528/29b54e88a5224c42b721bddb2109c015.png
[http_www.cnblogs.com_maypattis]: http://www.cnblogs.com/maypattis/