线程池源码分析_01 FutureTask源码分析

Posted 2021-05-21 兴趣使然の草帽路飞

• 文章参考：硬核手撕Java线程池FutureTask源码

---

1、FutureTask简介

Future是我们在使用java实现异步时最常用到的一个类，我们可以向线程池提交一个Callable，并通过future对象获取执行结果。

• FutureTask的使用场景：FutureTask可用于异步获取执行结果或取消执行任务的场景。通过传入Runnable或者Callable的任务给FutureTask，直接调用其run方法或者放入线程池执行，之后可以在外部通过FutureTask的get方法异步获取执行结果，因此，FutureTask非常适合用于耗时的计算，主线程可以在完成自己的任务后，再去获取结果。另外，FutureTask还可以确保即使调用了多次run方法，它都只会执行一次Runnable或者Callable任务，或者通过cancel取消FutureTask的执行等。
• 关于FutureTask的基本使用可以参考文章：FutureTask的用法

---

2、源码分析

成员属性

// 表示当前task的状态
private volatile int state;
// 表示当前任务尚未执行
private static final int NEW          = 0;
// 表示当前任务正在结束，尚未完全结束，一种临界状态
private static final int COMPLETING   = 1;
// 表示当前任务正常结束
private static final int NORMAL       = 2;
// 表示当前任务执行过程中发生了异常。 内部封装的 callable.run() 向上抛出异常了
private static final int EXCEPTIONAL  = 3;
// 表示当前任务被取消
private static final int CANCELLED    = 4;
// 表示当前任务中断中..
private static final int INTERRUPTING = 5;
// 表示当前任务已中断
private static final int INTERRUPTED  = 6;

//submit(runnable/callable)：runnable 使用装饰者设计模式伪装成 Callable了。
private Callable<V> callable;

// 正常情况下：任务正常执行结束，outcome保存执行结果，作为callable返回值。
// 非正常情况：callable向上抛出异常，outcome保存异常信息
private Object outcome;

// 表示当前任务被线程执行期间，保存当前执行任务的线程对象引用。
private volatile Thread runner;

// 因为会有很多线程去get当前任务的结果，所以这里使用了一种数据结构 stack **头插、头取**的一个队列。
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();
    // callable就是程序员自己实现的业务类
    this.callable = callable;
    // 设置当前任务状态为NEW: 表示当前任务尚未执行
    this.state = NEW;
}

public FutureTask(Runnable runnable, V result) {
    // 使用装饰者模式将runnable转换为了callable接口，外部线程通过get获取
    // 当前任务执行结果时，结果可能为 null 也可能为传进来的值。
    this.callable = Executors.callable(runnable, result);
    // 设置当前任务状态为NEW: 表示当前任务尚未执行
    this.state = NEW;
}

成员方法

1. run方法

// submit(runnable/callable) -> newTaskFor(runnable) -> execute(task) -> pool
// 线程执行入口的方法
public void run() {
    // 条件一：state != NEW 条件成立，说明当前task已经被执行过了或者被cancel了，
    // 总之非NEW状态的任务，线程就不处理了，直接return;结束!
    // 条件二：!UNSAFE.compareAndSwapObject(this, runnerOffset,null, Thread.currentThread())
    // 条件成立：cas失败，当前任务被其它线程抢占了...
    if (state != NEW ||
            !UNSAFE.compareAndSwapObject(this, runnerOffset,
                    null, Thread.currentThread()))
        return;

    // 如果执行到这里，说明当前task一定是 NEW 状态，而且当前线程也抢占TASK成功！
    try {
        // callable 就是程序员自己封装逻辑的callable 或者装饰后的runnable
        Callable<V> c = callable;
        // 条件一：c != null 防止空指针异常
        // 条件二：state == NEW 防止外部线程 cancel掉当前任务。
        if (c != null && state == NEW) {
            // 结果的引用
            V result;
            // true 表示callable.run 代码块执行成功 未抛出异常
            // false 表示callable.run 代码块执行失败 抛出异常
            boolean ran;
            try {
                // 调用程序员自己实现的callable 或者 装饰后的runnable
                result = c.call();
                // c.call未抛出任何异常，ran会设置为true 表示任务执行成功
                ran = true;
            } catch (Throwable ex) {
                // 说明程序员自己写的逻辑块有bug了。
                result = null;
                // ran设置为false
                ran = false;
                // 将当前异常信息ex赋值给outcome
                setException(ex);
            }
            
            // run： 
            // true -> 代码块执行成功 未抛出异常
            // false -> 代码块执行失败 抛出异常
            if (ran)
                // 说明当前c.call正常执行结束了。
                // set方法: 以CAS的方式将result结果设置给outcome
                set(result);
        }
    } finally {
        // 将当前执行任务的线程置为null
        runner = null;
        // 当前任务的状态
        int s = state;
        // s >= INTERRUPTING：如果条件成立，说明当前任务处于中断中或者已中断状态...
        if (s >= INTERRUPTING)
            // 后面分析cancel()方法的时候再回头看这里就明白了！
            handlePossibleCancellationInterrupt(s);
    }
}

