【并发编程】线程池源码分析

谁践踏了优雅 2022-03-22 12:40 407阅读 0赞

# 前言 #

Github：[https://github.com/yihonglei/jdk-source-code-reading][https_github.com_yihonglei_jdk-source-code-reading]（java-concurrent）

# 一 ThreadPoolExecutor 概述 #

ThreadPoolExecutor 是多线程处理工具，可以通过 Executors 工厂方法进行配置。

ThreadPoolExecutor 提供了对线程池管理，避免线程频繁创建和销毁资源消耗，充分利用 CPU，同时也能对系统资源进行控制。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lobF9qeHk_size_16_color_FFFFFF_t_70][]

# 二 ThreadPoolExecutor 重要属性 #

## 1、AtomicInteger ##

AtomicInteger 类型的 ctl 代表了 ThreadPoolExecutor 中的控制状态，它是一个原子整数，

在ThreadPoolExecutor 里面，借助高低位包装了两个概念：workerCount 和 runState 。

### 1.1 workerCount ###

线程池中当前活动的线程数量，它占据 ctl 的低 29 位，这样，每当活跃线程数增加或减少时，ctl 直接做相应数目的增减即可。

而 ThreadPoolExecutor 中 COUNT\_BITS 就代表了 workerCount 所占位数，定义如下：

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

在 Java 中，一个 int 占据 32 位，而 COUNT\_BITS 的结果不言而喻，Integer 大小 32 减去 3，就是29；

另外，既然 workerCount 代表了线程池中当前活动的线程数量，那么它有个上下限阈值，下限是0，上限呢？

ThreadPoolExecutor 中 CAPACITY 就代表了 workerCount 的上限，它是ThreadPoolExecutor中理论上的最大活跃线程数，

其定义如下：

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

运算过程为 1 左移 29 位，也就是00000000 00000000 00000000 00000001 --> 001 0000 00000000 00000000 00000000，

再减去 1 的话，就是 000 11111 11111111 11111111 11111111，前三位代表线程池运行状态 runState，所以这里 workerCount 的

理论最大值就应该是 29 个 1，即 536870911；既然 workerCount 作为其中一个概念复合在 AtomicInteger ctl 中，

那么 ThreadPoolExecutor 理应提供从 AtomicInteger ctl 中解析出 workerCount 的方法，如下：

private static int workerCountOf(int c) { return c & CAPACITY; }

计算逻辑简单，传入的 c 代表的是 ctl 的值，即高 3 位为线程池运行状态 runState，低 29 位为线程池中当前活动的线程数量

workerCount，将其与 CAPACITY 进行与操作 &，也就是与 000 11111 11111111 11111111 11111111 进行与操作，

c 的前三位通过与 000 进行与操作，无论 c 前三位为何值，最终都会变成 000，也就是舍弃前三位的值，

而 c 的低 29 位与 29 个 1 进行与操作，c 的低 29 位还是会保持原值，这样就从 AtomicInteger ctl 中解析出了 workerCount 的值。

### 1.2 runState ###

线程池运行状态，占据ctl的高3位，有 RUNNING、SHUTDOWN、STOP、TIDYING、TERMINATED五种状态。

AtomicInteger ctl的定义如下：

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

线程池运行状态，它占据ctl的高3位，有RUNNING、SHUTDOWN、STOP、TIDYING、TERMINATED五种状态。

**1）RUNNING**

接受新任务，并处理队列任务。

private static final int RUNNING = -1 << COUNT_BITS;

\-1 在 Java 底层是由 32 个1表示的，左移 29 位的话，即 111 00000 00000000 00000000 00000000，也就是低 29 位全部为 0，

高 3 位全部为 1 的话，表示 RUNNING 状态，即 -536870912；

**2）SHUTDOWN**

不接受新任务，但会处理队列任务。

private static final int SHUTDOWN = 0 << COUNT_BITS;

0 在 Java 底层是由 32 个 0 表示的，无论左移多少位，还是 32 个 0，即 000 00000 00000000 00000000 00000000，

也就是低 29 位全部为 0，高 3 位全部为 0 的话，表示 SHUTDOWN 状态，即 0；

**3）STOP**

不接受新任务，不会处理队列任务，而且会中断正在处理过程中的任务。

private static final int STOP = 1 << COUNT_BITS;

