深入 Java 线程池：从设计思想到源码解读

public class ThreadPoolTest {
    static CountDownLatch latch = new CountDownLatch(100000);    static ExecutorService es = Executors.newFixedThreadPool(4);
    public static void main(String[] args) throws InterruptedException {
        long timeStart = System.currentTimeMillis();
        for (int i = 0; i < 100000; i++) {            newThread();            //executors();        }        latch.await();        System.out.println(System.currentTimeMillis() - timeStart);        es.shutdown();    }
    /**     * 使用线程池     */    public static void executors() {        es.submit(() -> {            latch.countDown();        });    }
    /**     * 直接new     */    public static void newThread() {        new Thread(() -> {            latch.countDown();        }).start();    }}

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

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));private static final int COUNT_BITS = Integer.SIZE - 3;//29private static final int CAPACITY   = (1 << COUNT_BITS) - 1;//约5亿// runState is stored in the high-order bitsprivate 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;

@Slf4j(topic = "ayue")public class ThreadExecutorPoolTest1 {
    public static void main(String[] args) {        ThreadPoolExecutor executor =             new ThreadPoolExecutor(1, 2, 3, TimeUnit.SECONDS, new ArrayBlockingQueue<>(1));        for (int i = 1; i < 4; i++) {            //执行任务            executor.execute(new MyTask(i));        }    }

    //任务    static class MyTask implements Runnable {
        private int taskNum;
        public MyTask(int num) {            this.taskNum = num;        }
        @Override        public void run() {            log.debug("线程名称：{},正在执行task：{}", Thread.currentThread().getName(), taskNum);            try {                //模拟其他操作                Thread.currentThread().sleep(1000);            } catch (InterruptedException e) {                e.printStackTrace();            }            log.debug("task{}执行完毕", taskNum);        }    }}

11:07:04.377 [pool-1-thread-2] DEBUG ayue - 线程名称：pool-1-thread-2,正在执行task：311:07:04.377 [pool-1-thread-1] DEBUG ayue - 线程名称：pool-1-thread-1,正在执行task：111:07:05.384 [pool-1-thread-2] DEBUG ayue - task3执行完毕11:07:05.384 [pool-1-thread-1] DEBUG ayue - task1执行完毕11:07:05.384 [pool-1-thread-2] DEBUG ayue - 线程名称：pool-1-thread-2,正在执行task：211:07:06.397 [pool-1-thread-2] DEBUG ayue - task2执行完毕

static class DefaultThreadFactory implements ThreadFactory {    private static final AtomicInteger poolNumber = new AtomicInteger(1);    private final ThreadGroup group;    private final AtomicInteger threadNumber = new AtomicInteger(1);    private final String namePrefix;    DefaultThreadFactory() {        SecurityManager s = System.getSecurityManager();        group = (s != null) ? s.getThreadGroup() :                              Thread.currentThread().getThreadGroup();        namePrefix = "pool-" + poolNumber.getAndIncrement() + "-thread-";    }    public Thread newThread(Runnable r) {        //线程名：namePrefix + threadNumber.getAndIncrement()        Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(),0);        if (t.isDaemon())            t.setDaemon(false);        if (t.getPriority() != Thread.NORM_PRIORITY)            t.setPriority(Thread.NORM_PRIORITY);        return t;    }}

@Slf4j(topic = "ayue")public class ThreadExecutorPoolTest1 {
    public static void main(String[] args) {        //自增线程id        AtomicInteger threadNumber = new AtomicInteger(1);        ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 2, 3, TimeUnit.SECONDS, new ArrayBlockingQueue<>(1), new ThreadFactory() {            @Override            public Thread newThread(Runnable r) {                return new Thread(r, "javatv-" + threadNumber.getAndIncrement());            }        });        for (int i = 1; i < 4; i++) {            executor.execute(new MyTask(i));        }    }

    static class MyTask implements Runnable {
        private int taskNum;
        public MyTask(int num) {            this.taskNum = num;        }
        @Override        public void run() {            log.debug("线程名称：{},正在执行task：{}", Thread.currentThread().getName(), taskNum);            try {                //模拟其他操作                Thread.currentThread().sleep(1000);            } catch (InterruptedException e) {                e.printStackTrace();            }            log.debug("task{}执行完毕", taskNum);        }    }}

14:08:07.166 [javatv-1] DEBUG ayue - 线程名称：javatv-1,正在执行task：114:08:07.166 [javatv-2] DEBUG ayue - 线程名称：javatv-2,正在执行task：314:08:08.170 [javatv-1] DEBUG ayue - task1执行完毕14:08:08.170 [javatv-2] DEBUG ayue - task3执行完毕14:08:08.170 [javatv-1] DEBUG ayue - 线程名称：javatv-1,正在执行task：214:08:09.172 [javatv-1] DEBUG ayue - task2执行完毕

for (int i = 1; i < 5; i++) {    executor.execute(new MyTask(i));}

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

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

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

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

public void execute(Runnable command) {    //如果没有任务直接抛出异常    if (command == null)        throw new NullPointerException();    //获取当前线程的状态+线程个数    int c = ctl.get();    /**     * workerCountOf,线程池当前线程数,并判断是否小于核心线程数     */    if (workerCountOf(c) < corePoolSize) {//如果小于        if (addWorker(command, true))            return;        c = ctl.get();    }    if (isRunning(c) && workQueue.offer(command)) {        // 这里是向任务队列投放任务成功，对线程池的运行中状态做二次检查        // 如果线程池二次检查状态是非运行中状态，则从任务队列移除当前的任务调用拒绝策略处理（也就是移除前面成功入队的任务实例）        int recheck = ctl.get();        if (! isRunning(recheck) && remove(command))            reject(command);        /* 走到下面的else if分支，说明有以下的前提：         * 1、待执行的任务已经成功加入任务队列         * 2、线程池可能是RUNNING状态         * 3、传入的任务可能从任务队列中移除失败（移除失败的唯一可能就是任务已经被执行了）         *         * 如果当前工作线程数量为0，则创建一个非核心线程并且传入的任务对象为null - 返回         * 也就是创建的非核心线程不会马上运行，而是等待获取任务队列的任务去执行          * 如果前工作线程数量不为0，原来应该是最后的else分支，但是可以什么也不做，         * 因为任务已经成功入队列，总会有合适的时机分配其他空闲线程去执行它。        else if (workerCountOf(recheck) == 0)            addWorker(null, false);    }    /* 走到这里说明有以下的前提：     * 1、线程池中的工作线程总数已经大于等于corePoolSize（简单来说就是核心线程已经全部懒创建完毕）     * 2、线程池可能不是RUNNING状态     * 3、线程池可能是RUNNING状态同时任务队列已经满了     *     * 如果向任务队列投放任务失败，则会尝试创建非核心线程传入任务执行     * 创建非核心线程失败，此时需要拒绝执行任务     */    else if (!addWorker(command, false))        reject(command);}

if (workerCountOf(c) < corePoolSize) {    if (addWorker(command, true))        return;    c = ctl.get();}

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 (;;) {            //线程数量            int wc = workerCountOf(c);            //线程数量超过容量，返回false            if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize))                return false;            //CAS增加线程数量，若成功跳出外层循环            if (compareAndIncrementWorkerCount(c))                break retry;            //否则失败，并更新c            c = ctl.get();  // Re-read ctl            //如果这时的线程池状态发生变化，重新对外层循环进行自旋            if (runStateOf(c) != rs)                continue retry;            // else CAS failed due to workerCount change; retry inner loop        }    }    //如果CAS成功了，则继续往下走    boolean workerStarted = false;    boolean workerAdded = false;    Worker w = null;    try {        //创建一个Worker，这个Worker实现了Runable，把它看成一个任务单元        w = new Worker(firstTask);        //这个Thread就是当前的任务单元Worker，即this        final Thread t = w.thread;        if (t != null) {            //加锁，因为可能有多个线程来调用            final ReentrantLock mainLock = this.mainLock;            mainLock.lock();            try {                // 再次检查线程池的状态，避免在获取锁前调用shutdown方法                int rs = runStateOf(ctl.get());                if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) {                    //如果t线程已经启动尚未终止，则抛出异常                    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();            }            //加入线程池后，启动该线程，上面已经设置为true            if (workerAdded) {                t.start();                workerStarted = true;            }        }    } finally {        //如果线程启动失败，则调用addWorkerFailed，回滚操作        if (! workerStarted)            addWorkerFailed(w);    }    return workerStarted;}

private final class Worker extends AbstractQueuedSynchronizer implements Runnable{        final Thread thread;    /** Initial task to run.  Possibly null. */    Runnable firstTask;        //构造方法    Worker(Runnable firstTask) {        //在调用runWorker前禁止中断        //当其它线程调用了线程池的 shutdownNow 时候，如果 worker 状态 >= 0 则会中断该线程        //具体方法在 interruptIfStarted() 中可以看到        setState(-1); // inhibit interrupts until runWorker        this.firstTask = firstTask;        this.thread = getThreadFactory().newThread(this);  }        /** Delegates main run loop to outer runWorker  */    public void run() {        runWorker(this);    }    //省略其他代码...}

final void runWorker(Worker w) {    Thread wt = Thread.currentThread();    Runnable task = w.firstTask;    w.firstTask = null;    // 调用unlock()是为了让外部可以中断    w.unlock(); // allow interrupts    // 线程退出的原因，true是任务导致，false是线程正常退出    boolean completedAbruptly = true;    try {        // 1. 如果firstTask不为null，则执行firstTask         // 2. 如果firstTask为null，则调用getTask()从队列获取任务        // 3. 阻塞队列的特性就是：当队列为空时，当前线程会被阻塞等待        while (task != null || (task = getTask()) != null) {            w.lock();            // 判断线程池的状态，如果线程池正在停止，则对当前线程进行中断操作            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 {                    //同 beforeExecute()                    afterExecute(task, thrown);                }            } finally {                task = null;//help gc                //统计当前worker完成了多少个任务                w.completedTasks++;                //释放锁                w.unlock();            }        }        completedAbruptly = false;    } finally {        // 处理线程退出，completedAbruptly为true说明由于任务异常导致线程非正常退出        processWorkerExit(w, completedAbruptly);    }}

private Runnable getTask() {    //判断最新一次的poll是否超时    //poll：取走BlockingQueue里排在首位的对象    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.        /**         * 条件1：线程池状态SHUTDOWN、STOP、TERMINATED状态         * 条件2：线程池STOP、TERMINATED状态或workQueue为空         * 条件1与条件2同时为true，则workerCount-1，并且返回null         * 注：条件2是考虑到SHUTDOWN状态的线程池不会接受任务，但仍会处理任务（前面也讲到了）         */        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {            decrementWorkerCount();            return null;        }        int wc = workerCountOf(c);        // Are workers subject to culling?        /*         * 该属性的作用是判断当前线程是否允许超时：         * 1.allowCoreThreadTimeOut         *   如果为 false（默认），核心线程即使在空闲时也保持活动状态。         *   如果为 true，则核心线程使用 keepAliveTime 超时等待工作。         * 2.wc > corePoolSize         *   当前线程是否已经超过核心线程数量。         */        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;        /*         * 判断当前线程是否可以退出：         * 1.wc > maximumPoolSize || (timed && timedOut)         *   wc > maximumPoolSize = true，说明当前的工作线程总数大于线程池最大线程数。         *   timed && timedOut = true，说明当前线程允许超时并且已经超时。         * 2.wc > 1 || workQueue.isEmpty()         *   工作线程总数大于1或者任务队列为空，则通过CAS把线程数减去1，同时返回null         */        if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) {            if (compareAndDecrementWorkerCount(c))                return null;            continue;        }        try {            /*       * 1.poll(long timeout, TimeUnit unit)：从BlockingQueue取出一个队首的对象，       * 如果在指定时间内，队列一旦有数据可取，则立即返回队列中的数据。否则直到时间超时还没有数据可取，返回失败。       *       * 2.take():取走BlockingQueue里排在首位的对象,若BlockingQueue为空,阻断进入等待状态直到BlockingQueue有新的数据被加入。            *       *       * 如果timed为true，通过poll()方法做超时拉取，keepAliveTime时间内没有等待到有效的任务，则返回null。       *             * 如果timed为false，通过take()做阻塞拉取，会阻塞到有下一个有效的任务时候再返回（一般不会是null）。       */            Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :                workQueue.take();            if (r != null)                return r;            //通过poll()方法从任务队列中拉取任务为null            timedOut = true;        } catch (InterruptedException retry) {            timedOut = false;        }    }}

boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) {    if (compareAndDecrementWorkerCount(c))        return null;    continue;}

if (isRunning(c) && workQueue.offer(command)) {    int recheck = ctl.get();    if (! isRunning(recheck) && remove(command))        reject(command);    else if (workerCountOf(recheck) == 0)        addWorker(null, false);}

private void processWorkerExit(Worker w, boolean completedAbruptly) {    // 因为抛出用户异常导致线程终结，直接使工作线程数减1即可    // 如果没有任何异常抛出的情况下是通过getTask()返回null引导线程正常跳出runWorker()方法的while死循环从而正常终结，这种情况下，在getTask()中已经把线程数减1    if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted        decrementWorkerCount();
    final ReentrantLock mainLock = this.mainLock;    mainLock.lock();    try {        // 全局的已完成任务记录数加上此将要终结的Worker中的已完成任务数        completedTaskCount += w.completedTasks;        // 工作线程集合中移除此将要终结的Worker        workers.remove(w);    } finally {        mainLock.unlock();    }         // 见下一小节分析，用于根据当前线程池的状态判断是否需要进行线程池terminate处理    tryTerminate();
    int c = ctl.get();    // 如果线程池的状态小于STOP，也就是处于RUNNING或者SHUTDOWN状态的前提下：    // 1.如果线程不是由于抛出用户异常终结，如果允许核心线程超时，则保持线程池中至少存在一个工作线程    // 2.如果线程由于抛出用户异常终结，或者当前工作线程数，那么直接添加一个新的非核心线程    if (runStateLessThan(c, STOP)) {        if (!completedAbruptly) {            // 如果允许核心线程超时，最小值为0，否则为corePoolSize            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;            // 如果最小值为0，同时任务队列不空，则更新最小值为1            if (min == 0 && ! workQueue.isEmpty())                min = 1;            // 工作线程数大于等于最小值，直接返回不新增非核心线程            if (workerCountOf(c) >= min)                return; // replacement not needed        }        addWorker(null, false);    }}