/**
 * 以CAS的方式设置结果v给outcome
 */
protected void set(V v) {
    // 使用CAS方式设置当前任务状态为完成中...
    // 有没有可能失败呢？ 外部线程等不及了，直接在set执行CAS之前将task取消了。（很小概率事件）
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        // 将v赋值给outcome
        outcome = v;
        // 将结果v赋值给outcome之后，马上会将当前任务状态修改为NORMAL（正常结束状态）。
        UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state

        // 当线程被挂起之后，如果任务线程执行完毕，就会唤醒等待线程。
        // 这一步就是在finishCompletion里面做的, 后面分析get() 的时候再说~
        // 移除并唤醒所有等待线程,执行done,置空callable
        finishCompletion();
    }
}

/**
 * 将当前异常信息t赋值给outcome
 */
protected void setException(Throwable t) {
    // 使用CAS方式设置当前任务状态为 完成中..
    // 有没有可能失败呢？ 外部线程等不及了，直接在set执行CAS之前 将 task取消了。  很小概率事件。
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        // 引用的是 callable 向上层抛出来的异常。
        outcome = t;
        // 将当前任务的状态 修改为 EXCEPTIONAL
        UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
        
        // 当线程被挂起之后，如果任务线程执行完毕，就会唤醒等待线程。
        // 这一步就是在finishCompletion里面做的, 后面分析get() 的时候再说~
        // 移除并唤醒所有等待线程,执行done,置空callable
        finishCompletion();
    }
}

2. get方法

// get获取当前任务执行结束后得到的结果：
// 注意：get不一定是只有一个线程去得到任务执行后的结果，多个线程等待获取当前任务执行完成后的结果这种场景也是有的！
public V get() throws InterruptedException, ExecutionException {
    // 获取当前任务状态
    int s = state;
    // s <= COMPLETING：
    // 条件成立：(当前任务可能正处于：未执行0、正在执行1、正在完成1 状态，总之是还没有结束)。 
    // 那么，此时调用get的外部线程会被阻塞在get方法上。
    if (s <= COMPLETING)
        // awaitDone执行完后会返回task当前状态，如果该方法执行期间，task被中断了，则会直接抛出中断异常:
        // awaitDone是futureTask实现阻塞的关键方法: 等待任务执行完毕,如果任务取消或者超时则停止!
        s = awaitDone(false, 0L);
    // report(s)去获取最终task执行结束得到的结果
    return report(s);
}

// 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 {
    // 0 不带超时
    final long deadline = timed ? System.nanoTime() + nanos : 0L;
    // 引用当前线程封装成 WaitNode对象（**头插、头取**的一个队列。）
    WaitNode q = null;
    // 表示当前线程 waitNode对象是否入队/压栈
    boolean queued = false;
    // 自旋
    for (;;) {
        // 条件成立：说明当前线程唤醒是被其它线程使用中断这种方式喊醒的。interrupted()
        // 返回true 后会将 Thread的中断标记重置回false.
        if (Thread.interrupted()) {
            // 当前线程节点出队
            removeWaiter(q);
            // 上抛，使get方法抛出中断异常。
            throw new InterruptedException();
        }

        // 假设当前线程是被其它线程使用unpark(thread) 唤醒的话，会正常自旋，走下面逻辑：

        // 获取当前任务最新状态
        int s = state;
        // 条件成立：说明当前任务已经有结果了.. （可能是完成、异常、中断等等）
        if (s > COMPLETING) {
            // 条件成立：说明已经为当前线程创建过WaitNode了，此时需要将 node.thread = null helpGC
            if (q != null)
                q.thread = null;
            // 直接返回当前状态.
            return s;
        }
        // 条件成立：说明当前任务接近完成状态、或者接近失败状态...
        // 这里让当前线程再释放cpu ，进行下一次抢占cpu：
        else if (s == COMPLETING) // cannot time out yet
            Thread.yield();
        // 条件成立：第一次自旋，当前线程还未创建 WaitNode 对象，此时为当前线程创建 WaitNode对象
        else if (q == null)
            q = new WaitNode();
        // 条件成立：第二次自旋，当前线程已经创建 WaitNode对象了，但是node对象还未入队
        else if (!queued){
            // 当前线程node节点 next 指向原队列的头节点 waiters 一直指向队列的头！
            q.next = waiters;
            // cas方式设置waiters引用指向当前线程node， 成功的话 queued == true 否则，可能其它线程先你一步入队了。
            queued = UNSAFE.compareAndSwapObject(this, waitersOffset, waiters, q);
        }
        // 第三次自旋，会到这里：表示是否设置了超时时间
        else if (timed) {
            nanos = deadline - System.nanoTime();
            if (nanos <= 0L) {
                // 已经超时的话,移除等待节点
                removeWaiter(q);
                return state;
            }
             // 未超时,将当前线程挂起指定时间
            LockSupport.parkNanos(this, nanos);
        }
        else
            // 当前get操作的线程就会被park了。线程状态会变为 WAITING状态，相当于休眠了..
            // 除非有其它线程将你唤醒 或者 将当前线程中断。
            // 如果当前线程被其他线程唤醒，醒来时，还是从这里向下继续执行(继续进入自旋for进行条件判断)
            // (在finishCompletion中会唤醒这个挂起的线程！)
            LockSupport.park(this);
    }
}