1 在 Java 底层是由前面的 31 个 0 和 1 个 1 组成的，左移 29 位的话，即 001 00000 00000000 00000000 00000000，

也就是低 29 位全部为 0，高 3 位为 001 的话，表示 STOP 状态，即536870912；

**4）TIDYING**

所有的任务已结束，workerCount 为 0，线程过渡到 TIDYING 状态，将会执行 terminated() 钩子方法。

private static final int TIDYING = 2 << COUNT_BITS;

2 在 Java 底层是由前面的 30 个 0 和 1 个 10 组成的，左移 29 位的话，即 010 00000 00000000 00000000 00000000，

也就是低 29 位全部为 0，高 3 位为 010 的话，表示 TIDYING 状态，即 1073741824；

**5）TERMINATED**

terminated() 方法已经完成。

private static final int TERMINATED = 3 << COUNT_BITS;

2 在 Java 底层是由前面的 30 个 0 和 1 个 11 组成的，左移 29 位的话，即 011 00000 00000000 00000000 00000000，

也就是低 29 位全部为 0，高 3 位为 011 的话，表示 TERMINATED 状态，即 1610612736；

由上面我们可以得知，运行状态的值按照 RUNNING --> SHUTDOWN --> STOP --> TIDYING --> TERMINATED 顺序值是递增的，

这些值之间的数值顺序很重要。随着时间的推移，运行状态单调增加，但是不需要经过每个状态。那么，可能存在的线程池状态的

转换是什么呢？

**1）RUNNING -> SHUTDOWN**

调用 shutdownNow() 方法后，或者线程池实现了 finalize() 方法，在里面调用了 shutdown() 方法，即隐式调用；

**2）(RUNNING or SHUTDOWN) -> STOP**

调用 shutdownNow() 方法后；

**3）SHUTDOWN -> TIDYING**

线程池和队列均为空时；

**4）STOP -> TIDYING**

线程池为空时；

**5）TIDYING -> TERMINATED**

terminated() 钩子方法完成时。

我们再来看下是实现获取运行状态的 runStateOf() 方法，代码如下：

private static int runStateOf(int c) { return c & ~CAPACITY; }

~ 是按位取反的意思，CAPACITY 表示的是高位的 3 个 0，和低位的 29 个 1，而 ~CAPACITY 则表示高位的 3 个 1，2 低位的 9 个 0，

然后再与入参 c 执行按位与操作，即高 3 位保持原样，低 29 位全部设置为 0，也就获取了线程池的运行状态 runState。

最后，我们再看下原子变量 ctl 的初始化方法 ctlOf()，代码如下：

private static int ctlOf(int rs, int wc) { return rs | wc; }

传入的 rs 表示线程池运行状态 runState，其是高 3 位有值，低 29 位全部为 0 的 int，而 wc 则代表线程池中有效线程的

数量 workerCount，其为高 3 位全部为 0，而低 29 位有值得 int，将 runState 和 workerCount 做或操作|处理，即用 runState 的

高 3 位，workerCount 的低 29 位填充的数字，而默认传入的 runState、workerCount 分别为 RUNNING 和 0。

## 2、BlockingQueue<Runnable> workQueue ##

阻塞队列（[LinkedBlokingQueue源码分析][LinkedBlokingQueue]）。当核心线程处理不过来的时候，任务就放入队列里面等待执行。

## 3、HashSet<Worker> workers ##

workers 是包含线程池中所有工作线程 worker 的集合，仅仅当拥有 mainLock 锁时才能访问它；

## 4、long completedTaskCount ##

completedTaskCount 是已完成任务的计数器，只有在 worker 线程的终止，仅仅当拥有 mainLock 锁时才能访问它；

## 5、volatile ThreadFactory threadFactory ##

线程工厂。用于创建线程池线程。

## 6、volatile RejectedExecutionHandler handler ##

饱和策略。当线程池达到饱和之后，提交的任务交给饱和策略处理（[饱和策略实现原理][Link 1]）。

## 7、volatile long keepAliveTime ##

线程的存活时间。这个存活时间主要控制的是大于 corePoolSize 之后创建的线程，如果这部分线程

等待 keepAliveTime 时间后，还没任务处理，哪么这部分线程就会退出线程池，可以比喻成线程池的临时工，

