Java 线程池 submit 阻塞获取结果实现原理

@Testpublic void testSubmit() throws ExecutionException, InterruptedException {    // 创建一个核心线程数为5的线程池    ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 10, 30, TimeUnit.SECONDS, new ArrayBlockingQueue<>(50));
    // 创建一个计算任务    Callable<Integer> myTask = new Callable<Integer>() {
        @Override        public Integer call() throws Exception {            int result = 0;            for (int i = 0; i < 10000; i++) {                result += i;            }            Thread.sleep(1000);            return result;        }    };
    log.info("start submit task .....");    Future<Integer> future = threadPoolExecutor.submit(myTask);
    Integer sum = future.get();    log.info("get submit result: [{}]", sum);
    // use sum do other things}

// AbstractExecutorService#submitpublic <T> Future<T> submit(Callable<T> task) {    // 判空处理    if (task == null) throw new NullPointerException();    // 将提交的任务包装成RunnableFuture    RunnableFuture<T> ftask = newTaskFor(task);    // 最终还是调用execute方法执行任务    execute(ftask);    return ftask;}

// AbstractExecutorService#newTaskForprotected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {    // 调用FutureTask的构造方法返回RunnableFuture对象    return new FutureTask<T>(callable);}

// 存储当前任务运行状态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;

// 存放真正提交的原始任务private Callable<V> callable;

返回的结果或从get()中抛出的异常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();        // 设置要执行的任务        this.callable = callable;        // 初始化时任务状态为NEW        this.state = NEW;       }

public void run() {    // 状态机不为NEW表示执行完成或任务被取消了，直接返回    // 状态机为NEW，同时将runner设置为当前线程，保证同一时刻只有一个线程执行run方法，如果设置失败也直接返回    if (state != NEW ||        !UNSAFE.compareAndSwapObject(this, runnerOffset,                                     null, Thread.currentThread()))        return;    try {        Callable<V> c = callable;        // 取出原始的任务检测不为空 且 再次检查状态为NEW(双重校验)        if (c != null && state == NEW) {            // 任务运行的结果            V result;            // 任务是否运行是否正常， true：正常， false-异常            boolean ran;            try {                // 任务执行，将结果返回给result                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        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) {    // 将状态由NEW更新为COMPLETING    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {        // 保存任务的结果        outcome = v;       // 更新状态的最终状态-NORMAL        UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state        // 通用的完成操作，主要作用就是唤醒阻塞在waiters队列中请求get的线程        finishCompletion();    }}

// FutureTask#setExceptionprotected void setException(Throwable t) {    // 将状态由NEW更新为COMPLETING    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {        // 将异常信息保存到输出结果中        outcome = t;        // 更新状态机处理异常的最终状态-EXCEPTIONAL        UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state        // 通用的完成操作，主要作用就是唤醒阻塞在waiters队列中请求get的线程        finishCompletion();    }}

// 阻塞获取任务结果public V get() throws InterruptedException, ExecutionException {    int s = state;    // 任务还没有执行完成，通过awaitDone方法进行阻塞等待    if (s <= COMPLETING)        s = awaitDone(false, 0L);    // 返回结果    return report(s);}
// 超时阻塞获取任务结果public V get(long timeout, TimeUnit unit)    throws InterruptedException, ExecutionException, TimeoutException {    // 判空处理    if (unit == null)        throw new NullPointerException();    int s = state;    // 任务还没有执行完成，通过awaitDone方法进行阻塞等待    if (s <= COMPLETING &&        // 如果awaitDone返回的结果还是小于等于COMPLETING，表示运行中，那么直接抛出超时异常        (s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING)        throw new TimeoutException();    // 返回结果    return report(s);}

// 线程阻塞等待方法， timed等于 true表示阻塞等待有时间限制nanos, false表示没有，一直阻塞private int awaitDone(boolean timed, long nanos) throws InterruptedException {    // 计算阻塞超时时间点    final long deadline = timed ? System.nanoTime() + nanos : 0L;    WaitNode q = null;    // 表示q是否添加到waiters栈中，默认false    boolean queued = false;    // 自旋操作    for (;;) {        // 如果阻塞线程被中断则将当前线程从阻塞队列中移除        if (Thread.interrupted()) {            // 从waiters栈中移除WaitNode，            removeWaiter(q);            // 返回中断移除            throw new InterruptedException();        }
        // 获取任务的状态        int s = state;        // 如果任务的状态大于COMPLETING，表示线程运行结束了，直接返回        if (s > COMPLETING) {             // 任务已经完成时直接返回结果            if (q != null)                q.thread = null;            // 返回状态            return s;        }        // 如果任务状态是COMPLETING             else if (s == COMPLETING)             // 如果任务执行完成，但还差最后一步最终完成，则让出CPU给任务执行线程继续执行            Thread.yield();        // 如果任务状态小于COMPLETING，说明任务还在运行中        // 如果q为空的情况            else if (q == null)            // 新进来的线程添加等待节点            q = new WaitNode();        // 如果任务还在运行中并且当前线程节点还不在waiters栈中，那么就加入           else if (!queued)            // 上一步节点创建完，还没将其添加到waiters栈中，因此在下一个循环就会执行此处进行入栈操作，并将当前线程的等待节点置于栈顶            queued = UNSAFE.compareAndSwapObject(this, waitersOffset,                                                 q.next = waiters, q);        // 如果任务还在运行中并且timed为true，表示有超时限制        else if (timed) {            // 如果设置了阻塞超时时间，则进行检查是否达到阻塞超时时间，达到了则删除当前线程的等待节点并退出循环返回，否则继续阻塞            nanos = deadline - System.nanoTime();            // 如果nanos小于等于0            if (nanos <= 0L) {                // 从waiters栈中移除                removeWaiter(q);                //返回状态                return state;            }            // 超时阻塞当前线程，超过时间，就会恢复            LockSupport.parkNanos(this, nanos);        }        // 如果任务还在运行中并且timed为false，没有有超时限制        else            // 一直阻塞当前线程            LockSupport.park(this);    }}

// 获取任务结果方法：正常执行则直接返回结果，否则抛出异常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 void finishCompletion() {    // 遍历waiters栈中的每个元素;    for (WaitNode q; (q = waiters) != null;) {        // cas设置waiters中q节点数据为null，成功的话，进入到if中        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;                q.next = null; // unlink to help gc                q = next;            }            break;        }    }  //钩子方法，有子类去实现    done();    // 设置原来的任务callable为null    callable = null;        // to reduce footprint}

// 取消任务，参数mayInterruptIfRunning为true，会中断运行中的线程，false不会public boolean cancel(boolean mayInterruptIfRunning) {    // 如果FutureTask的状态不是NEW或者CAS设置失败时，直接返回false    if (!(state == NEW &&          UNSAFE.compareAndSwapInt(this, stateOffset, NEW,              mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))        return false;    try {           // 如果参数mayInterruptIfRunning为true,中断        if (mayInterruptIfRunning) {            try {                Thread t = runner;                if (t != null)                    t.interrupt();            } finally { // final state                //cas修改状态为INTERRUPTED                UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);            }        }    } finally {        // 唤醒其他等待的线程        finishCompletion();    }    return true;}