线程池源码分析

线程池源码分析

文章参考：硬核手撕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的用法、FutureTask的用法及两种常用的使用场景 2、源码分析

成员属性

// 表示当前任务（task）的状态
private volatile int state;
// NEW 表示当前任务尚未执行
private static final int NEW          = 0;
// COMPLETING：表示当前任务正在结束，但是尚未完全结束，一种临界状态
private static final int COMPLETING   = 1;
// NORMAL：当前任务正常结束，没有发生任何异常
private static final int NORMAL       = 2;
// EXCEPTIONAL：当前任务执行过程中发生了异常，内部封装的 callable 对象的 run() 再执行过程中向上抛出了异常
private static final int EXCEPTIonAL  = 3;
//cancelled表示当前任务被取消了(调用cancel方法)
// CANCELLED：表示当前任务被取消（调用 cancel 方法）
private static final int CANCELLED    = 4;
// INTERRUPTING：表示当前任务正在中断状态中
private static final int INTERRUPTING = 5;
// INTERRUPTED：表示当前任务已经中断（注意，中断在程序中只是一个标志位，通知）
private static final int INTERRUPTED  = 6;

// submit(Runable / Callable) Runnable 使用适配者模式伪装成 Callable
private Callable callable;

// 正常情况下：任务正常执行结束，outcome 保存执行结果，callable 的返回值
// 非正常情况下：Callable 向上抛出异常，outcome 保存异常
private Object outcome; // non-volatile, protected by state reads/writes

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

// 因为会有很多线程去get当前任务的结果，所以 这里使用了一种数据结构 stack 头插 头取的一个队列
private volatile WaitNode waiters;

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

构造方法

// 这里说明一下线程池submit(runnbale/callable)的过程中，runnable通过适配器转化为callable对象的过程
// 1.线程池提交无返回结果的runnable对象
public Future submit(Runnable task) {
    if (task == null) throw new NullPointerException();
    // 这里，如果没有传入返回结果的话，默认为null
    RunnableFuture ftask = newTaskFor(task, null);
    execute(ftask);
    return ftask;
}

// 2.线程池提交有返回结果的runnable对象
public  Future submit(Runnable task, T result) {
    if (task == null) throw new NullPointerException();
    // 通过newTaskFor方法创建FutureTask对象
    RunnableFuture ftask = newTaskFor(task, result);
    execute(ftask);
    return ftask;
}

// 通过newTaskFor方法创建FutureTask对象
protected  RunnableFuture newTaskFor(Runnable runnable, T value) {
    return new FutureTask(runnable, value);
}

public FutureTask(Callable callable) {
    if (callable == null)
        throw new NullPointerException();
    // callable 就是调用者自己实现的业务类
    this.callable = callable;
    // 设置当前任务状态为NEW 状态
    this.state = NEW;       // ensure visibility of callable
}

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

// 这里，Executors.callable(runnable, result)方法就是将runnable对象转化为callable对象的方法了
public static  Callable callable(Runnable task, T result) {
    if (task == null)
        throw new NullPointerException();
    // 这里可以看出，返回值是一个 RunnableAdapter 对象
    return new RunnableAdapter(task, result);
}

// RunnableAdapter对象的构成，实现了Callable接口，这就是适配器模式将runable转化为callable的过程
static final class RunnableAdapter implements Callable {
    final Runnable task;
    final T result;
    RunnableAdapter(Runnable task, T result) {
        this.task = task;
        this.result = result;
    }
    public T call() {
        task.run();
        return result;
    }
}

成员方法

2.1、run() 方法