就像饭店一样，生意火的时候，全职的员工忙不过来，找了些临时工来帮忙，当没啥事干的时候，

这部分临时工就不用了，就像这部分线程退出了一样。

如果通过 allowCoreThreadTimeOut 设置了允许核心线程退出，这个时间也是核心线程空闲后自动退出的时间。

## 8、volatile boolean allowCoreThreadTimeOut ##

默认值为 false，如果为 false，core 线程在空闲时依然存活挂起；如果为 true，则 core 线程等待工作，直到时间超时至 keepAliveTime；

## 9、volatile int corePoolSize ##

线程池的核心线程数。每一次任务的提交只要当前线程池已创建线程数小于核心线程数（这里是小于核心线程数可以创建核心线程，

因为这个时候你再创建一个核心线程，刚好达到核心线程的数量，如果是小于等于，你再创建一个核心线程，就超过了核心线程

定义的数量了），都会创建线程执行任务，无论你当前线程池是否有空闲线程。这些线程是全职工，即便没有任务执行，

他们也会等待任务。也即，当线程池里没有任务任务时，也会有 corePoolSize 个忠诚的线程在等着执行任务。

## 10、volatile int maximumPoolSize ##

线程池最大线程池数。无论你提交多少任务，线程池里最多能创建 maximumPoolSize 个工作线程。

# 三 线程池创建 #

调用线程工厂创建线程池或自定义线程池最后都要调用ThreadPoolExecutor构造方法，

ThreadPoolExecutor有多个重载构造方法，其中一个源代码如下：

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {
        if (corePoolSize < 0 ||
            maximumPoolSize <= 0 ||
            maximumPoolSize < corePoolSize ||
            keepAliveTime < 0)
            throw new IllegalArgumentException();
        if (workQueue == null || threadFactory == null || handler == null)
            throw new NullPointerException();
        this.acc = System.getSecurityManager() == null ?
                null :
                AccessController.getContext();
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }

每个参数的含义，下面源码说明。

# 四 线程池工作原理 #

JDK默认线程池工作原理简易图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lobF9qeHk_size_16_color_FFFFFF_t_70 1][]

1）任务提交到线程池，首先判断线程池有效线程数是否小于 corePoolSize，如果小于，则创建线程执行任务，无论是否有空闲线程；

但是线程有一个例外，就是线程可能创建失败，如果创建失败了，任务也会添加到队列；

2）如果核心线程已满，判断阻塞队列是否已满，如果未满，则将任务加入队列等待执行；

3）如果阻塞队列已满，判断线程池有效线程数是否小于 maximumPoolSize，如果小于，创建线程执行任务；

4）如果线程池已满，提交的任务交由饱和策略处理（handler）；

# 五 源码解析 #

下面基于jdk8，以ThreadPoolExecutor为典型，从源码层面分析线程池工作原理。

## 1、ThreadPoolExecutor\#execute() ##

**execute() 负责将任务提交到线程池。**

源代码中解释任务提交线程池只有三个步骤，但是最后一步可以拆成2个步骤，

总共四个步骤感觉比较好理解些，这四个步骤对应上图线程池工作原理图。

public void execute(Runnable command) {
            // 如果command为空，直接抛空指针异常
            if (command == null)
                throw new NullPointerException();
            // 获取线程池有效线程数
            int c = ctl.get();
            /**
             * 1、任务提交到线程池，首先判断线程池中当前活跃线程数是否小于 corePoolSize，如果小于，
             * 则调用 addWorker() 创建工作线程去执行提交的任务。
             */
            if (workerCountOf(c) < corePoolSize) {
                if (addWorker(command, true))
                    return;
                c = ctl.get();
            }
             /**
              * 2、如果线程池线程数不小于 corePoolSize，调用阻塞队列 workQueue.offer()
              * 往队列添加任务等待执行。
              */
            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);
            }
            /**
             * 3、如果放入队列满了，放不进去，判断有效线程数是否大于等于最大线程数maximumPoolSize，
             * 如果大于等于最大线程数，则不能继续创建线程执行任务。
             */
            else if (!addWorker(command, false))
                // 4、线程池达到饱和状态后，新提交的任务交由饱和策略处理，即reject()方法处理。
                reject(command);
        }

1）任务提交小于corePoolSize，则通过addWorker()创建工作线程执行提交的任务；

