线程池ThreadPoolExecutor源码分析

一、线程池基本概念和线程池前置知识

1.1 Java中创建线程的方式有哪些

传统答案：

• 继承Thread类

        通过继承Thread类并重写其run方法来创建线程。具体步骤包括定义Thread类的子类，在子类中重写run方法以实现线程的具体逻辑，然后创建子类的实例并调用其start方法来启动线程。

• 实现Runnable接口

        通过实现Runnable接口并重写其run方法来创建线程。这种方式相较于继承Thread类更为灵活，因为Java不支持多重继承，而实现接口则没有这个问题。实现Runnable接口后，需要将其实现类的实例作为参数传递给Thread类的构造函数，然后调用Thread对象的start方法来启动线程。

• 实现Callable接口

        Callable接口与Runnable接口类似，但其call方法可以返回值，并且可以抛出异常。创建Callable实现类的实例后，通常使用FutureTask类来包装它，并将FutureTask对象作为Thread对象的target来创建并启动新线程。调用FutureTask对象的get方法可以获取子线程执行结束后的返回值。

• 创建线程池

        Executor框架提供了多种线程池的实现，如单线程池（newSingleThreadExecutor）、固定大小的线程池（newFixedThreadPool）、可缓存的线程池（newCachedThreadPool）以及定时线程池（newScheduledThreadPool）等。使用线程池可以更有效地管理线程的创建、销毁和调度，从而提高系统的性能和稳定性。

1.2 为什么要使用线程池

1. 降低资源消耗：在创建和销毁线程时，会消耗一定的系统资源，包括CPU和内存。通过线程池，可以重复利用已创建的线程，从而降低这种消耗。
2. 提高响应速度：当新任务到达时，如果线程池中已有空闲线程，则任务可以立即执行，无需等待线程的创建。
3. 提高线程的可管理性：线程是稀缺资源，如果无限制地创建线程，不仅会消耗大量系统资源，还可能降低系统的稳定性和性能。线程池提供了统一的线程管理和调优机制，可以方便地对线程进行监控和管理。

1.3 线程池的核心参数

在源码中找到ThreadPoolExecutor类中参数最多的一个构造方法

这七个就是线程池的核心参数：

• int corePoolSize, 
• int maximumPoolSize,
• long keepAliveTime,
• TimeUnit unit,
• BlockingQueue<Runnable> workQueue,
• ThreadFactory threadFactory,
• RejectedExecutionHandler handler

1. corePoolSize（核心线程数）：
   • 含义：线程池中始终保持活动的线程数量，即使这些线程处于空闲状态，也不会被销毁(默认情况下)。
   • 作用：当任务提交到线程池时，如果当前线程数未达到corePoolSize，则会创建新线程来执行任务。
2. maximumPoolSize（最大线程数）：
   • 含义：线程池中允许的最大线程数量。
   • 作用：当任务队列已满，且当前线程数小于maximumPoolSize时，线程池会创建新的线程来处理任务。但线程池不会无限制地创建线程，最大线程数由maximumPoolSize指定。
3. keepAliveTime（空闲线程存活时间）：
   • 含义：当线程池中的线程数量超过corePoolSize时，空闲线程在指定的时间内没有被使用时，将被销毁，直到线程池中的线程数量减少到corePoolSize为止。
   • 计量单位：由unit参数指定。
4. unit（时间单位）：
   • 含义：keepAliveTime的时间单位。
   • 可选值：TimeUnit.DAYS（天）、TimeUnit.HOURS（小时）、TimeUnit.MINUTES（分）、TimeUnit.SECONDS（秒）、TimeUnit.MILLISECONDS（毫秒）、TimeUnit.MICROSECONDS（微秒）、TimeUnit.NANOSECONDS（纳秒）。
5. workQueue（工作队列）：
   • 含义：用于保存等待执行的任务的队列。
   • 类型：线程池提供了多种工作队列的实现，如ArrayBlockingQueue（基于数组的有界阻塞队列）、LinkedBlockingQueue（基于链表的无界阻塞队列）、SynchronousQueue（一个不缓存任务的阻塞队列）等。
   • 作用：当新任务提交到线程池时，如果线程数未达到corePoolSize，则创建新线程处理；如果线程数已达到corePoolSize但小于maximumPoolSize，且工作队列未满，则将任务放入工作队列；如果工作队列已满且线程数小于maximumPoolSize，则创建新线程处理；如果工作队列已满且线程数已达到maximumPoolSize，则执行拒绝策略。
