java并发学习之Executor源码解析

@FunctionalInterface
public interface Runnable {
    public abstract void run();
}

Runnable task = () -> {
    // do something
};
new Thread(task).start(); // 开启新线程执行

public interface Executor {
    void execute(Runnable command);
}

public interface ExecutorService extends Executor {
    void shutdown();
    Future<?> submit(Runnable task);
    <T> Future<T> submit(Callable<T> task);
    ......
}

/**
 * An {@link Executor} that provides methods to manage termination and
 * methods that can produce a {@link Future} for tracking progress of
 * one or more asynchronous tasks.

@FunctionalInterface
public interface Callable<V> {
    V call() throws Exception;
}

public Future<?> submit(Runnable task) {
    if (task == null) throw new NullPointerException();
    // 生成一个依赖于原任务的新任务
    RunnableFuture<Void> ftask = newTaskFor(task, null);
    // 执行新任务
    execute(ftask);
    // 返回新任务(充当任务跟踪器)
    return ftask;
}

protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
    return new FutureTask<T>(runnable, value);
}

public interface RunnableFuture<V> extends Runnable, Future<V> {
    void run();
}

public FutureTask(Callable<V> callable) {
    if (callable == null)
        throw new NullPointerException();
    this.callable = callable;
    this.state = NEW;       // ensure visibility of callable
}

public void run() {
    // 如果任务不是新状态，直接返回
    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 {
                // 执行原任务
                result = c.call();
                ran = true;
            } catch (Throwable ex) {
                result = null;
                ran = false;
                // 有异常设置异常
                setException(ex);
            }
            if (ran)
                // 设置outcome存储返回值
                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);
    }
}

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; // 返回结果即outcome
    if (s >= CANCELLED)
        throw new CancellationException();
    throw new ExecutionException((Throwable)x);
}

public boolean cancel(boolean mayInterruptIfRunning) {
    // 状态变为CANCELLED
    if (!(state == NEW &&
          UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
              mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
        return false;
    // 如果正在运行调用interrupt阻断正在执行的任务
    try {    // in case call to interrupt throws exception
        if (mayInterruptIfRunning) {
            try {
                Thread t = runner;
                if (t != null)
                    t.interrupt();
            } finally { // final state
                UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
            }
        }
    } finally {
        finishCompletion();
    }
    return true;
}