// report(s)去获取最终task执行结束得到的结果
@SuppressWarnings("unchecked")
private V report(int s) throws ExecutionException {
    // 正常情况下，outcome 保存的是callable运行结束的结果
    // 非正常情况下，保存的是 callable 抛出的异常。
    Object x = outcome;
    // 条件成立(正常情况)：当前任务状态正常结束
    if (s == NORMAL)
        // 直接返回callable运算结果
        return (V)x;

    // 条件成立(非正常情况)：当前任务是被取消或中断状态
    if (s >= CANCELLED)
        // 抛异常！
        throw new CancellationException();

    // 执行到这，说明callable接口实现中，是有bug的...
    throw new ExecutionException((Throwable)x);
}

// node出队方法：
private void removeWaiter(WaitNode node) {
    if (node != null) {
        node.thread = null;
        retry:
        for (;;) {          // restart on removeWaiter race
            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) // check for race
                        continue retry;
                }
                else if (!UNSAFE.compareAndSwapObject(this, waitersOffset,
                        q, s))
                    continue retry;
            }
            break;
        }
    }
}

// 当线程被挂起之后，如果任务线程执行完毕，就会唤醒等待线程。这一步就是在finishCompletion里面做的
/**
 * 移除并唤醒所有等待线程,执行done,置空callable
 */
private void finishCompletion() {
    // 遍历等待节点：
    // q指向waiters链表的头结点
    for (WaitNode q; (q = waiters) != null;) {
        // 使用cas设置 waiters为null 
        // 为了防止外部线程使用cancel取消当前任务，也会触发finishCompletion方法。（小概率事件）
        if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
            for (;;) {
                // 获取当前node节点封装的 thread
                Thread t = q.thread;
                // 条件成立：说明当前线程不为null
                if (t != null) {
                    q.thread = null;// help GC
                    // 唤醒当前节点对应的线程（在awaitDone方法最后一个else判断中park，在此处唤醒）
                    LockSupport.unpark(t);
                }
                // next: 当前节点的下一个节点
                WaitNode next = q.next;
                // next == null 说明是最后一个节点，则直接break即可
                if (next == null)
                    break;
                q.next = null; // unlink to help gc
                q = next;
            }
            break;
        }
    }
    // 模板方法,可以被覆盖
    done();
    // 将callable 设置为null helpGC
    callable = null;
}

3. cancel方法

// 将当前线程的任务取消(中断)掉：
public boolean cancel(boolean mayInterruptIfRunning) {
    // 条件一：state == NEW 成立，表示当前任务处于运行中或者处于线程池任务队列中..
    // 条件二：UNSAFE.compareAndSwapInt(this, stateOffset, NEW, mayInterruptIfRunning ? INTERRUPTING : CANCELLED))
    // 条件成立：说明CAS修改状态成功，可以去执行下面逻辑了，否则返回false，表示cancel失败。
    if (!(state == NEW &&
            UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
                    mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
        return false;

    try {
        if (mayInterruptIfRunning) {
            try {
                // 执行当前FutureTask 的线程，有可能现在是null，是null 的情况是： 当前任务在 队列中，还没有线程获取到它呢。。
                Thread t = runner;
                // 条件成立：说明当前线程 runner ，正在执行task.
                if (t != null)
                    // 给runner线程一个中断信号.. 如果你的程序是响应中断 会走中断逻辑..假设你程序不是响应中断的..啥也不会发生。
                    t.interrupt();
            } finally { // final state
                // 设置任务状态为 中断完成。
                UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
            }
        }
    } finally {
        // 唤醒所有get()阻塞的线程。
        finishCompletion();
    }
    return true;
}

---

• 如果文章有帮助，点赞支持一下哈~