6. threadFactory（线程工厂）：
   • 含义：用于创建新线程的工厂。
   • 作用：可以自定义线程的名称、优先级等属性。
7. handler（拒绝策略）：
   • 含义：当任务无法加入工作队列且线程池已达到最大线程数时，用于处理新提交的任务的策略。
   • 类型：常见的拒绝策略包括抛出异常、丢弃任务、丢弃队列中最老的任务、将任务分给调用线程来执行等。

二、线程池任务处理策略

线程池执行任务的方法是execute方法。

想要查看执行流程的话，需要查看的就是execute方法的源码。

将源码文本粘贴出来，逐步分析：

    // 任务交给线程池处理时，一般会执行execute方法，并传递任务
    // command 就是传递过来的任务
    public void execute(Runnable command) {
        // 非空校验
        if (command == null)
            throw new NullPointerException();
     
        // 以下是核心业务流程
        // ctl 是什么？ctl是线程池的一个核心属性。
        // 想要了解线程池的执行流程需要先知道线程池的核心属性
        int c = ctl.get();
        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 (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        else if (!addWorker(command, false))
            reject(command);
    }

 由于在execute方法中使用到了线程池的核心属性 ctl，所以我们先看一下ctl

粘贴出来分析：

    // 线程池的核心属性
    // AtomicInteger是系统底层保护的int类型，通过对int类型的数据进行封装，提供执行方法的控制进行值的原子操作。
    // 可以理解为private final int ctl = 0;
    // ctl 存储了线程池的两个核心属性：线程池状态和工作线程个数
    // int类型占32个比特位
    // 线程池状态：基于ctl的高三位存储线程池状态
    // 工作线程个数：基于ctl的低29位存储工作线程个数
    // 那么线程池中最多可以有多少个工作线程呢？答案是2^29个
    private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
 
    // Integer.SIZE=32  所以COUNT_BITS =29
    private static final int COUNT_BITS = Integer.SIZE - 3;
 
    // CAPACITY 就是2^29 也就是线程池中工作线程数的最大值
    private static final int CAPACITY   = (1 << COUNT_BITS) - 1;
 
    // 下面5个属性是线程池的状态
    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;

线程池的生命周期（五个状态）：

整明白了线程池的核心属性，下面可以继续看ThreadPoolExecutor的execute方法了

    // 任务交给线程池处理时，一般会执行execute方法，并传递任务
    // command 就是传递过来的任务
    public void execute(Runnable command) {
        // 非空校验
        if (command == null)
            throw new NullPointerException();
     
        // 以下是核心业务流程
        // ctl.get()：拿到存储线程池状态和工作线程个数的核心属性
        int c = ctl.get();
        // workerCountOf（） 获取工作线程个数 corePoolSize：核心线程数
        // 判断当前工作线程数是否小于核心线程数（构建线程池时指定的）
        if (workerCountOf(c) < corePoolSize) {
            // 创建工作线程
            //第一个参数 command ：传递的任务
            //第二个参数 是否是核心线程，true：创建核心线程；false：创建非核心线程
            // addWorker方法 返回值是布尔类型  代表创建是否成功           
            if (addWorker(command, true))
                // 创建成功，结束，任务给核心线程处理
                return;
            //创建失败 重新获取核心属性ctl
            c = ctl.get();
        }
        
 
         // 如果当前工作线程数已经达到核心线程数 执行下面的语句
         // isRunning(c) 判断当前线程池是不是running状态 如果是直接将任务扔进工作队列
         // offer方法：扔到工作队列，成功返回true，失败返回false
        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);
        }
        // 尝试构建非核心线程去处理当前任务
        else if (!addWorker(command, false))
            // 如果非核心线程创建失败 执行拒绝策略
            reject(command);
    }

从上面源码可以看出，执行流程为：核心线程-工作队列-非核心线程-拒绝策略

如果想要修改默认的顺序可以重写execute方法来实现。

问题1：

下面代码块，是在工作线程数量小于核心线程数量时执行的代码，作用是，工作线程数量小于核心线程数量时，创建一个核心线程，并把任务给这个核心线程。

  if (workerCountOf(c) < corePoolSize) {
                    
            if (addWorker(command, true))
                // 创建成功，结束，任务给核心线程处理
                return;
            //创建失败 重新获取核心属性ctl
            c = ctl.get();
        }

