目标

1、线程池中线程的运行状态
2、worker的计算
3、线程池中ctr的作用
4、任务添加流程
5、任务处理的逻辑

总结

先写出流程结论, 后续再分析时通过代码对这些结论进行验证;
  1、添加Task时, 首先判断WorkerCount与corePoolSize的关系, 如果WorkerCount < corePoolSize, 此时会无条件创建一个Worker来执行Task, 同时将ctl的低位执行+1操作, 此时在创建Worker时, 会将Task与Worker进行关联, Worker执行完Task以后将Task进行释放;
  2、如果WorkerCount ≥ corePoolSize, 此时首先会将Task添加到任务队列中, 添加完成之后, 会再次判断WorkerCount, 此时只有当WorkerCount == 0时才会再次创建新的WorkerCount来执行任务, 但是此时通过addWorker创建Worker时, 没有将Worker与Task进行关联. 到这里有三个疑问: 1. WorkerCount ≥ corePoolSize与WorkerCount==0有没有冲突? 2. 如果后者WorkerCount > 0, 并不会调用addWorker创建新的线程, 那么此时提交的Task被谁执行? 3. 如果后者WorkerCount == 0, 那么addWorker创建的Worker又是如何执行Task的呢?
  3、如果WorkerCount ≥ corePoolSize && workQueue.offer(task) == false, 也就是说第二个条件任务队列已满的情况下, 此时会触发addWorker尝试为当前Task创建新的Worker, 此时为何说是尝试? 因为如果线程池中线程数量已经大于线程池运行的最大线程数时, 此时将不会再为新的Task创建Worker, 而是采取拒绝该Task的方式
  4、接下来是任务处理的流程, 同时也回答了第二步中的几个疑问, 当线程执行完Task之后, 尝试从任务队列中获取Task, 针对线程获取Task分两种情况: (1) 核心线程获取Task, 如果不允许核心线程超时的情况下, 如果workQueue.isNotEmpty, 此时核心线程将一直处于挂起状态, 直到任务队列中添加了元素. (2) 针对普通线程, 如果当前workQueue.isEmpty, 那么普通线程会挂起指定时间后进行释放操作, 同时执行ctr--操作.
  5、所以第二条的疑问也得到的解释, 任务被添加到任务队列之前, 线程池中的线程刚好全部被释放, 此时会重新创建线程执行workQueue中的Task. 任务被添加到任务队列之后, 如果WorkerCount != 0, 此时会直接唤醒等待挂起的线程, 避免线程的不必要创建.

一、关于Executor

1.1 线程池的顶层接口Executor

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

1.2 线程池的第二层接口ExecutorService

public interface ExecutorService extends Executor {
    void shutdown();
    List<Runnable> shutdownNow();
    boolean isShutdown();
    boolean isTerminated();
    <T> Future<T> submit(Callable<T> task);
    <T> Future<T> submit(Runnable task, T result);
    Future<?> submit(Runnable task);
}

1、我们经常用到的几个线程池都是实现于ExecutorService接口, ExecutorService对Executor进行了扩展, 支持Callable和Runnable;
2、先不考虑ExecutorService每个方法到底什么作用, 待后边分析到线程池管理线程流程时自然明了;

1.3 线程池的提供者Executors:

public class Executors {
    public static ExecutorService newFixedThreadPool(int nThreads) {...}
    public static ExecutorService newWorkStealingPool() {...}
    public static ExecutorService newSingleThreadExecutor() {...}
    public static ExecutorService newCachedThreadPool() {...}
    public static ScheduledExecutorService newSingleThreadScheduledExecutor() {...}
    public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {...}
    private Executors() {}
}

  1、先忽略这些方法的内部实现, 以及传入的参数, 现在分析还为时过早, 属于强行记忆;
  2、这些方法主要返回两大类线程池ExecutorService和ScheduledExecutorService, 而ScheduledExecutorService又是继承于ExecutorService, 所以后续会重点关注这两个类;
  3、接下来开始分析ScheduledExecutorService和ExecutorService是如何管理内部维护的线程池;
  4、对Executors 方法展开, 发现ExecutorService和ScheduledExecutorService分别指向ThreadPoolExecutor和ScheduledThreadPoolExecutor, 而这两个类又都间接实现ExecutorService接口, 所以都支持execute(RunnableImpl)和submit(CallableImpl)方式;
先分析execute(RunnableImple)这种方式;
  5、关于四个方法的具体分析在模块三

在进行分析之前, 先列出线程池中线程的几个状态以及构建线程池时所需要的参数, 后续分析源码时遇到即回头来进行补充说明:
线程的几个状态值 :

