FutureTask详解 –源码分析– 基于JDK 1.8

曾经终败给现在 2021-11-01 07:58 361阅读 0赞

在之前我们已经了解Future的使用：

[Callable和Future详解 – 基于JDK 1.8][Callable_Future_ _ _JDK 1.8]

现在来看一下Future的唯一实现类FutureTask的内部原理分析吧；

**目录**

1、类的结构

（1）、继承关系：

2、类中属性与方法

（1）、各种状态值

（2）、其它属性

（3）、构造方法

（4）、大致实现思路

（5）、Runnable接口实现

（6）、阻塞获取执行结果get方法：

（7）、取消线程任务执行cancel()方法

（8）至此，几个方法比较简单，不详细分析：

--------------------

# 1、类的结构 #

## （1）、继承关系： ##

IDEA查看结构如下，实现了Future和Runnable接口，所以FutureTask既能当做一个Runnable线程执行，也可以作为Future获取Callable的执行结果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xlanVzdGRvaXQ_size_16_color_FFFFFF_t_70][]

# 2、类中属性与方法 #

## （1）、各种状态值 ##

各种状态值，FutureTask的核心操作，就是围绕的state的这几个属性值的更改来执行业务逻辑；

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xlanVzdGRvaXQ_size_16_color_FFFFFF_t_70 1][]

//任务state状态值
    private volatile int state;
    //0初态：线程新建的初始状态值
    private static final int NEW          = 0;
    //1中间态：瞬时状态，任务已经开始执行
    private static final int COMPLETING   = 1;
    //2终态：任务正常完成，并且任务执行结果已经保存到outcome属性中
    private static final int NORMAL       = 2;
    //3终态：任务执行发生异常，异常信息保存到outcome属性中
    private static final int EXCEPTIONAL  = 3;
    //4终态：任务还没开始执行或者已经开始执行，但还未执行完时，
    //调用线程执行了cancel(false)方法取消任务且不中断任务执行线程的状态
    private static final int CANCELLED    = 4;
    //5中间态：任务已经开始执行但还未执行完时，
    //调用线程执行了cancel(true)方法取消任务，
    //并且要中断任务执行线程，中断中，一个瞬时状态值；
    private static final int INTERRUPTING = 5;
    //6终态：任务已经开始执行但还未执行完时，调用线程执行了cancel(true)方
    //法取消任务并且要中断任务执行线程，最后的状态；
    private static final int INTERRUPTED  = 6;

## （2）、其它属性 ##

/**接收任务的callable，运行后则失效，任务本身*/
    private Callable<V> callable;
    /** 线程执行完成，正常异常的结果保存信息对象，
    非volatile 修饰，读写受state状态值保护*/
    private Object outcome; // non-volatile, protected by state reads/writes
    /**正在执行callable任务的线程，在run期间是cas操作的，任务执行者*/
    private volatile Thread runner;
    /** 使用Treiber堆栈来保存等待线程，栈的头节点对象*/
    private volatile WaitNode waiters;
    
    /** CAS操作主要针对3个属性，包括state、runner和waiters，
    说明这3个属性基本是会被多个线程同时访问的。state属性表示
    任务状态，waiters属性表示栈头节点指针， runner属性
    表示FutureTask中的runnable或callable线程
    （为什么需要一个属性来记录执行任务的线程呢？
    为了在中断或取消当前的任务时，用来操作的属性）；
    上述3个属性的偏移量分别如下，记录了多线程需要CAS操作的偏移量，
    只要涉及到cas操作的属性，都有偏移量值*/
    private static final long stateOffset;
    private static final long runnerOffset;
    private static final long waitersOffset;

## （3）、构造方法 ##

构造方法，接收callable线程，如果传入的是runnable，采用适配器模式，也包装成callable，初始化线程状态为state = 0 (NEW)：

