ThreadPoolExecutor中知识点很多，本文只是从7个构造参数入手，看看其运转的核心思路。重点不是扣代码，是体会设计思想哈！
欢迎纠错和沟通。

ThreadPoolExecutor

以下是构造ThreadPoolExecutor的7大参数。

public ThreadPoolExecutor(int corePoolSize,

                          int maximumPoolSize,

                          long keepAliveTime,

                          TimeUnit unit,

                          BlockingQueue<Runnable> workQueue,

                          ThreadFactory threadFactory,

                          RejectedExecutionHandler handler) {

corePoolSize 和 maximumPoolSize 以及 workQueue（BlockingQueue）共同决定了线程池中线程的数量。
下面几条是我总结的观点：

1. 
   
2. 核心线程数小于等corePoolSize。
   
3. 普通线程在workQueue满了后才会创建。
   
4. 普通线程在任务结束后存活时长为：keepAliveTime*unit。
   
5. 任务总数如果超过了workQueue的容量+普通线程数，会触发 RejectedExecutionHandler。
   
6. 最好自定义ThreadFactory来创建线程，方便标识线程名等。
   
7. ThreadPoolExecutor提供了hook的方法beforeExecute()、afterExecute()
   
8. shutdown()不会立即停止线程池中的未完成的任务，shutdownNow()会。
   

这里有个设计上的亮点：它使用了一个AtomicInteger类型的ctl来同时记录线程池状态和线程池中线程数。ctl中高3位记录状态，低29位代表线程数量。
这种方法值得学习，除了节省变量，也减少了线程池状态和当前线程数量同步问题。

@ReachabilitySensitive

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

private static final int COUNT_BITS = Integer.SIZE - 3;

private static final int CAPACITY   = (1 << COUNT_BITS) - 1;


// runState is stored in the high-order bits

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;


// Packing and unpacking ctl

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; }

核心代码区域execute()：

public void execute(Runnable command) {

    if (command == null)

        throw new NullPointerException();


    int c = ctl.get();

    if (workerCountOf(c) < corePoolSize) {

        //如果当前worker数量小于corePoolSize则创建新的核心Worker

        if (addWorker(command, true))

            return;

        c = ctl.get();

    }

    //线程池正在运行且可以正常添加任务(即workQueue还没有满)，此时等待任务执行即可。注意此处已经将新的Runnable存储到了workQueue里了。

    if (isRunning(c) && workQueue.offer(command)) {

        int recheck = ctl.get();

        if (! isRunning(recheck) && remove(command))

        //二次确认，此时线程池不属于运行状态，且刚添加进去的任务还没有执行，则reject

            reject(command);

        else if (workerCountOf(recheck) == 0)

        //线程池showdown中，后续应该什么都不会做直接return

            addWorker(null, false);

    }

    //workerQueue中满了，创建非核心worker，如果不成功则reject

    else if (!addWorker(command, false))

        reject(command);

}

看下addWorker()到底干了啥

private boolean addWorker(Runnable firstTask, boolean core) {


   retry:

    for (;;) {

        int c = ctl.get();

        int rs = runStateOf(c);

        //如果当前线程池停止了，或者 当前task为空且workQueue中没有任务了。这些情况可以直接退出该方法。

        // Check if queue empty only if necessary.

        if (rs >= SHUTDOWN &&

            ! (rs == SHUTDOWN &&

               firstTask == null &&

               ! workQueue.isEmpty()))

            return false;


        for (;;) {

            int wc = workerCountOf(c);

            if (wc >= CAPACITY ||

                wc >= (core ? corePoolSize : maximumPoolSize))

                return false;

            if (compareAndIncrementWorkerCount(c))

                break retry;

            c = ctl.get();  // Re-read ctl

            if (runStateOf(c) != rs)

                continue retry;

            // else CAS failed due to workerCount change; retry inner loop

        }

    }


    boolean workerStarted = false;

    boolean workerAdded = false;

    Worker w = null;

    try {

        w = new Worker(firstTask);

        final Thread t = w.thread;

        if (t != null) {

            final ReentrantLock mainLock = this.mainLock;

            mainLock.lock();

            try {

                // Recheck while holding lock.

                // Back out on ThreadFactory failure or if

                // shut down before lock acquired.

                int rs = runStateOf(ctl.get());


                if (rs < SHUTDOWN ||

                    (rs == SHUTDOWN && firstTask == null)) {

                    if (t.isAlive()) // precheck that t is startable

                        throw new IllegalThreadStateException();

                    workers.add(w);

                    int s = workers.size();

                    if (s > largestPoolSize)

                        largestPoolSize = s;

                    workerAdded = true;

                }

            } finally {

                mainLock.unlock();

            }

            if (workerAdded) {

                t.start();

                workerStarted = true;

            }

        }

    } finally {

        if (! workerStarted)

            addWorkerFailed(w);

    }

    return workerStarted;

}

