深刻理解线程池原理篇

如今CPU都是有多个核心，并行已经成为事实，一方面咱们但愿最大限度利用机器性能（利用多线程提升吞吐率），另外一方面机器的硬件资源是有限的，咱们也不能无限制的去申请，幸运的是，JDK已经为咱们提供了ExecutorService的实现，还提供了Executors工厂类方便咱们生成模板线程池，可是简单背后必定是复杂，这篇文章就是深刻线程池的源码去一探究竟。

开始以前，须要明确几个概念，方便后面理解线程池的运行原理。

核心线程（corePool）：线程池最终执行任务的角色确定仍是线程，同时咱们也会限制线程的数量，因此咱们能够这样理解核心线程，有新任务提交时，首先检查核心线程数，若是核心线程都在工做，并且数量也已经达到最大核心线程数，则不会继续新建核心线程，而会将任务放入等待队列。

等待队列 (workQueue)：等待队列用于存储当核心线程都在忙时，继续新增的任务，核心线程在执行完当前任务后，也会去等待队列拉取任务继续执行，这个队列通常是一个线程安全的阻塞队列，它的容量也能够由开发者根据业务来定制。

非核心线程：当等待队列满了，若是当前线程数没有超过最大线程数，则会新建线程执行任务，那么核心线程和非核心线程到底有什么区别呢？说出来你可能不信，本质上它们没有什么区别，建立出来的线程也根本没有标识去区分它们是核心仍是非核心的，线程池只会去判断已有的线程数（包括核心和非核心）去跟核心线程数和最大线程数比较，来决定下一步的策略。

线程活动保持时间 (keepAliveTime)：线程空闲下来以后，保持存活的持续时间，超过这个时间尚未任务执行，该工做线程结束。

饱和策略 (RejectedExecutionHandler)：当等待队列已满，线程数也达到最大线程数时，线程池会根据饱和策略来执行后续操做，默认的策略是抛弃要加入的任务。

一图剩千言，上一张图归纳线程池的基本运做流程。

按个人习惯，先提出几个问题，而后带着问题去寻找答案。

1. 线程池的线程是如何作到复用的。
2. 线程池是如何作到高效并发的。
3. 从线程池的设计中，咱们能学到什么？

ThreadPoolExecutor

JDK中线程池的核心实现类是ThreadPoolExecutor，先看这个类的第一个成员变量ctl，AtomicInteger这个类能够经过CAS达到无锁并发，效率比较高，这个变量有双重身份，它的高三位表示线程池的状态，低29位表示线程池中现有的线程数，这也是Doug Lea一个天才的设计，用最少的变量来减小锁竞争，提升并发效率。

//CAS，无锁并发
    private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
    //表示线程池线程数的bit数
    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
    //1110 0000 0000 0000 0000 0000 0000 0000（很耿直的我）
    private static final int RUNNING    = -1 << COUNT_BITS;
    //0000 0000 0000 0000 0000 0000 0000 0000（很耿直的我）
    private static final int SHUTDOWN   =  0 << COUNT_BITS;
    //0010 0000 0000 0000 0000 0000 0000 0000（很耿直的我）
    private static final int STOP       =  1 << COUNT_BITS;
    //0100 0000 0000 0000 0000 0000 0000 0000（很耿直的我）
    private static final int TIDYING    =  2 << COUNT_BITS;
    //0110 0000 0000 0000 0000 0000 0000 0000（很耿直的我）
    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; }
    //组装状态和数量，成为ctl
    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. * 判断状态c是否比s小，下面会给出状态流转图 */ private static boolean runStateLessThan(int c, int s) { return c < s; } //判断状态c是否不小于状态s private static boolean runStateAtLeast(int c, int s) { return c >= s; } //判断线程是否在运行 private static boolean isRunning(int c) { return c < SHUTDOWN; } 复制代码

关于线程池的状态，有5种，

1. RUNNING, 运行状态，值也是最小的，刚建立的线程池就是此状态。
2. SHUTDOWN，停工状态，再也不接收新任务，已经接收的会继续执行
3. STOP，中止状态，再也不接收新任务，已经接收正在执行的，也会中断
4. 清空状态，全部任务都中止了，工做的线程也所有结束了
5. TERMINATED，终止状态，线程池已销毁

它们的流转关系以下：

execute/submit

向线程池提交任务有这2种方式，execute是ExecutorService接口定义的，submit有三种方法重载都在AbstractExecutorService中定义，都是将要执行的任务包装为FutureTask来提交，使用者能够经过FutureTask来拿到任务的执行状态和执行最终的结果，最终调用的都是execute方法，其实对于线程池来讲，它并不关心你是哪一种方式提交的，由于任务的状态是由FutureTask本身维护的，对线程池透明。

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

    public <T> Future<T> submit(Callable<T> task) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<T> ftask = newTaskFor(task);
        execute(ftask);
        return ftask;
    }