private static final int RUNNING    = -1 << COUNT_BITS;
private static final int SHUTDOWN   =  0 << COUNT_BITS;
private static final int STOP       =  1 << COUNT_BITS;
private static final int TIDYING    =  2 << COUNT_BITS;
private static final int TERMINATED =  3 << COUNT_BITS;

构建线程池时所需参数 :

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {
        ...
}

下面结合对ThreadPoolExecutor源码的分析, 对每个参数的作用进行总结描述:

二、分析线程池里面涉及到的位运算

public class ThreadPoolExecutor extends AbstractExecutorService {
    // 默认情况下ctl = 11100000 00000000 00000000 00000000;
    private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
    private static final int COUNT_BITS = Integer.SIZE - 3;
    // CAPACITY = 0001111 11111111 11111111 11111111
    // ~CAPACITY = 11100000 00000000 00000000 00000000
    private static final int CAPACITY   = (1 << COUNT_BITS) - 1;
    // runState is stored in the high-order bits
    // 将线程池的运行状态记录在bits的高位
    // RUNNING = 11100000 00000000 00000000 00000000
    private static final int RUNNING    = -1 << COUNT_BITS;
    // SHUTDOWN = 00000000 00000000 00000000 00000000
    private static final int SHUTDOWN   =  0 << COUNT_BITS;
    // STOP = 00100000 00000000 00000000 00000000
    private static final int STOP       =  1 << COUNT_BITS;
    // TIDYING = 01000000 00000000 00000000 00000000
    private static final int TIDYING    =  2 << COUNT_BITS;
    // TERMINATED = 01100000 00000000 00000000 00000000
    private static final int TERMINATED =  3 << COUNT_BITS;
    // 通过计算获取高位三位的值, 进而判断当前线程池的运行状态
    private static int runStateOf(int c)     { return c & ~CAPACITY; }
    // CAPACITY的高三位是0, 所以线程池中Worker的数量存储在低位29位中.
    private static int workerCountOf(int c)  { return c & CAPACITY; }
    //
    private static int ctlOf(int rs, int wc) { return rs | wc; }
    /*
     * Bit field accessors that don't require unpacking ctl.
     * These depend on the bit layout and on workerCount being never negative.
     */
    private static boolean runStateLessThan(int c, int s) {
        return c < s;
    }
    private static boolean runStateAtLeast(int c, int s) {
        return c >= s;
    }
    private static boolean isRunning(int c) {
        return c < SHUTDOWN;
    }
}

上面的运算结合下面实际操作进行分析.

三、任务提交

3.1 ThreadPoolExecutor.execute

注意Thread与Worker的关系

虽然代码少, 但是要表达的意思却是很多的;
public void execute(Runnable command) {
    // c默认 = RUNNING, 二进制 = 11100000 00000000 00000000 00000000
    int c = ctl.get();
    // 1. 通过workerCountOf内部与CAPACITY进行与运行得到当前线程池中Worker的数量;
    // 2. 然后将当前WorkerCount与corePoolSize进行比较, 如果WorkerCount < corePoolSize, 
    //    则进入if内部通过addWorker尝试为当前Runnable创建线程.
    // 3. 如果当前WorkerCount ≥ corePoolSize
    if (workerCountOf(c) < corePoolSize) {
        // 1. 注意此时传参command(command != null), true;
        // 2. 针对返回的结果, 有以下几种情况:
        //    true:   结合addWorker源码可知, 如果线程池处于运行状态, 返回true;
        //    false:  只有线程池处于非运行状态, 才可能返回false, 这种情况不考虑;
        if (addWorker(command, true))  
            return;
        // 再次获取当前线程池的一个状态
        c = ctl.get();
    }
    // 1. 因为我们一直假设线程池处于运行状态, 所以此时默认isRunning(c)=true, 直接分析第二个条件;
    // 2. 针对第二个条件, workQueue.offer:
    //    (1) true: 任务队列未满;
    //    (2) false: 任务队列已满;  
    // 3. 然后如果当WorkerCount ≥ corePoolSize时, 首先将任务进行入队操作.
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        // 只考虑线程池处于RUNNGING的场景, 跳过该if语句;
        if (!isRunning(recheck) && remove(command))
            reject(command);
        // 再次获取线程池中Worker的数量, 此时只有当WorkerCount == 0时, 才会进入到if语句中
        // 调用addWorker创建新的Worker实例, 而结合上下文可知, Worker与Thread是一个1:1对应
        // 的关系, 所以如果WorkerCount == 0, 也就是说当前线程池中线程数量为0;
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);     
    }
    // 1. 能执行到这里需要满足以下几个条件:
    //    (1) wc > corePoolSize即当前线程池中工作线程数量＞核心线程数量;
    //    (2) workQueue.offer(command) = false, 即任务队列已满;
    // 2. 结合addWorker源码分析可知, 如果任务被成功执行, 返回true, 如果WorkerCount ≥ maximumPoolSize, 
    //    返回false, 此时对该任务才去拒绝策略;
    else if (!addWorker(command, false))
        reject(command);    
    }
}