2）如果corePoolSize已经满了，并都不空闲，则任务添加到阻塞队列；

3）如果阻塞队列也满了，则看是否还能创建工作线程去帮忙执行任务，能，则创建；

4）如果线程池饱和了，则提交的任务交由饱和策略处理；

## **2、**ThreadPoolExecutor\#**addWorker()** ##

**addWorker() 负责创建工作线程执行任务。**

这个方法接受一个boolean类型的core变量，标识是否是核心线程创建，如果大家细心的话，

在execute()源码中线程数小于corePoolSize时，调用addWorker()时传进来的是true，队列满了后再调时传的是false。

这个boolean的core取决于线程池线程数是与corePoolSize比较大小，还是与maximumPoolSize

比较大小，来决定能否创建相应的线程。

如果core是true，这个时候是判断是否能创建核心线程，如果线程池数量小于corePoolSize，

则能创建，否则不能创建核心线程；

如果core是false，这个时候是判断是否能再创建线程，如果线程池数量小于maximumPoolSize，

则能创建，否则不能创建非核心线程。

private boolean addWorker(Runnable firstTask, boolean core) {
        retry:
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);
    
            // Check if queue empty only if necessary.
            if (rs >= SHUTDOWN &&
                ! (rs == SHUTDOWN &&
                   firstTask == null &&
                   ! workQueue.isEmpty()))
                return false;
    
            for (;;) {
            	// 获取线程池有效线程数
                int wc = workerCountOf(c);
                // wc >= CAPACITY 如果超过理论上创建线程的最大容量，则不继续创建线程
                // wc >= (core ? corePoolSize : maximumPoolSize) 这里的 core 是 execute()调用addWorker()传入的参数，
                // core 为true，表示创建核心线程，则判有效线程是否超过了核心线程，超了，则不能继续创建
                // core 为 false， 表示创建非核心线程，则有效线程是否超过了最大线程数，超了，则不能继续创建
                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 {
            // 创建工作线程，worker实现Runnable接口
            w = new Worker(firstTask);
            // w.thread 返回的是工作线程，不是咱们提交的任务线程，任务线程的执行依赖于worker工作线程的执行
            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) {
                    /**
                     * 看到这个start()方法，是不是有一种久违的感觉，咱们从提交任务到线程池，一路走来，
                     * 终于看到了线程启动的方法，注意这个start是Worker的是start，不是我们提交的实际任务的start。
                     */
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }

## **3、Worker工作线程结构** ##

Worker继承了AQS框架，同时实现了Runnable接口，让Worker拥有了同步和线程的特性。

private final class Worker
            extends AbstractQueuedSynchronizer
            implements Runnable
        {
            /**
             * This class will never be serialized, but we provide a
             * serialVersionUID to suppress a javac warning.
             */
            private static final long serialVersionUID = 6138294804551838833L;
    
            /** Thread this worker is running in.  Null if factory fails. */
            final Thread thread;
            /** Initial task to run.  Possibly null. */
            // 这个才是我们提交的哪个业务线程，作为Worker线程的成员变量
            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)
             */
             /**
              * 创建线程，线程工厂传进去的是this，也就是Worker，而不是我们提交的任务，
              * 我们提交到线程池的任务被赋值为firstTask，成为了Worker的一个成员属性。
              */
            Worker(Runnable firstTask) {
                setState(-1); // inhibit interrupts until runWorker
                this.firstTask = firstTask;
                /**
                 * 创建线程，线程工厂传进去的是this，也就是Worker，而不是我们提交的任务，
                 * 我们提交到线程池的任务被赋值为firstTask，成为了Worker的一个成员属性。
                 */
                this.thread = getThreadFactory().newThread(this);
            }
    
            /** Delegates main run loop to outer runWorker  */
            /**
             * 我们上面已经通过start启动了线程，线程最终会执行对应启动线程的run()方法，
             * 即Workder里面的run方法。Worker里面的run()方法调用了runWorker方法，
             * 在runWorker里面才是真正执行我们提交的任务的地方。
             */
            public void run() {
                runWorker(this);
            }
    
            // Lock methods
            //
            // The value 0 represents the unlocked state.
            // The value 1 represents the locked state.
    
            protected boolean isHeldExclusively() {
                return getState() != 0;
            }
    
            protected boolean tryAcquire(int unused) {
                if (compareAndSetState(0, 1)) {
                    setExclusiveOwnerThread(Thread.currentThread());
                    return true;
                }
                return false;
            }
    
            protected boolean tryRelease(int unused) {
                setExclusiveOwnerThread(null);
                setState(0);
                return true;
            }
    
            public void lock()        { acquire(1); }
            public boolean tryLock()  { return tryAcquire(1); }
            public void unlock()      { release(1); }
            public boolean isLocked() { return isHeldExclusively(); }
    
            void interruptIfStarted() {
                Thread t;
                if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
                    try {
                        t.interrupt();
                    } catch (SecurityException ignore) {
                    }
                }
            }
        }