那么问题是为什么创建核心线程失败后要使用 c = ctl.get();来重新获取核心属性呢？

举个例子来回答这个问题，假如核心线程数是5，当前工作线程数是4，在并发情况下，有两个线程同时进入到if代码块里面，都去执行addWorker(command, true)方法，创建核心工作线程，由于addWorker方法内部通过一定的方式保证了原子性，所以只能创建成功1个核心工作线程，另一个不会创建成功（返回false），这种情况，不会直接return，会去继续执行方法下面的代码，下面的代码会使用到核心线程属性ctl，而此时显然ctl核心线程属性已经发生了变化（另外一个线程创建成功了一个工作线程，工作线程数量发生了变化），需要重新获取最新的。

问题2：

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

这段代码中，主要实现的是如果线程池是Running状态，将任务放入工作队列的功能。

那么isRunning(c) 和workQueue.offer(command)两句代码不是就能实现这个功能了吗，为什么还要写上述代码？

因为在多线程环境中，当执行完isRunning(c) 和执行workQueue.offer(command)之间的间隙，线程池的状态可以就发生了变化。

比如一开始线程池是Running状态，执行isRunning(c)得到返回值true，然后在执行workQueue.offer(command)代码前线程池状态改变，不再是Running状态了，不过由于之前获取到的isRunning(c)的返回值是true，所以还是会执行workQueue.offer(command)方法。

这样显然是有问题的，所以为了应对这种情况，在下面添加了以下代码：

// ctl.get();重新获取线程池核心属性
int recheck = ctl.get();
// 根据核心属性 重新校验线程池是否是Running状态，如果不是Running状态，移除前边添加的任务并执行拒绝策略
if (! isRunning(recheck) && remove(command))
    reject(command);
// 如果线程池状态还是为Running状态 则执行下面语句
// 如果线程池仍然在运行，但此时工作线程的数量为0（workerCountOf(recheck)返回0），则调用addWorker(null, false)方法来添加一个新的核心线程。
//null 作为第一个参数，表示这个新线程在启动时不带有任何初始任务 ，false代表创建非核心线程
else if (workerCountOf(recheck) == 0)
                addWorker(null, false);

关于   addWorker(null, false)：

1. firstTask 参数允许你指定一个新创建的线程应该首先运行的任务。如果 firstTask 是 null，那么新线程将不会立即执行任何任务，而是会等待从工作队列中取出一个任务来执行。

2. 在线程池的工作机制中，如果当前线程数少于 corePoolSize（核心线程数），并且有新任务提交，那么线程池会尝试启动一个新线程来执行这个任务，而不是将任务放入工作队列。

三、创建工作线程的流程

查看创建工作线程的流程主要是查看addWorker方法的源码

对源码逐步解析:

