深入了解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;//29
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;//约5亿
// 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;

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

<code data-type="codeline">11:07:04.377 [pool-1-thread-2] DEBUG ayue - 线程名称：pool-1-thread-2,正在执行task：3</code><code data-type="codeline">11:07:04.377 [pool-1-thread-1] DEBUG ayue - 线程名称：pool-1-thread-1,正在执行task：1</code><code data-type="codeline">11:07:05.384 [pool-1-thread-2] DEBUG ayue - task3执行完毕</code><code data-type="codeline">11:07:05.384 [pool-1-thread-1] DEBUG ayue - task1执行完毕</code><code data-type="codeline">11:07:05.384 [pool-1-thread-2] DEBUG ayue - 线程名称：pool-1-thread-2,正在执行task：2</code><code data-type="codeline">11:07:06.397 [pool-1-thread-2] DEBUG ayue - task2执行完毕</code>

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

<code data-type="codeline">14:08:07.166 [javatv-1] DEBUG ayue - 线程名称：javatv-1,正在执行task：1</code><code data-type="codeline">14:08:07.166 [javatv-2] DEBUG ayue - 线程名称：javatv-2,正在执行task：3</code><code data-type="codeline">14:08:08.170 [javatv-1] DEBUG ayue - task1执行完毕</code><code data-type="codeline">14:08:08.170 [javatv-2] DEBUG ayue - task3执行完毕</code><code data-type="codeline">14:08:08.170 [javatv-1] DEBUG ayue - 线程名称：javatv-1,正在执行task：2</code><code data-type="codeline">14:08:09.172 [javatv-1] DEBUG ayue - task2执行完毕</code>

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