## 4、runWorker逻辑 ##

runWorker正真执行我们提交的任务。

final void runWorker(Worker w) {
            Thread wt = Thread.currentThread();
            // 从Worker取成员变量firstTask，即为我们提交到线程池的任务。
            Runnable task = w.firstTask;
            w.firstTask = null;
            w.unlock(); // allow interrupts
            boolean completedAbruptly = true;
            try {
                // while 循环，如果task不为空，或者getTask()从队列里面能取到任务，while就不会退出。
                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方法，task为我们提交的任务，终于执行了我们提交的任务里面的run方法。
                            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);
            }
        }

## 5、getTask()源码 ##

从阻塞队列获取任务。

private Runnable getTask() {
        boolean timedOut = false;
        
        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);
    
            /**
            * 1、allowCoreThreadTimeOut 为true，允许核心线程退出；
            * 2、wc > corePoolSize 为true，允许超过核心线程以外的线程退出；
            */
            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
    
            if ((wc > maximumPoolSize || (timed && timedOut))
                && (wc > 1 || workQueue.isEmpty())) {
                if (compareAndDecrementWorkerCount(c))
                    return null;
                continue;
            }
    
            try {
                /**
                * timed 为true，即允许线程退出，无论是核心还是非核心，通过poll()非阻塞方式获取任务。
                * timed 为false， 即不允许线程退出，则通过take()阻塞方式获取任务，没有任务，线程将阻塞，不会退出线程。
                */
                Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
                if (r != null)
                    return r;
                timedOut = true;
            } catch (InterruptedException retry) {
                timedOut = false;
            }
        }
    }

## 6、processWorkerExit(Worker w, boolean completedAbruptly) ##

1）当allowCoreThreadTimeOut=false时，不允许核心线程退出，即会保留核心线程数量的工作线程一直存活，

一直从阻塞队列等待获取任务执行，则该方法用于退出的就是超出核心线程数量的哪些线程。

2）当allowCoreThreadTimeOut=true时，则允许核心线程退出，则该方法用于退出核心线程和非核心线程。

private void processWorkerExit(Worker w, boolean completedAbruptly) {
            if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
                decrementWorkerCount();
    
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                /**
                 * 移出工作线程，与下面的
                 */
                completedTaskCount += w.completedTasks;
                workers.remove(w);
            } finally {
                mainLock.unlock();
            }
    
            tryTerminate();
    
            /**
             * allowCoreThreadTimeOut是否允许核心线程退出，如果允许，则 min为0，
             * workerCountOf(c) >= min为true，就会被return掉，不会走addWorker(null, false);去创建工作线程，
             * 核心线程就是通过与上面的workers.remove(w);配合被移出了，没有去创建；
             * 如果allowCoreThreadTimeOut设置为false，则min就是corePoolSize，则会创建corePoolSize核心线程，
             * 核心线程得以保持住不灭，而非核心线程被上面移出后在下面被return了，则退出。
             */
            int c = ctl.get();
            if (runStateLessThan(c, STOP)) {
                if (!completedAbruptly) {
                    int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
                    if (min == 0 && ! workQueue.isEmpty())
                        min = 1;
                    if (workerCountOf(c) >= min)
                        return; // replacement not needed
                }
                addWorker(null, false);
            }
        }

# 六 Executors #

为了我们快速创建线程池，jdk提供了一些常用场景的线程池创建工厂 Executors，底层都是 ThreadPoolExecutor 。

## 1、Executors\#newFixedThreadPool() ##

创建固定大小线程池。

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