// 创建工作线程（核心线程和非核心线程都是基于addWorker创建的）
// 第一个参数：任务，第二个参数：指定是核心还是非核心
private boolean addWorker(Runnable firstTask, boolean core) {
 
        // 第一部分代码
        // 做两件事情：
        // 1.判断线程池状态（外层for循环）
        // 2.判断线程个数（内层for循环），然后基于cas修改ctl属性，给工作线程个数+1
 
    
        retry:  // 给外层for循环取一个名称retry
        for (;;) { // 死循环 相当于while（true）
            int c = ctl.get(); // 拿到核心属性ctl
            int rs = runStateOf(c); // rs:高三位的线程池状态
 
            // rs >=SHUTDOWN 即rs为SHUTDOWN/STOP/TIDYING/TERMINATED 即不是Running状态
            // 根据前面线程池五个属性的那张图可知线程池不能接收新任务
            if (rs >= SHUTDOWN &&
                // 第二个判断 解决在SHUTDOWN 状态下，没有工作线程，但是工作队列有任务
                // 要构建一个线程处理阻塞队列（工作队列）任务的情况
                // rs == SHUTDOWN &&firstTask == null &&! workQueue.isEmpty())的情况不能走return语句
                ! (rs == SHUTDOWN &&firstTask == null &&! workQueue.isEmpty()))
                return false;
 
 
            for (;;) {
                int wc = workerCountOf(c); // 工作线程个数
                // CAPACITY :工作线程最大值
                if (wc >= CAPACITY ||
                    // 核心线程：判断corePoolSize 
                    // 非核心线程：判断maximumPoolSize
                    // core 是addWorker的第二个参数 true代表核心线程，false代表创建非核心线程
                    wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
                // 为什么要用CAS的方式修改？
                // 为了避免多线程并发创建工作线程，导致破坏设置的核心参数（比如设置核心参数是5，创建了6个）
                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
            }
        }
 
 
        // 第二部分代码
        // 做两件事情：1.创建工作线程 2.启动工作线程
        
        
        boolean workerStarted = false; 
        boolean workerAdded = false;
        Worker w = null;// Worker 对象就是工作线程
        try {
            // 创建工作线程，并把任务交给Worker 对象
            w = new Worker(firstTask);
            // 将new Worker时创建的thread拿到
            final Thread t = w.thread;
            // 判断使用线程池的用户，指定的线程工厂构建的thread不是null
            if (t != null) {
                // 同步锁   为什么要加锁？
                // 下面workers.add(w); 按住Ctrl+单击 查看可知 workers是HashSet类型
                // HashSet不是线程安全的 为了保证线程安全，所以加锁了，不加锁不安全
                final ReentrantLock mainLock = this.mainLock;
                mainLock.lock();
                try {
                   
                    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) {
                    t.start();// 启动一个线程，执行run方法
                    workerStarted = true;
                }
            }
        } finally {
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }

根据源码总结：

线程池在创建工作线程时：

• 判断线程池是否符合要求。
• 判断工作线程个数是否符合要求，并且基于CAS保证原子性。
• new Worker创建工作线程，并添加HashSet中，基于ReentrantLock保证原子性。
• 拿到Worker对象中的thread，执行start方法，启动线程。

四、工作线程Worker

线程池中的工作线程就是Worker对象，查看一下Worker里面做了什么事情。

  // Worker是工作线程，Worker也会存储一个任务（只存储第一个任务）
private final class Worker
        extends AbstractQueuedSynchronizer
        implements Runnable
    {      
 
        final Thread thread; // 工作线程
 
        Runnable firstTask; // 任务
     
        volatile long completedTasks;
 
        Worker(Runnable firstTask) {
            setState(-1); // inhibit interrupts until runWorker
            // 任务存放
            this.firstTask = firstTask;
            // 线程构建 传入this 当前对象
            // 创建的线程调用start启动时，执行的是谁的run方法？
            // 执行的是Worker对象里面的run方法
            this.thread = getThreadFactory().newThread(this);
        }
 
        // addWorker方法中执行thread.start()方法后，执行的是Worker对象中的run方法
        public void run() {
            runWorker(this);
        }
}

AQS是什么?

AQS是AbstractQueuedSynchronizer，是JUC包下的并发基础类，很多同步内容都是基于AQS实现的，比如：

• ReentrantLock
• ReentrantReadWriteLock
• CountDownLatch
• Semaphore
• 线程池中的Worker对象也是基于AQS做了一个实现

Worker继承AQS干嘛？

Worker线程继承了AQS后，可以使用基于CAS修改的属性state

在shutdown状态下空闲线程要执行interrupt中断，工作中的线程，不能执行interrupt

工作线程在处理任务前，会先执行lock方法（将state从0改为1），也就是正在干活的线程state==1

在终端线程前，判断每一个线程的state，如果为0直接interrupt，如果为1什么也不做。

五、执行/拉取任务流程

5.1 执行任务

执行任务，就是启动工作线程后，执行了Worker对象中的run方法，run方法中执行了runWorker方法。 

工作线程直接通过task.run执行任务，并且线程池预留了beforeExecute和afterExecute方法，可以在任务执行前后做一些额外处理。

// runWorker传递的参数就是Worker对象本身
final void runWorker(Worker w) {
        // 拿到工作线程中的thread
        Thread wt = Thread.currentThread();
        // 拿到Worker对象中存储的第一个任务
        Runnable task = w.firstTask;
        w.firstTask = null;
        w.unlock(); // allow interrupts
        boolean completedAbruptly = true;
        try {
            // 如果Worker对象在启动时携带了任务，那就优先执行携带的任务
            while (task != null || (task = getTask()) != null) {
                w.lock();
                 // 判断线程池状态是不是stop 如果时stop强制中断当前线程
                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 {
                        // 钩子函数  执行任务之后执行该函数，默认该方法方法体为空，没有实现，需要的话重写这个方法
                        afterExecute(task, thrown);
                    }
                } finally {
                    task = null;
                    w.completedTasks++;
                    w.unlock();
                }
            }
            completedAbruptly = false;
        } finally {
            processWorkerExit(w, completedAbruptly);
        }
    }

5.2 拉取任务

工作线程在处理完自带的任务后会直接基于getTask方法，从阻塞队列中拉取任务。

如果是核心线程，默认情况下，会基于take方法在工作队列中拉取任务。

如果是非核心线程，会基于poll方法，拉取指定时间任务。（时间到了直接告辞）

// 工作线程从阻塞队列拉取任务的操作
private Runnable getTask() {
        boolean timedOut = false; // Did the last poll() time out?
 
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);
 
            // 判断线程池状态 如果状态已经变为stop或者状态为shutdown且工作队列任务都处理完毕
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
                // 工作线程个数-1 并且干掉当前工作线程 
                // 线程正常死亡：run方法结束
                decrementWorkerCount();
                return null;
                //  return null： runWarker方法中 while循环结束，线程正常消亡；
                // 因为是Worker类中的run 方法中调用的runWarker方法，并且run方法中没有其他代码
                // 所以runWarker结束，run方法也就结束了，线程也就消亡了
            }
 
            int wc = workerCountOf(c);
 
            // 核心线程执行take  非核心线程执行poll方法（poll方法拉取最大空闲时间）
            // 线程池中的核心线程可以基于keepAliveTime（最大空闲时间）去结束吗？
            // 或者说线程池中的核心线程一定会永远存放在线程池里面吗？
            // 不一定 有一个属性allowCoreThreadTimeOut：是否允许核心线程超时，默认是false，但可以设置为true
            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
 
            if ((wc > maximumPoolSize || (timed && timedOut))
                && (wc > 1 || workQueue.isEmpty())) {
                if (compareAndDecrementWorkerCount(c))
                    return null;
                continue;
            }
 
            try {
             
                Runnable r = timed ?
                    // poll方法，拉取阻塞队列任务，指定keepAliveTime时间
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    // take方法，死等任务，知道中断
                    workQueue.take();
                if (r != null)
                    return r;
                timedOut = true;
            } catch (InterruptedException retry) {
                timedOut = false;
            }
        }
    }

