Exectors框架 源码分析

// 返回一个 Future 对象，表示将要执行的线程的状态
// Future 对象的 get 方法会一直阻塞直到算出值
<T> Future<T> submit(Callable<T> task);
// Runnable 本来是不返回 Future 对象的，这里使用了一个参数传递
<T> Future<T> submit(Runnable task, T result);
// 普通的执行，类似于 execute 方法
Future<?> submit(Runnable task);

// 启动一次顺序关闭，执行以前提交的任务，但不接受新任务。
void shutdown();
// 尝试终止正在执行的线程，暂停处理正在等待的任务，并返回等待执行的任务列表。
List<Runnable> shutdownNow();
//当前线程结束
boolean isShutdown();
//全部结束
boolean isTerminated();

//  调用的都是 newTaskFor 的 Runnable 参数方法
public Future<?> submit(Runnable task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<Void> ftask = newTaskFor(task, null);
    execute(ftask);
    return ftask;
}
public <T> Future<T> submit(Runnable task, T result) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<T> ftask = newTaskFor(task, result);
    execute(ftask);
    return ftask;
}
//  调用的 newTaskFor 的 Callable 方法
public <T> Future<T> submit(Callable<T> task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<T> ftask = newTaskFor(task);
    execute(ftask);
    return ftask;
}

// 这两个是 AES 的核心方法，主要就是创建任务  是 submit 的依赖
protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
    return new FutureTask<T>(runnable, value);
}
protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
    return new FutureTask<T>(callable);
}

// 取消任务
boolean cancel(boolean mayInterruptIfRunning);
// 状态判断
boolean isCancelled();
boolean isDone();
// 阻塞获取结果
V get() throws InterruptedException, ExecutionException;
// 超时阻塞获取结果
V get(long timeout, TimeUnit unit)

// 任务状态
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;
// 结果
private Object outcome; // non-volatile, protected by state reads/writes
// 执行任务的线程，他是 volatile 的
private volatile Thread runner;

// 任务的后备队列
    private final BlockingQueue<Runnable> workQueue;
    // 锁
    private final ReentrantLock mainLock = new ReentrantLock();
    // 用来支持等待中断的
    private final Condition termination = mainLock.newCondition();
    // 存放的工作线程，只有当获取到锁的时候才能访问这个 Set
    private final HashSet<Worker> workers = new HashSet<Worker>();
    // 线程池最大数量
    private int largestPoolSize;
    // 完成的线程数，只有在获取锁的时候才能更新这个值
    private long completedTaskCount;

    //==============================================================================
    // 这里有提到用户自定义的变量我们都是用 volatile 来修饰 以保证获取到最新的值
    //==============================================================================
    // 线程创建工厂类
    private volatile ThreadFactory threadFactory;
    // 当任务队列饱和或者线程池关闭后 再往里面提交任务时候的执行策略
    private volatile RejectedExecutionHandler handler;
    // 默认的执行策略是采用的 AbortPolicy (这是一个函数式接口的子类，里面实现的方法默认是抛异常)
    private static final RejectedExecutionHandler defaultHandler = new AbortPolicy();
    // 非核心线程的存活时间
    private volatile long keepAliveTime;
    // 是否允许核心线程具有存活时间，允许则上面的参数也会作用于核心线程
    private volatile boolean allowCoreThreadTimeOut;
    // 核心线程的大小
    private volatile int corePoolSize;
    // 最大线程数
    private volatile int maximumPoolSize;
    // 池控参数  非常重要！！！！
    private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
    // ctl 的解包 -> workerCount 和 runState
    private static int runStateOf(int c)     { return c & ~CAPACITY; }
    private static int workerCountOf(int c)  { return c & CAPACITY; }
    // 打包操作 两个变量或一下
    private static int ctlOf(int rs, int wc) { return rs | wc; }

// 线程
final Thread thread;
// 该线程当前领到的任务
Runnable firstTask;

public ThreadPoolExecutor(int corePoolSize,  //核心线程容量
                          int maximumPoolSize,//线程池的允许线程的最大容量
                          long keepAliveTime,//存活时间
                          TimeUnit unit,//时间单位
                          BlockingQueue<Runnable> workQueue,//任务队列
                                                                             ThreadFactory threadFactory) {//线程工厂
       }

public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        int c = ctl.get();
        // 如果正在运行的线程数小于核心池大小 可以添加一个新的线程领取这个任务
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            // 如果说添加新线程失败了我们需要重新获取线程池的工作程数量，可能有变化
            // 等看到 addWorker 的时候再说
            c = ctl.get();
        }
        // 如果线程池在工作状态，我们就把这个任务放到后备队列
        if (isRunning(c) && workQueue.offer(command)) {
            // 重新检查 ctl
            int recheck = ctl.get();
            // 如果线程池关闭了，我们需要做回退动作，也就是撤销刚才放入的任务
            // 如果撤销成功，执行拒绝策略
            if (! isRunning(recheck) && remove(command))
                reject(command);
            // 如果撤销失败，并且没有工作线程不管他
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        // 线程池已经关闭了 添加线程也失败则执行拒绝
        else if (!addWorker(command, false))
            reject(command);
    }