// 前置流程：submit(runbale/callable) => RunnableFuture newTaskFor(runnable) => execute(futureTask) =>pool
// 线程任务执行的入口：
public void run() {
    // 条件一：state != NEW 条件成立：说明当前task已经被执行过了或者被 cancel了，总之非NEW状态的任务，线程就不处理了
    // 条件二：!UNSAFE.compareAndSwapObject(this, runnerOffset,null, Thread.currentThread())
    //  条件成立：CAS失败，当前任务被其他线程抢占了
    //  条件失败：CAS成功，当前执行任务的线程是当前线程，不是其他线程
    if (state != NEW ||
        !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                     null, Thread.currentThread()))
        return;
    // 前提条件，当前task一定是NEW 状态，而且当前线程也抢占TASK成功
    try {
        // Callable 就是程序员自己封装逻辑的callable或者runable被适配器伪装的callable
        Callable c = callable;
        // 条件一：c != null 防止空指针
        // 条件二：state == NEW 防止外部线程cancel当前任务
        if (c != null && state == NEW) {
            // 保留结果的引用，目前是null
            V result;
            // true：表示callable.run代码执行成功，未抛出异常
            // false：表示callable.run代码执行失败，抛出异常
            boolean ran;
            try {
                // 调用程序员自己实现的callable 或者runnable经过适配器后的callable对象
                result = c.call();
                // c.call 正常执行，未抛出任务异常，ran会被设置为true
                ran = true;
            } catch (Throwable ex) {
                // 说明程序员自己写的逻辑块有bug了
                result = null;
                ran = false;
                // 回头再说
                setException(ex);
            }
            //
            if (ran)
                // 说明当前c.call正常执行结束了
                // set就是设置执行结果的
                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)
            // 回头再说，讲了cancel就明白了
            handlePossibleCancellationInterrupt(s);
    }
}

// 以 CAS 的方式设置结果 v 给 outcome
protected void set(V v) {
    // 使用CAS 的方式设置当前任务状态为正在完成中
    // 有没有可能失败呢？
    // 只有一种情况：外部线程等不及了，直接爱set执行CAS之前，将当前任务cancel了，小概率事件
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        // 将结果赋值给outcome之后，马上会将当前任务状态修改为 NORMAL 正常结束状态
        outcome = v;
        UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
        // 猜一猜，该方法会做什么工作？
        // 最起码要把 get() 方法在此阻塞的线程唤醒
        finishCompletion();
    }
}

// 执行程序员定义的callable逻辑的时候报错执行的方法
protected void setException(Throwable t) {
    // 使用CAS 的方式设置当前任务状态为正在完成中
    // 有没有可能失败呢？
    // 只有一种情况：外部线程等不及了，直接爱set执行CAS之前，将当前任务cancel了，小概率事件
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        // outcome 引用的是 callable 向上层抛出来的异常
        outcome = t;
        // 将当前任务状态修改为 EXCEPTIonAL 当前任务执行过程中发生了异常，内部封装的 callable 对象的 run() 再执行过程中向上抛出了异常
        UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
        // 回头再说
        finishCompletion();
    }
}

2、get() 方法

// 场景：多个线程等待当前任务完成后的结果
public V get() throws InterruptedException, ExecutionException {
    // 获取当前任务状态
    int s = state;
    // 条件成立：未执行、正在执行、正在结束但是尚未完全结束，调用get的外部线程会被阻塞在外部方法上
    if (s <= COMPLETING)
        // 返回task当前的状态，可能当前线程在里面已经park一会了
        s = awaitDone(false, 0L);
    return report(s);
}

// awaitDone 方法：awaitDone 方法是futureTask 实现阻塞的关键方法

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()方法会将 Thread 当前线程的中断标志位重新设置为false
        if (Thread.interrupted()) {
            // 当前线程的waitNode出队
            removeWaiter(q);
            // 向get方法抛出中断异常
            throw new InterruptedException();
        }

        // 假设当前线程是被其他线程使用unpark(thread) 唤醒的话，会正常自旋，走下面的逻辑
        // 获取当前任务的最新状态
        int s = state;
        // 条件成立：说明当前任务已经有结果了，可能是好结果，可能是坏
        if (s > COMPLETING) {
            // 条件成立：说明已经为当前线程创建过node了，此时需要将node.thread = null，help GC
            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对象了，但是 WaitNode对象还未入队
        else if (!queued)
            // 将当前线程的WaitNode节点指向原队列的头结点，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操作的线程就会被挂起了，线程状态会变为WAITING状态，相当于休眠了
            // 除非有其他线程将你唤醒或者将当前线程中断。
            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);
}

