Java线程池一：基本概念与FutureTask

概述

Java的线程池一直在用，最近的抽奖也用到线程池来实现异步，但是对于线程的配置还有具体的实现还是很懵懂的，所以打算对线程池的源码进行阅读。

解析

jdk提供的线程池的基本入口是 java.util.concurrent.Executor

Executor可以理解为线程的执行器，优雅的解耦了任务处理机制中的任务提交和任务如何运行（也包含线程的使用，调度）

Executor

public interface Executor {
    /**
     * Executes the given command at some time in the future.  The command
     * may execute in a new thread, in a pooled thread, or in the calling
     * thread, at the discretion of the {@code Executor} implementation.
     *
     * @param command the runnable task
     * @throws RejectedExecutionException if this task cannot be
     * accepted for execution
     * @throws NullPointerException if command is null
     */
    void execute(Runnable command);
}

Executor框架包括3大部分:

1. 任务。也就是工作单元，包括被执行任务需要实现的接口：Runnable接口或者Callable接口;
2. 任务的执行。也就是把任务分派给多个线程的执行机制，包括Executor接口及继承自Executor接口的ExecutorService接口;
3. 异步计算的结果。包括Future接口及实现了Future接口的FutureTask类。

runnable实现类，RunnableFuture是jdk线程池运行单元，它的实现类是FutureTask；通过RunnableFuture对运算单元的封装使得线程池与Runnable、Callable接口解耦。

ExecutorService

ExecutorService在Executor的基础上定义了更多的方法，例如任务批量提交、是否停止、是否中断等。

Executor实现类:

RunnableFuture

RunnableFuture接口统一定义了运行单元。

public interface RunnableFuture<V> extends Runnable, Future<V> {
    /**
     * Sets this Future to the result of its computation
     * unless it has been cancelled.
     */
    void run();
}

FutureTask

继承关系

public class FutureTask<V> implements RunnableFuture<V> {}

成员变量

// 运行状态
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;

/** The underlying callable; nulled out after running */
private Callable<V> callable;
/** The result to return or exception to throw from get() */
private Object outcome; // non-volatile, protected by state reads/writes
/** The thread running the callable; CASed during run() */
private volatile Thread runner; // 运行这个单元的线程
/** Treiber stack of waiting threads */
private volatile WaitNode waiters; //单向链，每个节点存储等待的线程对象

// jdk9新增的变量句柄，用于替换unsafe类
private static final VarHandle STATE;
private static final VarHandle RUNNER;
private static final VarHandle WAITERS;

构造器

如果是Runnable对象会使用Executors.callable工具进行转换（适配器）。

初始状态是 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
}

主要方法

执行且记录返回值run()

public void run() {
    // 状态必须是NEW，且设置执行线程为当前线程成功
    if (state != NEW ||
        !RUNNER.compareAndSet(this, null, Thread.currentThread()))
        return;
    try {
        Callable<V> c = callable;
        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 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)
            handlePossibleCancellationInterrupt(s);
    }
}

设置异常setException(Throwable t)

状态改为完成COMPLETING

protected void setException(Throwable t) {
    if (STATE.compareAndSet(this, NEW, COMPLETING)) {
        outcome = t;
        STATE.setRelease(this, EXCEPTIONAL); // final state
        finishCompletion();
    }
}

设置返回值set()

状态改为完成COMPLETING

protected void set(V v) {
    if (STATE.compareAndSet(this, NEW, COMPLETING)) {
        outcome = v;
        STATE.setRelease(this, NORMAL); // final state
        finishCompletion();
    }
}

完成后续操作finishCompletion()

释放WaitNode（等待运行结束获取结果的线程）