六、线程池关闭流程和锁

shutdown方法：

• 将线程池状态修改为shutdown
• 将空闲的工作线程直接中断
• 在确认工作队列中的任务全部处理完并且工作线程个数为0，自动改为tidying状态
• 在tidying状态下，执行terminated方法，变为terminated状态
• 到这线程池结束

    // 优雅的关闭线程池
    // 为什么要加锁？
    // 会操作Worker对象
    public void shutdown() {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            checkShutdownAccess();
            advanceRunState(SHUTDOWN);// 线程池状态修改为shutdown
            interruptIdleWorkers();// 中断空闲的工作线程
            onShutdown(); // hook for ScheduledThreadPoolExecutor
        } finally {
            mainLock.unlock();
        }
        tryTerminate();
    }

shutdownNow方法：

• 将线程池状态修改为stop
• 将所有的工作线程直接中断
• 在工作线程个数为0之后，自动改为tidying状态
• 在tidying状态下，执行terminated方法，变为terminated状态
• 到这线程池结束

---

在工作中核心线程数、最大线程数、最大空闲时间、任务队列怎么设置比较好？

如果为了充分发挥硬件性能，一般只需要考虑三个信息的设置

• 核心线程数：最重要的属性，你要根据你的任务类型，判断线程数设置多少合适

        任务类型：

                CPU密集型任务：线程一直在干活，不希望CPU做上下文切换。

                io密集型任务：因为线程干一会儿歇一会儿。

                混合型任务：因为混合型偶尔要求CPU一直调度，偶尔不干活，可以切换。

                想要设置好核心线程数，去发挥服务器硬件性能，需要动态的调试和压测。为了避免调试参数时反复重启，并且成本太高，可以直接设置动态线程池，因为线程池提供了Set方法设置核心参数，以及get方法查看核心参数，可以在压测时，根据CPU占用率和使用情况，来调整核心线程数。

• 工作队列长度：根据服务器内存使用情况来调试，同时也要查看好jvm的堆内存大小。
• 拒绝策略：根据任务来决定，如果任务就是写个日志啥的，那就直接扔了，如果任务是必须要执行的，那就直接重试，或者存储到数据库，后期做同意补偿操作。