// 当前线程所对应的waitNode出队
private void removeWaiter(WaitNode node) {
    // 条件成立：链表不为空
    if (node != null) {
        // 将当前线程的waitNode的thread为null
        node.thread = null;
        retry:
        // 自旋 *** 作
        for (;;) {          // restart on removeWaiter race
            // 对该链表进行迭代
            for (WaitNode pred = null, q = waiters, s; q != null; q = s) {
                // 将s设置为当前节点的下一个节点
                s = q.next;
                // 条件成立：说明q不是我们当前线程所对应的waitNode节点
                if (q.thread != null)
                    pred = q;
                // 前置条件：说明当前遍历的节点就是我们当前线程所对应的waitNode节点
                // 当pred前驱节点不为空的时候，说明我们当前线程所对应的waitNode节点不是头结点
                else if (pred != null) {
                    // 将前驱节点的next指针指向当前遍历节点的下一个节点
                    pred.next = s;
                    // 这里为什么要判断? 为了防止pred也再执行出队逻辑，所以返回外层重新自旋
                    if (pred.thread == null) // check for race
                        continue retry;
                }
                // 前置条件：说明我们当前线程所对应的waitNode节点是头结点
                // 所以，只需要通过CAS方式将当前遍历的头节点q设置为q的next节点s
                // 然后继续迭代判断链表中其他节点的线程是否需要出队
                else if (!UNSAFE.compareAndSwapObject(this, waitersOffset,
                                                      q, s))
                    continue retry;
            }
            break;
        }
    }
}

// 当线程被挂起之后，如果任务线程执行完毕，就会唤醒等待线程，这一步就是在finishCompletion中做的

private void finishCompletion() {
    // assert state > COMPLETING;
    // 自旋 *** 作，q指向waiters链表的头结点
    for (WaitNode q; (q = waiters) != null;) {
        // 使用 CAS 方式设置waiters为null是因为怕外部线程使用cancel取消当前任务也会触发finishComplete方法，小概率事件
        // 竞争失败的线程会通过外层循环退出，因为CAS把waiters置为了null，所以会退出循环
        if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
            // 自旋 *** 作
            for (;;) {
                // 获取当前waitNode节点封装的 thread
                Thread t = q.thread;
                // 条件不成立：说明当前线程不为null
                if (t != null) {
                    q.thread = null; // help GC
                    // 唤醒当前节点对应的线程
                    LockSupport.unpark(t);
                }
                // next：当前节点的下一个节点
                WaitNode next = q.next;
                if (next == null)
                    break;
                q.next = null; // unlink to help gc
                q = next;
            }
            break;
        }
    }

    done();

    // 将callable设置为null，help GC
    callable = null;        // to reduce footprint
}

3、cancel() 方法

// 将当前线程的任务取消（中断）掉
// mayInterruptIfRunning：true 表示让线程处于正在中断状态 INTERRUPTING
// false：表示让线程处于CANCELLED状态
public boolean cancel(boolean mayInterruptIfRunning) {
    // 条件一：state == NEW 成立：表示当前任务处于运行中或者处于线程池任务队列中
    // 条件二：UNSAFE.compareAndSwapInt(this, stateOffset, NEW, mayInterruptIfRunning ? INTERRUPTING : CANCELLED)
    //  条件成立：说明修改状态成功，可以执行下面逻辑了，否则返回false，表示cancel失败
    if (!(state == NEW &&
          UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
              mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
        return false;
    try {    // in case call to interrupt throws exception
        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;
}

FutureTask 源码分析就到此结束了，下面会开始分析线程池源码一系列知识。