复制代码

重点看execute的实现

public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        //第一步，获取ctl
        int c = ctl.get();
        //检查当前线程数是否达到核心线程数的限制，注意线程自己是不区分核心仍是非核心，后面会进一步验证
        if (workerCountOf(c) < corePoolSize) {
            //若是核心线程数未达到，会直接添加一个核心线程，也就是说在线程池刚启动预热阶段，
            //提交任务后，会优先启动核心线程处理
            if (addWorker(command, true))
                return;
            //若是添加任务失败，刷新ctl，进入下一步
            c = ctl.get();
        }
        //检查线程池是不是运行状态，而后将任务添加到等待队列，注意offer是不会阻塞的
        if (isRunning(c) && workQueue.offer(command)) {
           //任务成功添加到等待队列，再次刷新ctl
            int recheck = ctl.get();
           //若是线程池不是运行状态，则将刚添加的任务从队列移除并执行拒绝策略
            if (! isRunning(recheck) && remove(command))
                reject(command);
            //判断当前线程数量，若是线程数量为0，则添加一个非核心线程，而且不指定首次执行任务
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
       //添加非核心线程，指定首次执行任务，若是添加失败，执行异常策略
        else if (!addWorker(command, false))
            reject(command);
    }
    
    /*
     * addWorker方法申明
     * @param core if true use corePoolSize as bound, else
     * maximumPoolSize. (A boolean indicator is used here rather than a
     * value to ensure reads of fresh values after checking other pool
     * state).
     * @return true if successful
     */
    private boolean addWorker(Runnable firstTask, boolean core) {
    //.....
    }
复制代码

这里有2个细节，能够深挖一下。

1. 能够看到execute方法中没有用到重量级锁，ctl虽然能够保证自己变化的原子性，可是不能保证方法内部的代码块的原子性，是否会有并发问题？
2. 上面提到过，addWorker方法能够添加工做线程（核心或者非核心），线程自己没有核心或者非核心的标识，core参数只是用来肯定 当前线程数的比较对象是线程池设置的核心线程数仍是最大线程数，真实状况是否是这样？

addWorker

添加线程的核心方法，直接看源码

private boolean addWorker(Runnable firstTask, boolean core) {
       //至关于goto，虽然不建议滥用，但这里使用又以为没一点问题
        retry:
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);
            //若是线程池的状态到了SHUTDOWN或者之上的状态时候，只有一种状况还须要继续添加线程，
            //那就是线程池已经SHUTDOWN，可是队列中还有任务在排队,并且不接受新任务（因此firstTask必须为null）
           //这里还继续添加线程的初衷是，加快执行等待队列中的任务，尽快让线程池关闭
            // Check if queue empty only if necessary.
            if (rs >= SHUTDOWN &&
                ! (rs == SHUTDOWN &&
                   firstTask == null &&
                   ! workQueue.isEmpty()))
                return false;

            for (;;) {
                int wc = workerCountOf(c);
               //传入的core的参数，惟一用到的地方，若是线程数超过理论最大容量，若是core是true跟最大核心线程数比较，不然跟最大线程数比较
                if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
                //经过CAS自旋，增长线程数+1，增长成功跳出双层循环，继续往下执行
                if (compareAndIncrementWorkerCount(c))
                    break retry;
               //检测当前线程状态若是发生了变化，则继续回到retry，从新开始循环
                c = ctl.get();  // Re-read ctl
                if (runStateOf(c) != rs)
                    continue retry;
                // else CAS failed due to workerCount change; retry inner loop
            }
        }
        //走到这里，说明咱们已经成功的将线程数+1了，可是真正的线程尚未被添加
        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
           //添加线程，Worker是继承了AQS，实现了Runnable接口的包装类
            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());
                    //检查线程状态，仍是跟以前同样，只有当线程池处于RUNNING，或者处于SHUTDOWN而且firstTask==null的时候，这时候建立Worker来加速处理队列中的任务
                    if (rs < SHUTDOWN ||
                        (rs == SHUTDOWN && firstTask == null)) {
                       //线程只能被start一次
                        if (t.isAlive()) // precheck that t is startable
                            throw new IllegalThreadStateException();
                      //workers是一个HashSet，添加咱们新增的Worker
                        workers.add(w);
                        int s = workers.size();
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        workerAdded = true;
                    }
                } finally {
                    mainLock.unlock();
                }
                if (workerAdded) {
                  //启动Worker
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }
复制代码

分析完addWorker的源码实现，咱们能够回答上面留下的二个疑问，

1. execute方法虽然没有加锁，可是在addWorker方法内部，加锁了，这样能够保证不会建立超过咱们预期的线程数，大师在设计的时候，作到了在最小的范围内加锁，尽可能减小锁竞争，
2. 能够看到，core参数，只是用来判断当前线程数是否超量的时候跟corePoolSize仍是maxPoolSize比较，Worker自己无核心或者非核心的概念。 ####继续看Worker是怎么工做的