private void finishCompletion() {
    // assert state > COMPLETING;
    for (WaitNode q; (q = waiters) != null;) {
        // waiters 置空
        if (WAITERS.weakCompareAndSet(this, q, null)) {
            // 释放WaitNode链表
            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;        // to reduce footprint
}

执行并且不记录返回值 runAndReset()

protected boolean runAndReset() {
    if (state != NEW ||
        !RUNNER.compareAndSet(this, null, Thread.currentThread()))
        return false;
    boolean ran = false;
    int s = state;
    try {
        Callable<V> c = callable;
        if (c != null && s == NEW) {
            try {
                // 执行不关心放回值
                c.call(); // don't set result
                ran = true;
            } catch (Throwable ex) {
                setException(ex);
            }
        }
    } 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
        s = state;
        if (s >= INTERRUPTING)
            handlePossibleCancellationInterrupt(s);
    }
    return ran && s == NEW;
}

可能中断处理handlePossibleCancellationInterrupt(int s)

中断状态让出cpu时间片

private void handlePossibleCancellationInterrupt(int s) {
    // It is possible for our interrupter to stall before getting a
    // chance to interrupt us.  Let's spin-wait patiently.
    if (s == INTERRUPTING)
        while (state == INTERRUPTING)
            Thread.yield(); // wait out pending interrupt

    // assert state == INTERRUPTED;

    // We want to clear any interrupt we may have received from
    // cancel(true).  However, it is permissible to use interrupts
    // as an independent mechanism for a task to communicate with
    // its caller, and there is no way to clear only the
    // cancellation interrupt.
    //
    // Thread.interrupted();
}

取消cancel(boolean mayInterruptIfRunning)

NEW状态才能取消或者中断

public boolean cancel(boolean mayInterruptIfRunning) {
    // 设置状态
    if (!(state == NEW && STATE.compareAndSet
            (this, NEW, mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
        return false;
    try {    // in case call to interrupt throws exception
        // 中断
        if (mayInterruptIfRunning) {
            try {
                Thread t = runner;
                if (t != null)
                    t.interrupt();
            } finally { // final state
                STATE.setRelease(this, INTERRUPTED);
            }
        }
    } finally {
        finishCompletion();
    }
    return true;
}

获取返回值get()

public V get() throws InterruptedException, ExecutionException {
    int s = state;
    if (s <= COMPLETING)
        s = awaitDone(false, 0L);
    return report(s);
}
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);
}

等待执行完成awaitDone(boolean timed, long nanos)

等待执行完成或者等待一定的时间，返回当前状态。

// time是否有等待时间限制
private int awaitDone(boolean timed, long nanos)
    throws InterruptedException {
    // The code below is very delicate, to achieve these goals:
    // - call nanoTime exactly once for each call to park
    // - if nanos <= 0L, return promptly without allocation or nanoTime
    // - if nanos == Long.MIN_VALUE, don't underflow
    // - if nanos == Long.MAX_VALUE, and nanoTime is non-monotonic
    //   and we suffer a spurious wakeup, we will do no worse than
    //   to park-spin for a while
    long startTime = 0L;    // Special value 0L means not yet parked
    WaitNode q = null;
    boolean queued = false;
    for (;;) {
        int s = state;
        // 如果已经完成则返回状态
        if (s > COMPLETING) {
            if (q != null)
                q.thread = null;
            return s;
        }
        else if (s == COMPLETING) // 刚完成再等待一下
            // We may have already promised (via isDone) that we are done
            // so never return empty-handed or throw InterruptedException
            Thread.yield();
        else if (Thread.interrupted()) { // 当前线程中断了，将当前线程移出等待队列
            removeWaiter(q);
            throw new InterruptedException();
        }
        else if (q == null) {
            if (timed && nanos <= 0L)
                return s;
            q = new WaitNode();
        }
        else if (!queued) // 还未在等待队列中，则加入
            queued = WAITERS.weakCompareAndSet(this, q.next = waiters, q);
        else if (timed) { // 有等待时间限制
            final long parkNanos;
            if (startTime == 0L) { // first time
                startTime = System.nanoTime();
                if (startTime == 0L)
                    startTime = 1L;
                parkNanos = nanos; // 需要等待的时间
            } else { // 已经醒来
                long elapsed = System.nanoTime() - startTime;
                if (elapsed >= nanos) {
                    removeWaiter(q);
                    return state;
                }
                parkNanos = nanos - elapsed; // 继续睡
            }
            // nanoTime may be slow; recheck before parking
            if (state < COMPLETING)  // 等待对应的时间
                LockSupport.parkNanos(this, parkNanos);
        }
        else
            LockSupport.park(this);
    }
}

参考

Java并发——Executor框架详解（Executor框架结构与框架成员）