上面代码，除了线程池状态校验以及保证代码的异步安全，核心就是：

 w = new Worker(firstTask);

 final Thread t = w.thread;

 t.start()；

创建一个Worker并启动其中的线程。

Worker

Worker是什么？Worker是ThreadPoolExecutor的内部类，ThreadPoolExecutor中把excute()传递进来的Runnable认为是一个Work，那么执行Work的就是Worker了。

private final class Worker

    extends AbstractQueuedSynchronizer

    implements Runnable

{


    /** Thread this worker is running in.  Null if factory fails. */

    final Thread thread;

    /** Initial task to run.  Possibly null. */

    Runnable firstTask;

    /** Per-thread task counter */

    volatile long completedTasks;


    /**

     * Creates with given first task and thread from ThreadFactory.

     * @param firstTask the first task (null if none)

     */

    Worker(Runnable firstTask) {

        setState(-1); // inhibit interrupts until runWorker

        this.firstTask = firstTask;

        this.thread = getThreadFactory().newThread(this);

    }


    public void run() {

        runWorker(this);

    }


 }

Worker的构造函数中持有了传入的Task,并通过ThreadFactory创建了个新的线程。

note: 新创建的线程持有了Worker对象自己，而Worker本身又实现了Runnable接口。所以当该线程启动时，run()就会被执行。

看下Worker的run()方法具体干了什么？

首先里面有个while循环，主要是获取task的，如果task一直能获取到，则就能执行到while内部。

1. 
   
2. 当前task非空，即核心线程创建的时候，自带了一个task。会触发task.run()方法，并在它前后分别又beforeExecute(wt, task)和afterExecute(task, thrown)hook点。
   
3. 当前task==null，就需要到getTask()中获取。
   

final void runWorker(Worker w) {

    Thread wt = Thread.currentThread();

    Runnable task = w.firstTask;

    w.firstTask = null;

    w.unlock(); // allow interrupts

    boolean completedAbruptly = true;

    try {

        while (task != null || (task = getTask()) != null) {

            w.lock();

            // If pool is stopping, ensure thread is interrupted;

            // if not, ensure thread is not interrupted.  This

            // requires a recheck in second case to deal with

            // shutdownNow race while clearing interrupt

            if ((runStateAtLeast(ctl.get(), STOP) ||

                 (Thread.interrupted() &&

                  runStateAtLeast(ctl.get(), STOP))) &&

                !wt.isInterrupted())

                wt.interrupt();

            try {

                beforeExecute(wt, task);

                Throwable thrown = null;

                try {

                    task.run();

                } catch (RuntimeException x) {

                    thrown = x; throw x;

                } catch (Error x) {

                    thrown = x; throw x;

                } catch (Throwable x) {

                    thrown = x; throw new Error(x);

                } finally {

                    afterExecute(task, thrown);

                }

            } finally {

                task = null;

                w.completedTasks++;

                w.unlock();

            }

        }

        completedAbruptly = false;

    } finally {

        processWorkerExit(w, completedAbruptly);

    }

}

getTask()就跟ThreadPoolExecutor构造函数中的BlockQueue关联上了。当前线程如果执行完第一个Task后，应该怎么办呢？如果还不满足结束条件(比如说存活时间超过keepAlive*unit)，则会向worksQueue（BlockQueue）中所要任务，然后执行。

private Runnable getTask() {

    boolean timedOut = false; // Did the last poll() time out?


    for (;;) {

        int c = ctl.get();

        int rs = runStateOf(c);


        // Check if queue empty only if necessary.

        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {

            decrementWorkerCount();

            return null;

        }


        int wc = workerCountOf(c);


        // Are workers subject to culling?

        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;


        if ((wc > maximumPoolSize || (timed && timedOut))

            && (wc > 1 || workQueue.isEmpty())) {

            if (compareAndDecrementWorkerCount(c))

                return null;

            continue;

        }


        try {

            Runnable r = timed ?

                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :

                workQueue.take();

            if (r != null)

                return r;

            timedOut = true;

        } catch (InterruptedException retry) {

            timedOut = false;

        }

    }

}

上面前段代码就是判断当前线程池是否还能继续存在，如果能存在并且未超时，那么就从workQueue中取task()。Runnable r = timed ?workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS):workQueue.take();

/**

 * Retrieves and removes the head of this queue, waiting if necessary

 * until an element becomes available.

 *

 * @return the head of this queue

 * @throws InterruptedException if interrupted while waiting

 */

E take() throws InterruptedException;

是个阻塞的行为。所以TheadPoolExecutor中的workQueue参数传入哪种类型的BlockQueue直接影响了未执行任务的执行顺序。 不过需要注意的是即使使用了优先级队列，高优先级的任务也不一定比低优先级任务先执行，因为任务是由线程发起的，workQueue没法影响线程的执行顺序。