public FutureTask(Callable<V> callable) {
        if (callable == null)
            throw new NullPointerException();
        this.callable = callable;
        this.state = NEW;       // ensure visibility of callable
    }
    
    public FutureTask(Runnable runnable, V result) {
        this.callable = Executors.callable(runnable, result);
        this.state = NEW;       // ensure visibility of callable
    }
    // Executors.callable(runnable, result)  方法
    public static <T> Callable<T> callable(Runnable task, T result) {
        if (task == null)
            throw new NullPointerException();
        return new RunnableAdapter<T>(task, result);
    }
    
    /**一种适配器模式，Runnable转Callable线程对象；
     * A callable that runs given task and returns given 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;
        }
    }

## （4）、大致实现思路 ##

在FutureTask中会涉及到两类线程，一类是执行任务的线程，并且只有一个线程，FutureTask的run方法就负责执行该线程；

另一类则是获取任务执行结果的线程，它可以有多个，这些线程可以并发执行，每一个线程都是独立的，都可以调用get方法来获取任务的执行结果。

如果任务通过之前的run方法执行完成，则返回结果；如果任务还没有执行完成，不能及时获取线程执行结果，在并发编程中使用队列通常是将当前线程包装成某种类型的数据结构（这里创建WaitNode对象节点，把当前线程包装到该对象节点中），扔到等待队列中（这里是后进先出Treiber栈集合）挂起，直到其它线程任务执行完毕后，唤醒挂起节点线程，这一点有点类似于我们之前分析过的AQS队列；下面是队列栈中的节点对象结构：

static final class WaitNode {
        volatile Thread thread;
        volatile WaitNode next;
        WaitNode() {
     thread = Thread.currentThread();
     }
    }

FutureTask中的这个单向链表是当做栈来使用的，确切来说是当做Treiber栈来使用的，不了解Treiber栈是个啥的可以简单的把它当做是一个线程安全的栈，它使用CAS来完成入栈出栈操作(一种后进先出LIFO的结构)。为啥要使用一个线程安全的栈呢，因为同一时刻可能有多个线程都在获取任务的执行结果，如果任务还在执行过程中，则这些线程就要被包装成WaitNode扔到Treiber栈的栈顶，即完成入栈操作，这样就有可能出现多个线程同时入栈的情况，因此需要使用CAS操作保证入栈的线程安全，对于出栈的情况也是同理。

由于FutureTask中的队列本质上是一个Treiber栈，那么使用这个队列就只需要一个指向栈顶节点的指针就行了，在FutureTask中，就是waiters属性，事实上，它就是整个栈链表的头节点：

/** Treiber stack of waiting threads */
    private volatile WaitNode waiters;

## （5）、Runnable接口实现 ##

Runnable接口实现，执行任务run方法，在FutureTask的构造方法中，我们发现是支持传入Runnable对象的线程，Runnable虽然被适配器模式转换成了Callable，但是既然实现Runnable接口, 就得实现run方法, 看一下FutureTask中的run实现逻辑：

public void run() {
    //判断 state 是否是new,如果不是new则返回，防止并发重复执行；
    // compareAndSwapObject 方法找到 runnerOffset偏移量处
    //（即runner当前线程）设置为Thread.currentThread()；
    //上面两个条件都满足了，才不执行return；
        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 {
                // 调用callable的call方法执行计算；
                //state只要不是NEW状态，就说明任务已执行完成；
                //call方法执行完成以后，后面才设置state中间态、终态；
                    result = c.call();
                    ran = true;
                } catch (Throwable ex) {
                    result = null;
                    ran = false;
                //执行异常，异常对象赋值到outcome属性
                    setException(ex);
                }
                if (ran)
                //执行成功，结果值赋值到outcome属性
                    set(result);
            }
        } finally {
            // 最后线程执行完，将线程runner置空
            runner = null;
            //确保不会漏掉判断中断的情况，
    	    //如果线程还是在中断中，等待到中断后，INTERRUPTED  
            int s = state;
            if (s >= INTERRUPTING)
                handlePossibleCancellationInterrupt(s);
        }
    }
    
            //调用callable的call方法异常结束时，调用此方法,状态做CAS更改：
            //NEW -> COMPLETING -> EXCEPTIONAL
            protected void setException(Throwable t) {
            //先设置为中间态COMPLETING
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
            //异常信息赋值到outcome属性
            outcome = t;
            //异常后设置终态EXCEPTIONAL
            UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); 
            finishCompletion();
        }
    }
    
    private void finishCompletion() {
        // assert state > COMPLETING;
        //最外层for循环，在栈头节点waiters不为空的时候一直自旋，
        //直到把waiters设置null成功；
        for (WaitNode q; (q = waiters) != null;) {
            if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
                //把栈头节点waiters设空以后，开始遍历栈节点，
                //遍历并置空栈中所有等待的线程节点，然后唤醒他们
                for (;;) {
                    Thread t = q.thread;
                    if (t != null) {
    	                //置空节点
                        q.thread = null;
                        //唤醒栈节点线程，由于是for循环，后面依次唤醒
                        LockSupport.unpark(t);
                    }
                    WaitNode next = q.next;
    		        //如果到栈栈尾了，则结束循环；
                    if (next == null)
                        break;
                    q.next = null; // unlink to help gc
                    q = next;
                }
                break;
            }
        }
        //done方法此处实现为空，为了提供给子类实现在任务执行完之前要
        //做的一些操作
        done();
        callable = null;        // to reduce footprint
    }

