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++) {
            //執(zhí)行任務(wù)
            executor.execute(new MyTask(i));
        }
    }


    //任務(wù)
    static class MyTask implements Runnable {

        private int taskNum;

        public MyTask(int num) {
            this.taskNum = num;
        }

        @Override
        public void run() {
            log.debug("線程名稱：{},正在執(zhí)行task：{}", Thread.currentThread().getName(), taskNum);
            try {
                //模擬其他操作
                Thread.currentThread().sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            log.debug("task{}執(zhí)行完畢", taskNum);
        }
    }
}

<code data-type="codeline">11:07:04.377 [pool-1-thread-2] DEBUG ayue - 線程名稱：pool-1-thread-2,正在執(zhí)行task：3</code><code data-type="codeline">11:07:04.377 [pool-1-thread-1] DEBUG ayue - 線程名稱：pool-1-thread-1,正在執(zhí)行task：1</code><code data-type="codeline">11:07:05.384 [pool-1-thread-2] DEBUG ayue - task3執(zhí)行完畢</code><code data-type="codeline">11:07:05.384 [pool-1-thread-1] DEBUG ayue - task1執(zhí)行完畢</code><code data-type="codeline">11:07:05.384 [pool-1-thread-2] DEBUG ayue - 線程名稱：pool-1-thread-2,正在執(zhí)行task：2</code><code data-type="codeline">11:07:06.397 [pool-1-thread-2] DEBUG ayue - task2執(zhí)行完畢</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("線程名稱：{},正在執(zhí)行task：{}", Thread.currentThread().getName(), taskNum);
            try {
                //模擬其他操作
                Thread.currentThread().sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            log.debug("task{}執(zhí)行完畢", taskNum);
        }
    }
}

