线程池ThreadPoolExecutor源码剖析

一、Java构建线程的方式

继承Thread （也实现了Runnable）
实现Runnable
实现Callable （与Runnable区别…）
线程池方式（Java提供了构建线程池的方式）[可以实现Runnable 和 Callable 功能]
- Java提供了Executors可以去创建（规范中不允许使用这种方式创建线程池，这种方式对线程的控制粒度比较低）
- 推荐手动创建线程池 ThreadPoolExecutor

二、线程池的7个参数

public ThreadPoolExecutor(int corePoolSize,  // 核心线程数
                          int maximumPoolSize, // 最大线程数
                          long keepAliveTime, // 最大空闲时间
                          TimeUnit unit, // 时间单位
                          BlockingQueue<Runnable> workQueue, // 阻塞队列
                          ThreadFactory threadFactory, // 线程工厂
                          RejectedExecutionHandler handler) { // 拒绝策略
}

三、线程池的执行流程

线程池执行流程
$[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-ygpetg5E-1676901244832)(C:\Users\b\AppData\Roaming\Typora\typora-user-images\image-20230220132010848.png)]$

为什么要先进阻塞再去尝试创建非核心线程：

饭店（线程池）-- 厨子（线程）-- 人多先排队（阻塞队列）-- 招厨子I(创建最大线程数) – 今日客满（拒绝）

四、线程池属性标识

4.1 核心属性

ThreadPoolExecutor 中的属性（核心成员变量）标识

学习课程： https://www.bilibili.com/video/BV1244y1n7bz/?p=4&spm_id_from=pageDriver&vd_source=c81fe4418bb0d2f341abd89cbfa157aahttps://www.bilibili.com/video/BV1244y1n7bz/?p=4&spm_id_from=pageDriver&vd_source=c81fe4418bb0d2f341abd89cbfa157aa

// 是一个int类型的数值，表达了两个意思，1：声明当前线程池的状态，2：声明线程池中的线程数
// 高3位是：线程池状态  低29位是：线程池中的线程个数
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; // 通过为运  算得出最大容量

// 线程池状态
// runState is stored in the high-order bits
private static final int RUNNING    = -1 << COUNT_BITS; // 111 代表线程池为RUNNING, 代表正常接收任务
private static final int SHUTDOWN   =  0 << COUNT_BITS; // 000 代表线程池为SHUTDOWN状态，不接收新任务，但是内部还会处理阻塞队列中的任务，正在进行的任务也正常处理
private static final int STOP       =  1 << COUNT_BITS; // O01 代表线程池为STOP状态，不接收新任务，也不去处理阻塞队列中的任务，同时会中断正在执行的任务
private static final int TIDYING    =  2 << COUNT_BITS; // 010 代表线程池为TIDYING状态，过渡的状态，代表当前线程池即将Game Over
private static final int TERMINATED =  3 << COUNT_BITS; // O11 代表线程池为TERMINATED, 要执行terminated(), 真的凉凉了

// Packing and unpacking ctl
private static int runStateOf(int c)     { return c & ~CAPACITY; } // 得到线程池的状态
private static int workerCountOf(int c)  { return c & CAPACITY; } // 得到当前线程池的线程数量
private static int ctlOf(int rs, int wc) { return rs | wc; }

// 想对位移更掌握，看雪花算法，达到手写的能力，位移方向和各种二进制运算就没问题了。

4.2 线程池状态变化

线程池状态变化
** $[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-i70mMtI0-1676901244832)(C:\Users\b\AppData\Roaming\Typora\typora-user-images\image-20230220145338221.png)]$
**

五、线程池的execute方法执行

5.1 从execute方法开始