关于传值的问题, 这里用一张表进行总结: 接下表中数据对execute代码进行分析

对上述表中的以及代码中的注释再次解释一下:
1、添加任务时, 如果WorkerCount < corePoolSize, 那么此时会直接创建一个线程来执行该任务;
2、添加任务时, 如果WorkerCount ≥ corePoolSize, 那么此时会先将当前线程进行入队操作, 然后再次判断WorkerCount, 只有当WorkerCount == 0 时, 才会触发addWorker创建Thread去执行任务队列中的任务. 现在思考一个问题, 如果WorkerCount != 0, 提交的任务被谁处理了呢?
3、如果WorkerCount ≥ corePoolSize并且任务队列已满, 此时会尝试调用addWorker创建新的线程来执行提交的任务.

3.2 ThreadPoolExecutor.addWorker

private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    for (;;) {
        // 再次获取当前线程池的状态: 高位存储线程池的运行状态.
        int c = ctl.get();
        // 通过高位获取当前线程池的运行状态;
        int rs = runStateOf(c);
        // 不考虑线程池中断的情况, 所以继续向下执行;
        if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty()))
            return false;
        for (;;) {
            // 通过低位29位的值获取当前线程池中WorkerCount;
            int wc = workerCountOf(c);
            // 忽略第一个条件, 直接看第二个条件, 第二个条件重点在于core;
            // 而core的传值又依赖于execute里面的几种情况:
            // 1. core = true:
            //    当wc < corePoolSize, 触发addWorker会传入core = true, 所以此时第二个条件为false;
            // 2. core = false:
            //    (1) wc ≥ corePoolSize && workQueue.offer == true && WorkerCount == 0
            //    (2) wc ≥ corePoolSize && workQueue.offer == false
            // 3. 不管core为true还是false, 只要线程池正常运行, 第二个条件就是始终为false;
            if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            // 只要程序能够正常运行, 都会在这里执行ctl+1操作, 而ctl低29位标志的是线程池中线程数量
            // 对ctl执行+1操作
            if (compareAndIncrementWorkerCount(c))
                break retry;
        }
    }
    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        // 关于firstTask取值情况上文已经用表进行了说明, 此处再简短的进行说明:
        // 1. firstTask != null:
        //    (1) wc < corePoolSize;
        //    (2) wc ≥ corePoolSize && workQueue.offer() == flase;
        // 2. firstTask == null:
        //    wc ≥ corePoolSize && workQueue.offer() == true && workerCount == 0;
        w = new Worker(firstTask); 
        final Thread t = w.thread;
        if (t != null) {
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                int rs = runStateOf(ctl.get());
                // 假设线程池是运行状态, 因此第一个条件为true;
                if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) {
                    workers.add(w);
                    int s = workers.size();
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    // 标志位, 通过该变量判断是否返回true/false;
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
            if (workerAdded) {
                // 触发其内部Thread.run的执行;
                t.start();    
                workerStarted = true;
            }
        }
    } finally {
        if (!workerStarted)
            // 如果执行失败, 将ctl进行-1操作;
            addWorkerFailed(w);
    }
    return workerStarted;
}

对addWorker进行总结:
  1、线程池处于正常运行状态, 只要触发了addWorker方法, 不论当前线程池中线程数量多少, 都会重新创建一个新的Worker与Thread, 然后通过该Worker与Thread来执行新的任务;
  2、有一个问题是当wc ≥ corePoolSize时, firstTask需要先被添加到workQueue中, 然后再次判断只有当wc == 0时, 才会触发addWorker创建新的Worker进行任务的执行, 那么问题来了, 此时调用addWorker传入的firstTask = null, 那么线程从哪里获取的任务进行执行呢?
  3、接着第二个条件如果wc != 0, 此时是不会触发addWorker的, 那么此时提交的任务又是被谁执行呢?
上面第二个和第三个问题在Worker.run中会得到答案

3.3 ThreadPoolExecutor.compareAndIncrementWorkerCount执行ctl+1操作

private boolean compareAndIncrementWorkerCount(int expect) {
    //ctl执行+1操作, 执行完这个操作之后, 开始创建Worker实例, 然后结合上下文
    //可知, ctl高三位记录的是当前线程池的状态, 低29位记录的是当前线程池中Worker数量.
    return ctl.compareAndSet(expect, expect + 1);
}

3.4 Worker结构