## （6）、阻塞获取执行结果get方法： ##

public V get() throws InterruptedException, ExecutionException {
        int s = state;
        //如果线程是NEW或者COMPLETING状态
        if (s <= COMPLETING)
            s = awaitDone(false, 0L);
        return report(s);
    }
    
    private int awaitDone(boolean timed, long nanos)
        throws InterruptedException {
        // timed是否等待超时标志为false，说明没有设超时等待时间，
        //将一直阻塞等待线程结果；
        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;
            }
            //如果任务正在执行中，那么线程让步，让出cpu资源继续等待
            else if (s == COMPLETING) // cannot time out yet
                Thread.yield();
            //q为空，说明还未进入等待栈，新建节点准备入栈
            else if (q == null)
                q = new WaitNode();
            //如果没有入栈，下面开始如栈操作
            else if (!queued)
            //先把新节点q的next指向栈头节点waiters，
            //就是说新入栈的节点为新的头节点，
            //即找到头节点指针waitersOffset偏移量，
            //把新节点q指向原头节点waiters指针处，
            //q称为栈的新头节点
                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);
        }
    }

通过上面的awaitDone()方法，我们可以看到，在for (;;)死循环后半部，都是else if的逻辑，也就是说，想要获取执行结果：

第一次肯定会先执行创建等待节点的方法q = **new** WaitNode()，后续都是else不执行，所以for循环第一次结束；

第二次for循环才有机会执行cas线程节点插入栈的方法，

queued = ***UNSAFE***.compareAndSwapObject(**this**, ***waitersOffset***, q.**next** = **waiters**, q)，如果插入成功，queued为true，否则直到插入栈成功；

第三次for循环，如果线程还没执行完，那么就调用LockSupport.*parkNanos**或者*LockSupport.*park*方法把当前想要获取结果的线程挂起阻塞，等待其它线程唤醒；

//执行完上述awaitDone()入栈方法后，返回一个状态，根据状态处理结果
    private V report(int s) throws ExecutionException {
        Object x = outcome;
        //正常情况的返回
        if (s == NORMAL)
            return (V)x;
        if (s >= CANCELLED)
        //线程被取消抛CancellationException 
            throw new CancellationException();
        //其它非正常情况抛ExecutionException
        throw new ExecutionException((Throwable)x);
    }

## （7）、取消线程任务执行cancel()方法 ##

public boolean cancel(boolean mayInterruptIfRunning) {
        //如果线程是初态NEW&&
        //cancel方法传入参数true，从stateOffset偏移量处，把状态从NEW初
        //态cas置成INTERRUPTING，如果cancel方法传参数为false，把状态
        //从NEW初态cas置成CANCELLED，并且方法执行返回false;
        if (!(state == NEW &&
              UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
                  mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
            return false;
        try {    // in case call to interrupt throws exception
            //cancel方法传入参数true
            if (mayInterruptIfRunning) {
                try {
                    Thread t = runner;
                    //中断当前线程
                    if (t != null)
                        t.interrupt();
                } finally { // final state
                //这里的cas执行，不考虑预期值（原始值），
                //与上面compareAndSwapInt方法不同的是，
                //少了一个预期值执行参数；
                //直接把状态，找到stateOffset内存偏移量处，
                //cas置为INTERRUPTED；
                    UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
                }
            }
        } finally {
            //执行后善方法，前面已经分析过
            finishCompletion();
        }
        return true;
    }

## （8）至此，几个方法比较简单，不详细分析：     ##

//判断任务是否取消
    public boolean isCancelled() {
        return state >= CANCELLED;
    }
    //判断任务是否执行完成
    public boolean isDone() {
        return state != NEW;
    }

[Callable_Future_ _ _JDK 1.8]: https://blog.csdn.net/lejustdoit/article/details/98876666
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xlanVzdGRvaXQ_size_16_color_FFFFFF_t_70]: /images/20211101/dafa4b9d94cd4163b9621782ff3f9b25.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xlanVzdGRvaXQ_size_16_color_FFFFFF_t_70 1]: /images/20211101/c92ed854822d463c8437517e815bd527.png