//Worker的run方法调用的是ThreadPoolExecutor的runWorker方法
    public void run() {
          runWorker(this);
    }


    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) {
               //这个lock在这里是为了若是线程被中断，那么会抛出InterruptedException，而退出循环，结束线程
                w.lock();
                //判断线程是否须要中断
                if ((runStateAtLeast(ctl.get(), STOP) ||
                     (Thread.interrupted() &&
                      runStateAtLeast(ctl.get(), STOP))) &&
                    !wt.isInterrupted())
                    wt.interrupt();
                try {
                   //任务开始执行前的hook方法
                    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 {
                       ////任务开始执行后的hook方法
                        afterExecute(task, thrown);
                    }
                } finally {
                    task = null;
                    w.completedTasks++;
                    w.unlock();
                }
            }
            completedAbruptly = false;
        } finally {
           //Worker退出
            processWorkerExit(w, completedAbruptly);
        }
    }

   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.
           //检查线程池的状态，若是已是STOP及以上的状态，或者已经SHUTDOWN，队列也是空的时候，直接return null，并将Worker数量-1
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
                decrementWorkerCount();
                return null;
            }

            int wc = workerCountOf(c);

           // 注意这里的allowCoreThreadTimeOut参数，字面意思是否容许核心线程超时，即若是咱们设置为false，那么只有当线程数wc大于corePoolSize的时候才会超时
           //更直接的意思就是，若是设置allowCoreThreadTimeOut为false，那么线程池在达到corePoolSize个工做线程以前，不会让闲置的工做线程退出
            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
          //确认超时，将Worker数-1，而后返回
            if ((wc > maximumPoolSize || (timed && timedOut))
                && (wc > 1 || workQueue.isEmpty())) {
                if (compareAndDecrementWorkerCount(c))
                    return null;
                continue;
            }

            try {
                //从队列中取任务，根据timed选择是有时间期限的等待仍是无时间期限的等待
                Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
                if (r != null)
                    return r;
                timedOut = true;
            } catch (InterruptedException retry) {
                timedOut = false;
            }
        }
    }
复制代码

如今咱们能够回答文章一开始提出的三个问题中的前2个了

1. 线程池的线程是如何作到复用的。 线程池中的线程在循环中尝试取任务执行，这一步会被阻塞，若是设置了allowCoreThreadTimeOut为true，则线程池中的全部线程都会在keepAliveTime时间超时后还未取到任务而退出。或者线程池已经STOP，那么全部线程都会被中断，而后退出。
2. 线程池是如何作到高效并发的。 看整个线程池的工做流程，有如下几个须要特别关注的并发点. ①: 线程池状态和工做线程数量的变动。这个由一个AtomicInteger变量 ctl来解决原子性问题。 ②: 向工做Worker容器workers中添加新的Worker的时候。这个线程池自己已经加锁了。 ③: 工做线程Worker从等待队列中取任务的时候。这个由工做队列自己来保证线程安全，好比LinkedBlockingQueue等。

怎么用好Executors

JDK已经给咱们提供了很方便的线程池工厂类Executors, 方便咱们快速建立线程池，可能在阅读源码以前，咱们在面对具体的业务场景时，到底该选择哪一种线程池配置是有疑问的，咱们来看一下.

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

newFixedThreadPool, 能够看到咱们须要传入一个线程数量的参数nThreads，这样线程池的核心线程数和最大线程数都会设成nThreads, 而它的等待队列是一个LinkedBlockingQueue，它的容量限制是Integer.MAX_VALUE, 能够认为是没有边界的。核心线程keepAlive时间0，allowCoreThreadTimeOut默认false。因此这个方法建立的线程池适合能估算出须要多少核心线程数量的场景。

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

newSingleThreadExecutor, 有且只有一个线程在工做，适合任务顺序执行，缺点可是不能充分利用CPU多核性能

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

newCachedThreadPool, 核心线程数0，最大线程数Integer.MAX_VALUE, 线程keepAlive时间60s，用的队列是SynchronousQueue，这种队列自己不会存任务，只作转发，因此newCachedThreadPool适合执行大量的，轻量级任务。

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
        return new ScheduledThreadPoolExecutor(corePoolSize);
    }
复制代码

newScheduledThreadPool, 执行周期性任务，相似定时器。

最后的问题：从线程池的设计中，咱们能学到什么？

以个人我的体会，大概有下面四点

1. 清楚实现原理，能够指导咱们更好的使用。
2. 在写并发程序的时候，尽量的缩小锁的范围，提升代码的吞吐率。
3. goto，不是必定不能用，而不是滥用，有些场景有奇效。
4. 若是你须要多个线程安全的int型变量，考虑利用位运算把它们合并为一个。

全文完，水平有限，有疑问，欢迎交流！