1）核心线程和最大线程都是入参nThreads，这样设置是避免非核心线程的线程创建和销毁带来性能损耗；

2）keepAliveTime设置为0，这个参数在这里没有意义，只是ThreadPoolExecutor构造必传参数而已，

因为默认是控制非核心线程退出的，但是这里不会创建非核心线程；

3）队列用的是LinkedBlockingQueue（无界阻塞队列，所谓的无界，是因为队列大小设置为int的最大值）；

4）应用场景：控制最大并发数，没有空闲线程执行任务时，都放在队列里面等待执行，任务不丢弃，

但是有些任务可能入队晚了，要很久才执行到，如果要求比较及时，需要自定义线程池，把这个队列控制为有界，

队列满了的走饱和策略及时处理。

## 2、Executors\#newCachedThreadPool() ##

创建缓冲队列线程池，线程池线程数量自动伸缩，任务越多，线程越多，理论上支持创建操作系统允许创建的线程数，

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

1）核心线程数设置为0，最大线程数设置为int最大值，允许创建出操作系统允许的线程数；

2）keepAliveTime设置为60s，控制非核心线程的退出；

3）阻塞队列用SynchronousQueue，SynchronousQueue不是一个正直的队列，因为他不维护我们提交的任务，

维护的一些线程，这些线程等待元素的加入和移出队列，SynchronousQueue只是作为数据交换的地方，

你要想把东西放上去，你得看看有没有人接手，没人接，你放不进去；

根据ThreadPoolExecutor的execute执行任务逻辑源码，可以知道，当提交第一个任务时，不会去创建核心线程的，

因为设置为0了，也入不了队，因为workQueue.offer(command)这个也会失败，他跟正常的队列不一样，

因为没有空闲线程去等待获取任务，数据交换不了，只能创建线程去执行任务，正常情况下当提交第二个任务时，

如果第一个已经执行完，并且线程还未退出，第二个任务直接走offer，因为有人接手，如果提得太快，

只能不断创建线程执行任务。当没有任务提交，超过keepAliveTime时间，所有线程都会退出。

感觉太绕了，举个生活中的例子！

比如某饭店，厨师做菜，做完菜一般都是放在厨房窗口（假如这个窗口只能放一盘菜，多了放不下，

这个窗口相当于SynchronousQueue，用于数据交换），服务员直接从菜架取菜上菜。

第一回合：当厨师做完第一道菜的时候（提交第一个任务），他会尝试往菜架上放，但是他一看，没有服务员在窗口，

放窗口失败，没人取，放了也白放，他只能喊A服务员接菜（创建线程执行任务），A这个时候从厨师手里接菜，去给

客人上菜；

第二回合：A服务员上完菜之后主动回到窗口，看看有没有要上的菜，他等60s（keepAliveTIme），如果没菜，

他就退出上菜这件事情（线程退出），如果A服务员等的60S内，厨师又做了第二到菜，这个时候厨师一看A服务员在，

二话不说，直接把菜放厨房窗口（workQueue.offer(command)放入成功），A服务员二话不说，直接从窗口端菜上菜；

第三回合：厨师好快，A服务员还没上菜回来，又做了第三道菜，厨师这会看，A服务员还没回来啊，又没法直接放厨房窗口，

这个时候直接叫B服务员来把菜端走上菜（创建新线程执行任务），B直接把菜接走。

第四回合：A服务员回来后等60s，看厨房窗口上没菜，退出上菜这件事（线程退出）,B送完菜，也回到窗口，等60s发现

也没菜，也退出上菜这件事情（线程退出），全部线程退出。

4）应用场景：这种线程池模式适合快速执行处理的任务，灵活控制空闲线程的回收，如果每个任务都太长，

就不能像这样无限创建线程处理任务，就像这个饭店，如果每个服务员上菜假如都特别慢，一直把人叫来上菜，

哪整个饭店的人都来送菜，没有人洗碗，没有人擦桌子等，饭店就崩溃了（无限制一直创建线程不回收，

操作系统允许的线程数是有限的，会导致系统崩溃）。

## 3、Executors\#newSingleThreadExecutor() ##

创建一个只有一个核心线程的线程池，任务按队列先进先出顺序执行。

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

1）核心线程和最大线程都设置为1，控制线程池只有一个线程；