// Worker与Thread、Runnable是1:1:1对应的一个关系, 但是firstTask可能存在为null的情况.
private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
    final Thread thread;
    Runnable firstTask;
    volatile long completedTasks;
    Worker(Runnable firstTask) {
        setState(-1); // inhibit interrupts until runWorker
        this.firstTask = firstTask;
        this.thread = getThreadFactory().newThread(this);
    }
}

四、关于Worker

4.1 Worker.run执行任务

private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
    public void run() {
        // 触发线程池中的runWorker方法;
        runWorker(this);
    }
}

public class ThreadPoolExecutor {
    final void runWorker(Worker w) {
        Thread wt = Thread.currentThread();
        Runnable task = w.firstTask;
        w.firstTask = null;
        w.unlock(); // allow interrupts
        boolean completedAbruptly = true;
        try {
            // 问题就在这里, 如果task == null的情况下, 当前线程通过getTask()会尝试从
            // 任务队列中获取任务进行执行;
            while (task != null || (task = getTask()) != null) {
                w.lock();
                task.run();
                task = null;
                w.completedTasks++;
                w.unlock();
            }
            completedAbruptly = false;
        } finally {
            processWorkerExit(w, completedAbruptly);
        }
    }
}

4.2 ThreadPoolExecutor.getTask线程从任务队列中获取任务

先写结论:
1. 当允许核心线程超时: 从元素队列中取出元素时, 如果元素队列为空, 当前线程会先挂起指定时间, 然后被结束;
2. 当不允许核心线程超时: 从元素队列中取出元素时, 如果当前元素队列为空, 当前线程为核心线程时, 线程被挂起,
   如果当前线程为非核心工作线程, 当前线程会挂起指定时间然后被结束;
private Runnable getTask() {
    boolean timedOut = false; // Did the last poll() time out?
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);
        // 目前只考虑线程池正常运行的情况, 所以不考虑这段流程;
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            decrementWorkerCount();
            return null;
        }
        // 再次获取WorkerCount;
        int wc = workerCountOf(c);
        // 对time取值有以下几种情况:
        // 1. timed = true:
        //    1.1 allowCoreThreadTimeOut = true: 通过allowCoreThreadTimeOut()进行赋值;
        //    1.2 wc > corePoolSize: 当前线程数量＞核心工作线程数量;
        // 2. timed = false:
        //    2.1 wc ≤ corePoolSize: 当前线程数量 ≤ 核心工作线程数量;
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
        // 第一个条件wc > maximumPoolSize认为是true, 所以重点在于后面两个条件:
        // (1) timed = true && timedOut == true
        //    针对第一个条件timed=true, 也就是说允许核心线程超时或者当前线程数量超过核心线程数量;
        //    第二个条件timedOut作为超时的标志, 结合下文如果r == null, 则认为当前线程超时成功;
        // (2) wc > 1 || workQueue.isEmpty()
        //    在当前线程超时成功的情况下, 如果任务队列此时还是为空的, 那么将会释放掉该线程, 通过
        //    进入if内执行ctl--操作
        // 通过对这段代码进行分析还可以发现一个结论, 对于超时的线程, 再进行释放之前还会再去尝试一下
        // 从任务队列中获取任务, 如果此时任务队列中还是没有任务, 然后释放该线程.
        if ((wc > maximumPoolSize || (timed && timedOut))
            && (wc > 1 || workQueue.isEmpty())) {
            if (compareAndDecrementWorkerCount(c))
                return null;
            continue;
        }
        // 1. timed = true:
        //    此时allowCoreThreadTimeOut = true || wc > corePoolSize;
        // 2. timed = false: 
        //    结合上文可知, 此时allowCoreThreadTimeOut = false && wc ≤ corePoolSize;
        // 3. BlockingQueue的特性就是pool(...)当前线程在等待指定时间后进行元素出队操作;
        // 4. take()方法如果元素队列为空, 当前线程是会一直处于挂起状态;
        Runnable r = timed ?
            workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
            workQueue.take();
        if (r != null)
            return r;
        // 执行到这里需要满足的条件是r == null, 然后通过timedOut来认为获取任务失败是一种超时的情况.
        timedOut = true;
    }
}

五、Executors四个常见线程池

5.1 Executors.newFixedThreadPool

public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
}

5.2 Executors.newSingleThreadExecutor

public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
}

5.3 Executors.newCachedThreadPool

public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
}

5.4 Executors.newScheduledThreadPool

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
    return new ScheduledThreadPoolExecutor(corePoolSize);
}

public class ScheduledThreadPoolExecutor  extends ThreadPoolExecutor implements ScheduledExecutorService {
    public ScheduledThreadPoolExecutor(int corePoolSize) {
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS, new DelayedWorkQueue());
    }
}
public class ThreadPoolExecutor extends AbstractExecutorService {
    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), defaultHandler);
    }
}