public void execute(Runnable command) {
    // 健壮性判断
    if (command == null)
        throw new NullPointerException();
    /*
     * Proceed in 3 steps:
     *
     * 1. If fewer than corePoolSize threads are running, try to
     * start a new thread with the given command as its first
     * task.  The call to addWorker atomically checks runState and
     * workerCount, and so prevents false alarms that would add
     * threads when it shouldn't, by returning false.
     *
     * 2. If a task can be successfully queued, then we still need
     * to double-check whether we should have added a thread
     * (because existing ones died since last checking) or that
     * the pool shut down since entry into this method. So we
     * recheck state and if necessary roll back the enqueuing if
     * stopped, or start a new thread if there are none.
     *
     * 3. If we cannot queue task, then we try to add a new
     * thread.  If it fails, we know we are shut down or saturated
     * and so reject the task.
     */
    // 拿到32位的int
    int c = ctl.get();
    // 获取 工作线程数 < 核心线程数
    if (workerCountOf(c) < corePoolSize) {
        // 进到if,代表可以创建 核心 线程数
        if (addWorker(command, true))
            // 到这结束
            return;
        // 如果if没进去，代表创建核心线程数失败，重新获取ct1
        c = ctl.get();
    }
    // 判断线程池是不是RUNNING,将任务添加到阻塞队列中
    if (isRunning(c) && workQueue.offer(command)) {
        // 再次获取ct1
        int recheck = ctl.get();
        // 再次判断是否是RUNNUING,  如果不是RUNNING,移除任务
        if (! isRunning(recheck) && remove(command))
            reject(command);  // 拒绝策略
        else if (workerCountOf(recheck) == 0) // 如果线程池处在RUNNING状态，BUT工作线程为0
            addWorker(null, false); // 阻塞队列有任务，但是没有工作线程，添加一个任务为空的工作线程处理阻塞队列中的任务
    }
    // 创建非核心线程，处理任务
    else if (!addWorker(command, false))
        reject(command); // 拒绝策略
}

5.2 通过上述源码，掌握了线程池的执行流程，再次查看addWorker方法内部做了什么处理

private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    // 经过大量的判断，给工作线程数标识+1，
    for (;;) {
        // 获取ct1,(32位)
        int c = ctl.get();
        // 获取线程池状态
        int rs = runStateOf(c);
        
        // 除了RUNNING都有可能
        // Check if queue empty only if necessary.
        if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())
            // rs == SHUTDOWN，如果不是SHUTDOWN,就代表是STOP或者跟高的状态，这时，不需要添加线程处理任务
			// 任务为空，如果任务为null, 并且线程池状态不是RUNNING，不需要处理
			// 阻塞队列不为nu11，如果阻塞队列为空，返回false,.外侧的！再次取反，获取true,不需要处理
           )
            // 构建工作线程失败！
            return false;

        for (;;) {
            // 获取工作线程个数
            int wc = workerCountOf(c);
            // 如果当前线程已经大于线程池最大容量，不去创建了
            // 判断wC是否超过核心线程或者最大线程
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                // 构建工作线程失败！
                return false;
            
            //将工作线程数+1，采用CAS的方式
            if (compareAndIncrementWorkerCount(c))
                // 成功就退出外侧for循环
                break retry;
            
          	// 重新获取ct1
            c = ctl.get();  // Re-read ctl
            // 重新判断线程池状态，如果有变化   如果状态没变化，重新执行内部循环即可
            if (runStateOf(c) != rs)
                // 结束这次外侧循环，开始下次外侧循环
                continue retry;
            // else CAS failed due to workerCount change; retry inner loop
        }
    }

    // worker开始 = false
    boolean workerStarted = false;
    // worker添加 = false
    boolean workerAdded = false;
    // Worker就是工作线程
    Worker w = null;
    try {
        // 创建Worker,传入任务
        w = new Worker(firstTask);
        // 从Worker中获取线程t
        final Thread t = w.thread;
        // 如果线程t不为nul1
        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());
                // 是RUNNING状态
                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) { // 是SHUTDOV状态，创建空任务工作线程，处理阻塞队列中的任务
                    // 线程是否是运行状态
                    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,添加工作线程成功
                    workerAdded = true;
                }
            } finally {
                // 释放锁
                mainLock.unlock();
            }
            if (workerAdded) {
                // 启动工作线程
                t.start();
                // 启动工作线程成功
                workerStarted = true;
            }
        }
    } finally {
        // 如果启动工作线程失败，调用下面方法
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted; // 返回工作是否启动
}

六、Worker的封装

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. */
    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)
     */
    Worker(Runnable firstTask) {
        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);
    }
}

看runWorker方法

final void runWorker(Worker w) {
    // 获取当前线程
    Thread wt = Thread.currentThread();
    // 拿到任务
    Runnable task = w.firstTask;
    // 先不关注
    w.firstTask = null;
    w.unlock(); // allow interrupts
    
    // 标识为true
    boolean completedAbruptly = true;
    try {
        // 任务不空，执行任务。  如果任务为空，通过getTask从阻塞队列中获取任务！
        while (task != null || (task = getTask()) != null) {
            w.lock(); // 加锁，避免你shutdown我任务也不会中断
            // 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
            
            // 获取当前状态，是否大于等于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);
    }
}