2）队列使用LinkedBlockingQueue无界阻塞；

3）场景就是任务按先进先出的顺序执行任务，主要用于执行少量控制任务执行顺序的场景，缺陷就是，如果并发量比较高，

线程池处理不过来，就不能用这种方式了。

## 4、Executors\#newScheduledThreadPool() ##

创建定时或周期执行任务的线程池，但是这个一般不用工厂方法，一般是直接通过ScheduledThreadPoolExecutor构造器构建。

package com.jpeony.concurrent.threadpool;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import java.util.concurrent.ScheduledThreadPoolExecutor;
    import java.util.concurrent.TimeUnit;
    
    /**
     * 定时或周期执行任务。
     *
     * @author yihonglei
     */
    public class NewScheduledThreadPoolTest {
        private final static Logger logger = LoggerFactory.getLogger(NewScheduledThreadPoolTest.class);
    
        public static void main(String[] args) {
            // 线程池
            ScheduledThreadPoolExecutor ste = new ScheduledThreadPoolExecutor(1);
            Runnable runnable = new Runnable() {
                @Override
                public void run() {
                    logger.info("ThreadName={},Time={}", Thread.currentThread().getName());
                }
            };
            ste.scheduleAtFixedRate(runnable, 5, 5, TimeUnit.SECONDS);
        }
    }

**Executors工厂方法：**

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

**ScheduledThreadPoolExecutor构造器： **

public ScheduledThreadPoolExecutor(int corePoolSize) {
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
              new DelayedWorkQueue());
    }

public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
                                                  long initialDelay,
                                                  long period,
                                                  TimeUnit unit) {
        if (command == null || unit == null)
            throw new NullPointerException();
        if (period <= 0)
            throw new IllegalArgumentException();
        ScheduledFutureTask<Void> sft =
            new ScheduledFutureTask<Void>(command,
                                          null,
                                          triggerTime(initialDelay, unit),
                                          unit.toNanos(period));
        RunnableScheduledFuture<Void> t = decorateTask(command, sft);
        sft.outerTask = t;
        delayedExecute(t);
        return t;
    }

1）核心线程个数可以自定义；

2）任务被包装成ScheduledFutureTask；

3）延时队里是延迟线程实现的根本DelayedWorkQueue，offer处理入队和通知，take处理出队和等待，

本质就是计算下次任务执行时间，通过ReentrantLock的Condition等待和通知机制实现延迟或定期执行。

# 七 总结 #

1、线程池核心概念：任务、核心线程数，工作队列，线程池最大线程数，饱和策略。

2、keepAliveTime 和 TimeUnit 都是针对非核心线程的，只有非核心线程才会存在存活多长时间退出的概念，

但是，如果设置 allowCoreThreadTimeOut=true，允许核心线程退出，那么这个参数对核心线程也生效。

3、核心线程在没有任务的时候，通过队列的 take() 一直阻塞，维持线程不退出，一旦有新任务提交线程池，

直接加入到队列，take() 上的阻塞线程被唤醒尝试获取任务执行，所以核心线程能维持等待任务状态，减少线程创建时间，

同时对系统资源进行控制。

4、任务提交后会被包装为 Worker 对象（工作线程），该对象实现了 Runnable 接口，实际提交的任务被赋值为

Worker 内的 firstTask 成员变量。Worker 调用 start() 方法执行 Worker 会执行 run() 方法，在 run() 里面调用 runWorker() 方法，

在 runWorker() 里面才是真正调用 firstTask 的 run() 方法。绕了半天才执行提交的任务的 run() 方法，处理我们的任务逻辑。

5、使用线程池有什么好处？避免线程创建和销毁资源消耗，充分利用cpu，同时也能对系统资源进行控制。

[https_github.com_yihonglei_jdk-source-code-reading]: https://github.com/yihonglei/jdk-source-code-reading
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lobF9qeHk_size_16_color_FFFFFF_t_70]: /images/20220322/6a739165a49b44009a479b19e8411ae2.png
[LinkedBlokingQueue]: https://blog.csdn.net/yhl_jxy/article/details/106544796
[Link 1]: https://blog.csdn.net/yhl_jxy/article/details/103218700
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lobF9qeHk_size_16_color_FFFFFF_t_70 1]: /images/20220322/b77b7920fa774dda958de10f405cebde.png