<code data-type="codeline">14:08:07.166 [javatv-1] DEBUG ayue - 線程名稱：javatv-1,正在執(zhí)行task：1</code><code data-type="codeline">14:08:07.166 [javatv-2] DEBUG ayue - 線程名稱：javatv-2,正在執(zhí)行task：3</code><code data-type="codeline">14:08:08.170 [javatv-1] DEBUG ayue - task1執(zhí)行完畢</code><code data-type="codeline">14:08:08.170 [javatv-2] DEBUG ayue - task3執(zhí)行完畢</code><code data-type="codeline">14:08:08.170 [javatv-1] DEBUG ayue - 線程名稱：javatv-1,正在執(zhí)行task：2</code><code data-type="codeline">14:08:09.172 [javatv-1] DEBUG ayue - task2執(zhí)行完畢</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) {
    //如果沒有任務(wù)直接拋出異常
    if (command == null)
        throw new NullPointerException();
    //獲取當(dāng)前線程的狀態(tài)+線程個(gè)數(shù)
    int c = ctl.get();
    /**
     * workerCountOf,線程池當(dāng)前線程數(shù),并判斷是否小于核心線程數(shù)
     */
    if (workerCountOf(c) < corePoolSize) {//如果小于
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    if (isRunning(c) && workQueue.offer(command)) {
        // 這里是向任務(wù)隊(duì)列投放任務(wù)成功，對(duì)線程池的運(yùn)行中狀態(tài)做二次檢查
        // 如果線程池二次檢查狀態(tài)是非運(yùn)行中狀態(tài)，則從任務(wù)隊(duì)列移除當(dāng)前的任務(wù)調(diào)用拒絕策略處理（也就是移除前面成功入隊(duì)的任務(wù)實(shí)例）
        int recheck = ctl.get();
        if (! isRunning(recheck) && remove(command))
            reject(command);
        /* 走到下面的else if分支，說明有以下的前提：
         * 1、待執(zhí)行的任務(wù)已經(jīng)成功加入任務(wù)隊(duì)列
         * 2、線程池可能是RUNNING狀態(tài)
         * 3、傳入的任務(wù)可能從任務(wù)隊(duì)列中移除失?。ㄒ瞥〉奈ㄒ豢赡芫褪侨蝿?wù)已經(jīng)被執(zhí)行了）
         *
         * 如果當(dāng)前工作線程數(shù)量為0，則創(chuàng)建一個(gè)非核心線程并且傳入的任務(wù)對(duì)象為null - 返回
         * 也就是創(chuàng)建的非核心線程不會(huì)馬上運(yùn)行，而是等待獲取任務(wù)隊(duì)列的任務(wù)去執(zhí)行 
         * 如果前工作線程數(shù)量不為0，原來應(yīng)該是最后的else分支，但是可以什么也不做，
         * 因?yàn)槿蝿?wù)已經(jīng)成功入隊(duì)列，總會(huì)有合適的時(shí)機(jī)分配其他空閑線程去執(zhí)行它。
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    /* 走到這里說明有以下的前提：
     * 1、線程池中的工作線程總數(shù)已經(jīng)大于等于corePoolSize（簡單來說就是核心線程已經(jīng)全部懶創(chuàng)建完畢）
     * 2、線程池可能不是RUNNING狀態(tài)
     * 3、線程池可能是RUNNING狀態(tài)同時(shí)任務(wù)隊(duì)列已經(jīng)滿了
     *
     * 如果向任務(wù)隊(duì)列投放任務(wù)失敗，則會(huì)嘗試創(chuàng)建非核心線程傳入任務(wù)執(zhí)行
     * 創(chuàng)建非核心線程失敗，此時(shí)需要拒絕執(zhí)行任務(wù)
     */
    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:
    //外層循環(huán)：判斷線程池狀態(tài)
    for (;;) {
        int c = ctl.get();
        //獲取線程池狀態(tài)
        int rs = runStateOf(c);
        // 檢查線程池的狀態(tài)是否存活.
        if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty()))
            return false;
        //內(nèi)層循環(huán)：線程池添加核心線程并返回是否添加成功的結(jié)果
        for (;;) {
            //線程數(shù)量
            int wc = workerCountOf(c);
            //線程數(shù)量超過容量，返回false
            if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            //CAS增加線程數(shù)量，若成功跳出外層循環(huán)
            if (compareAndIncrementWorkerCount(c))
                break retry;
            //否則失敗，并更新c
            c = ctl.get();  // Re-read ctl
            //如果這時(shí)的線程池狀態(tài)發(fā)生變化，重新對(duì)外層循環(huán)進(jìn)行自旋
            if (runStateOf(c) != rs)
                continue retry;
            // else CAS failed due to workerCount change; retry inner loop
        }
    }
    //如果CAS成功了，則繼續(xù)往下走
    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        //創(chuàng)建一個(gè)Worker，這個(gè)Worker實(shí)現(xiàn)了Runable，把它看成一個(gè)任務(wù)單元
        w = new Worker(firstTask);
        //這個(gè)Thread就是當(dāng)前的任務(wù)單元Worker，即this
        final Thread t = w.thread;
        if (t != null) {
            //加鎖，因?yàn)榭赡苡卸鄠€(gè)線程來調(diào)用
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                // 再次檢查線程池的狀態(tài)，避免在獲取鎖前調(diào)用shutdown方法
                int rs = runStateOf(ctl.get());
                if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) {
                    //如果t線程已經(jīng)啟動(dòng)尚未終止，則拋出異常
                    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();
            }
            //加入線程池后，啟動(dòng)該線程，上面已經(jīng)設(shè)置為true
            if (workerAdded) {
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        //如果線程啟動(dòng)失敗，則調(diào)用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;
    
    //構(gòu)造方法
    Worker(Runnable firstTask) {
        //在調(diào)用runWorker前禁止中斷
        //當(dāng)其它線程調(diào)用了線程池的 shutdownNow 時(shí)候，如果 worker 狀態(tài) >= 0 則會(huì)中斷該線程
        //具體方法在 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;
    // 調(diào)用unlock()是為了讓外部可以中斷
    w.unlock(); // allow interrupts
    // 線程退出的原因，true是任務(wù)導(dǎo)致，false是線程正常退出
    boolean completedAbruptly = true;
    try {
        // 1. 如果firstTask不為null，則執(zhí)行firstTask
         // 2. 如果firstTask為null，則調(diào)用getTask()從隊(duì)列獲取任務(wù)
        // 3. 阻塞隊(duì)列的特性就是：當(dāng)隊(duì)列為空時(shí)，當(dāng)前線程會(huì)被阻塞等待
        while (task != null || (task = getTask()) != null) {
            w.lock();
            // 判斷線程池的狀態(tài)，如果線程池正在停止，則對(duì)當(dāng)前線程進(jìn)行中斷操作
            if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() &&
                  runStateAtLeast(ctl.get(), STOP))) && !wt.isInterrupted())
                wt.interrupt();//中斷
            try {
                //該方法里面沒有內(nèi)容，可以自己擴(kuò)展實(shí)現(xiàn)，比如上面提到的線程池的監(jiān)控
                beforeExecute(wt, task);
                Throwable thrown = null;
                try {
                    //執(zhí)行具體的任務(wù)
                    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
                //統(tǒng)計(jì)當(dāng)前worker完成了多少個(gè)任務(wù)
                w.completedTasks++;
                //釋放鎖
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        // 處理線程退出，completedAbruptly為true說明由于任務(wù)異常導(dǎo)致線程非正常退出
        processWorkerExit(w, completedAbruptly);
    }
}

private Runnable getTask() {
    //判斷最新一次的poll是否超時(shí)
    //poll：取走BlockingQueue里排在首位的對(duì)象
    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：線程池狀態(tài)SHUTDOWN、STOP、TERMINATED狀態(tài)
         * 條件2：線程池STOP、TERMINATED狀態(tài)或workQueue為空
         * 條件1與條件2同時(shí)為true，則workerCount-1，并且返回null
         * 注：條件2是考慮到SHUTDOWN狀態(tài)的線程池不會(huì)接受任務(wù)，但仍會(huì)處理任務(wù)（前面也講到了）
         */
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            decrementWorkerCount();
            return null;
        }
        int wc = workerCountOf(c);
        // Are workers subject to culling?
        /*
         * 該屬性的作用是判斷當(dāng)前線程是否允許超時(shí)：
         * 1.allowCoreThreadTimeOut
         *   如果為 false（默認(rèn)），核心線程即使在空閑時(shí)也保持活動(dòng)狀態(tài)。
         *   如果為 true，則核心線程使用 keepAliveTime 超時(shí)等待工作。
         * 2.wc > corePoolSize
         *   當(dāng)前線程是否已經(jīng)超過核心線程數(shù)量。
         */
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
        /*
         * 判斷當(dāng)前線程是否可以退出：
         * 1.wc > maximumPoolSize || (timed && timedOut)
         *   wc > maximumPoolSize = true，說明當(dāng)前的工作線程總數(shù)大于線程池最大線程數(shù)。
         *   timed && timedOut = true，說明當(dāng)前線程允許超時(shí)并且已經(jīng)超時(shí)。
         * 2.wc > 1 || workQueue.isEmpty()
         *   工作線程總數(shù)大于1或者任務(wù)隊(duì)列為空，則通過CAS把線程數(shù)減去1，同時(shí)返回null
         */
        if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) {
            if (compareAndDecrementWorkerCount(c))
                return null;
            continue;
        }
        try {
            /*
       * 1.poll(long timeout, TimeUnit unit)：從BlockingQueue取出一個(gè)隊(duì)首的對(duì)象，
       * 如果在指定時(shí)間內(nèi)，隊(duì)列一旦有數(shù)據(jù)可取，則立即返回隊(duì)列中的數(shù)據(jù)。否則直到時(shí)間超時(shí)還沒有數(shù)據(jù)可取，返回失敗。
       *
       * 2.take():取走BlockingQueue里排在首位的對(duì)象,若BlockingQueue為空,阻斷進(jìn)入等待狀態(tài)直到BlockingQueue有新的數(shù)據(jù)被加入。     
       *
       *
       * 如果timed為true，通過poll()方法做超時(shí)拉取，keepAliveTime時(shí)間內(nèi)沒有等待到有效的任務(wù)，則返回null。
       *
             * 如果timed為false，通過take()做阻塞拉取，會(huì)阻塞到有下一個(gè)有效的任務(wù)時(shí)候再返回（一般不會(huì)是null）。
       */
            Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
            if (r != null)
                return r;
            //通過poll()方法從任務(wù)隊(duì)列中拉取任務(wù)為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) {
    // 因?yàn)閽伋鲇脩舢惓?dǎo)致線程終結(jié)，直接使工作線程數(shù)減1即可
    // 如果沒有任何異常拋出的情況下是通過getTask()返回null引導(dǎo)線程正常跳出runWorker()方法的while死循環(huán)從而正常終結(jié)，這種情況下，在getTask()中已經(jīng)把線程數(shù)減1
    if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
        decrementWorkerCount();

    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        // 全局的已完成任務(wù)記錄數(shù)加上此將要終結(jié)的Worker中的已完成任務(wù)數(shù)
        completedTaskCount += w.completedTasks;
        // 工作線程集合中移除此將要終結(jié)的Worker
        workers.remove(w);
    } finally {
        mainLock.unlock();
    }
     
    // 見下一小節(jié)分析，用于根據(jù)當(dāng)前線程池的狀態(tài)判斷是否需要進(jìn)行線程池terminate處理
    tryTerminate();

    int c = ctl.get();
    // 如果線程池的狀態(tài)小于STOP，也就是處于RUNNING或者SHUTDOWN狀態(tài)的前提下：
    // 1.如果線程不是由于拋出用戶異常終結(jié)，如果允許核心線程超時(shí)，則保持線程池中至少存在一個(gè)工作線程
    // 2.如果線程由于拋出用戶異常終結(jié)，或者當(dāng)前工作線程數(shù)，那么直接添加一個(gè)新的非核心線程
    if (runStateLessThan(c, STOP)) {
        if (!completedAbruptly) {
            // 如果允許核心線程超時(shí)，最小值為0，否則為corePoolSize
            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
            // 如果最小值為0，同時(shí)任務(wù)隊(duì)列不空，則更新最小值為1
            if (min == 0 && ! workQueue.isEmpty())
                min = 1;
            // 工作線程數(shù)大于等于最小值，直接返回不新增非核心線程
            if (workerCountOf(c) >= min)
                return; // replacement not needed
        }
        addWorker(null, false);
    }
}