ThreadPoolExecutor threadPoolExecutor= new ThreadPoolExecutor( 
           5,    //核心线程数
                 200,    //最大线程数量，控制资源并发
                 10,    //存活时间
                TimeUnit.SECONDS,    //时间单位
                new LinkedBlockingDeque<>(  100000),    //任务队列，大小100000个
        Executors.defaultThreadFactory(),    //线程的创建工厂
        new ThreadPoolExecutor.AbortPolicy());    //拒绝策略
        CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {    //开启异步编排，有返回值
            return 1;
        }, threadPoolExecutor).thenApplyAsync(res -> {    //串行化，接收参数并有返回值
            return res+1;
        }, threadPoolExecutor);
        Integer integer = future.get();    //获取返回值

new PriorityBlockingQueue<>((o1, o2) -> o1.length() - o2.length());

// 源码
// FixedThreadPool: 固定大小线程池
// CachedTheadPool: 缓存线程池
new ThreadExcutor(0, max_valuem, 60L, s, new SynchronousQueue<Runnable>());
new ThreadExcutor(n, n, 0L, ms, new LinkedBlockingQueue<Runable>()
// ScheduledThreadPoolExcutor: 定时任务线程池
ScheduledThreadPool, SingleThreadScheduledExecutor
// SingleThreadExecutor: 单线程化线程池
new ThreadExcutor(1, 1, 0L, ms, new LinkedBlockingQueue<Runable>())

/**
 * @Author: vince
 * @CreateTime: 2024/09/10
 * @Description: 测试类
 * @Version: 1.0
 */
public class Test {
    public static void main(String[] args) {
        ExecutorService pool = Executors.newFixedThreadPool(5);
        for (int i = 0; i < 3; i++) {
            pool.execute(() -> System.out.println("Hello World"));
        }
    }
}

package java.util.concurrent;
/**
 * 自定义线程池
 * @since 1.5
 * @author Doug Lea
 */
public class ThreadPoolExecutor extends AbstractExecutorService {
    // ...
}
public abstract class AbstractExecutorService implements ExecutorService {
    //  ...
}

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class ThreadPoolExample {
    public static void main(String[] args) {
        // 创建一个具有固定线程数的线程池
        ExecutorService executorService = Executors.newFixedThreadPool(5);
        // 提交任务给线程池执行
        for (int i = 0; i < 10; i++) {
            final int taskId = i;
            executorService.execute(() -> {
                System.out.println("Task ID : " + taskId + " 执行中...");
                // 模拟任务执行耗时
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("Task ID : " + taskId + " 完成");
            });
        }
        // 关闭线程池
        executorService.shutdown();
    }
}

public class ThreadPoolExample {
    public static void main(String[] args) {
        // 创建一个具有固定线程数的线程池
        ExecutorService executorService = Executors.newFixedThreadPool(5);
        // 提交任务给线程池执行
        for (int i = 0; i < 10; i++) {
            int taskId = i;
            Future<String> future = executorService.submit(() -> {
                System.out.println("Task ID : " + taskId + " 执行中...");
                // 模拟任务执行耗时
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                return "Task ID : " + taskId + " 完成";
            });
            // 获取任务的执行结果
            try {
                String result = future.get();
                System.out.println(result);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        // 关闭线程池
        executorService.shutdown();
    }
}

// 创建线程池
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                5, // 核心线程数
                200, // 最大线程数量
                10, // 空闲线程存活时间
                TimeUnit.SECONDS, // 时间单位
                new LinkedBlockingDeque<>(100000), // 任务队列，大小100000
                Executors.defaultThreadFactory(), // 线程工厂
                new ThreadPoolExecutor.AbortPolicy() // 拒绝策略
        );

new PriorityBlockingQueue<>((o1, o2) -> o1.length() - o2.length());

import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class ThreadPoolExample {
    public static void main(String[] args) {
        // 创建线程池
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                5, // 核心线程数
                200, // 最大线程数量
                10, // 空闲线程存活时间
                TimeUnit.SECONDS, // 时间单位
                new LinkedBlockingDeque<>(100000), // 任务队列，大小100000
                Executors.defaultThreadFactory(), // 线程工厂
                new ThreadPoolExecutor.AbortPolicy() // 拒绝策略
        );
        // 提交任务到线程池
        for (int i = 0; i < 10; i++) {
            threadPoolExecutor.execute(() -> {
                System.out.println(Thread.currentThread().getName() + " 正在执行任务");
            });
        }
        // 关闭线程池
        threadPoolExecutor.shutdown();
    }
}

/**
     * 不指定线程工厂的构造方法
     * @param corePoolSize 核心线程数
     * @param maximumPoolSize 最大线程数
     * @param keepAliveTime 线程存活时间
     * @param unit 线程存活时间单位
     * @param workQueue 任务队列
     * @param handler 拒绝策略
     */
    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              RejectedExecutionHandler handler) {
        // 不指定线程工厂，将使用默认工厂，调用7参数的构造方法
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
                Executors.defaultThreadFactory(), handler);
    }

/**
     * 构造方法初始化
     * @param corePoolSize 核心线程数
     * @param maximumPoolSize 最大线程数
     * @param keepAliveTime 非核心空闲线程存活时间
     * @param unit 非核心空闲线程存活时间单位
     * @param workQueue 任务队列
     * @param threadFactory 线程工厂(生产线程的地方)
     * @param handler 拒绝策略
     */
    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {
        // 1.参数校验：
        // 核心线程数不能为负
        // 最大线程数不能为0且必须大于核心线程数
        // 非核心空闲线程存活时间不能为负
        if (corePoolSize < 0 || maximumPoolSize <= 0 || maximumPoolSize < corePoolSize || keepAliveTime < 0)
            throw new IllegalArgumentException();
        // 任务队列、线程工厂和拒绝策略不能为空
        if (workQueue == null || threadFactory == null || handler == null)
            throw new NullPointerException();
        // 2.将传入的参数赋值给线程池的成员变量
        // 核心线程数
        this.corePoolSize = corePoolSize;
        // 最大线程数
        this.maximumPoolSize = maximumPoolSize;
        // 任务队列
        this.workQueue = workQueue;
        // 将存活时间转换为纳秒存储
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        // 线程工厂，用于创建新线程
        this.threadFactory = threadFactory;
        // 拒绝策略，当线程池无法执行任务时使用
        this.handler = handler; 
    }

/**
     * 直接抛异常拒绝策略
     * {@code RejectedExecutionException}.
     */
    public static class AbortPolicy implements RejectedExecutionHandler {
        /**
         * 构造方法
         */
        public AbortPolicy() {
        }
        /**
         * 拒绝执行方法：总是抛出RejectedExecutionException异常
         *
         * @param r 线程
         * @param e 线程池
         * @throws RejectedExecutionException 总是抛出该异常
         */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            // 代码只有一行，抛出异常
            throw new RejectedExecutionException("Task " + r.toString() +
                    " rejected from " +
                    e.toString());
        }
    }

/**
     * 同步调用拒绝策略
     */
    public static class CallerRunsPolicy implements RejectedExecutionHandler {
        /**
         * 创建一个 {@code CallerRunsPolicy} 实例。
         */
        public CallerRunsPolicy() {
        }
        /**
         * 拒绝执行方法
         * 此时任务将被丢弃。
         *
         * @param r 线程
         * @param e 线程池
         */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            // 如果线程池还活着，直接同步调用线程的run()方法
            if (!e.isShutdown()) {
                r.run();
            }
        }
    }

/**
     * 丢弃最老任务策略
     */
    public static class DiscardOldestPolicy implements RejectedExecutionHandler {
        /**
         * 为给定的执行器创建一个 {@code DiscardOldestPolicy} 实例。
         */
        public DiscardOldestPolicy() {
        }
        /**
         * 丢弃最老任务
         *
         * @param r 线程
         * @param e 线程池
         */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            // 如果线程池还活着，就让最老任务出队，执行最新任务
            if (!e.isShutdown()) {
                e.getQueue().poll();
                e.execute(r);
            }
        }
    }

/**
     * 丢弃新任务策略
     */
    public static class DiscardPolicy implements RejectedExecutionHandler {
        /**
         * 创建一个 {@code DiscardPolicy} 实例。
         */
        public DiscardPolicy() {
        }
        /**
         * 不执行任何操作，相当于丢弃任务 r
         *
         * @param r 线程
         * @param e 线程池
         */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        }
    }

/**
 * @Author: vince
 * @CreateTime: 2024/08/20
 * @Description: 丢最老和新任务拒绝策略
 * @Version: 1.0
 */
public class DiscardOldestAndNewestPolicy implements RejectedExecutionHandler {
    /**
     * 自定义拒绝策略，丢弃最老的和最新的任务，并打印日志。
     */
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        BlockingQueue<Runnable> queue = executor.getQueue();
        // 检查队列是否为空，不为空则移除最老的任务
        if (!queue.isEmpty()) {
            Runnable oldestTask = queue.poll();
            System.out.println("丢弃最老的任务: " + oldestTask.toString());
        }
        // 打印日志并丢弃最新的任务：不对线程对象执行操作，就相当于丢弃了它
        System.out.println("丢弃最新的任务: " + r.toString());
    }
}

/*
     * 状态和线程数：用来标记线程池状态（高3位），线程个数（低29位，所以线程个数最多是2^29-1）
     * 默认是RUNNING状态，线程个数为0
     */
    private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
    /*
     * 线程个数掩码位数：线程池中最大可以容纳的线程数量。
     *  int：4字节（32位）,数据范围是 `-2^31 ~ 2^31-1`。
     * 所以准确说是具体平台下Integer的二进制位数-3后的剩余位数才是线程的个数，
     * 32-3=29
     * 为什么COUNT_BITS是29而不是32？因为int类型是32位，我们需要将高3位保留用于表示线程池的状态。剩下的29位则用来表示线程个数。
     */
    private static final int COUNT_BITS = Integer.SIZE - 3;
    /*
     * 线程最大个数(低29位)00011111111111111111111111111111
     * 默认容量：2^29-1
     * 位运算1左移32位，相当于2的32次方
     * （左移<<：是将二进制数的位左移若干位，低位补0，相当于乘以2的n次方。）
     */
    private static final int CAPACITY = (1 << COUNT_BITS) - 1;
    /*
     * 线程池有5种状态，按大小排序如下：
     * RUNNING < SHUTDOWN < STOP < TIDYING < TERMINATED
     * RUNNING：运行。线程池处于正常状态，可以接受新的任务，同时会按照预设的策略来处理已有任务的执行。
     * SHUTDOWN：关闭。线程池处于关闭状态，不再接受新的任务，但是会继续执行已有任务直到执行完成。执行线程池对象的shutdown()时进入该状态。
     * STOP：停止。线程池处于关闭状态，不再接受新的任务，同时会中断正在执行的任务，清空线程队列。执行shutdownNow()时进入该状态。
     * TIDYING：整理。所有任务已经执行完毕，线程池进入该状态会开始进行一些结尾工作，比如及时清理线程池的一些资源。
     * TERMINATED：终止。线程池已经完全停止，所有的状态都已经结束了，线程池处于最终的状态。
     */
    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;
    /**
     * 可重入锁，用于加锁往workers里加线程
     */
    private final ReentrantLock mainLock = new ReentrantLock();
    /**
     * 一个HashSet，存有所有工作线程
     */
    private final HashSet<Worker> workers = new HashSet<Worker>();
    /**
     * 是否允许核心线程超时：默认是false，即核心线程永不超时
     * 如果是true，核心线程将根据参数的超时时间存活
     */
    private volatile boolean allowCoreThreadTimeOut;

/*
     * 状态和线程数：用来标记线程池状态（高3位），线程个数（低29位，所以线程个数最多是2^29-1）
     * 默认是RUNNING状态，线程个数为0
     */
    private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

private static final int COUNT_BITS = Integer.SIZE - 3;
    /*
     * 线程最大个数(低29位)00011111111111111111111111111111
     * 默认容量：2^29-1
     * 位运算1左移32位，相当于2的32次方
     * （左移<<：是将二进制数的位左移若干位，低位补0，相当于乘以2的n次方。）
     */
    private static final int CAPACITY = (1 << COUNT_BITS) - 1;

/*
     * 线程池有5种状态，按大小排序如下：
     * RUNNING < SHUTDOWN < STOP < TIDYING < TERMINATED
     * RUNNING：运行。线程池处于正常状态，可以接受新的任务，同时会按照预设的策略来处理已有任务的执行。
     * SHUTDOWN：关闭。线程池处于关闭状态，不再接受新的任务，但是会继续执行已有任务直到执行完成。执行线程池对象的shutdown()时进入该状态。
     * STOP：停止。线程池处于关闭状态，不再接受新的任务，同时会中断正在执行的任务，清空线程队列。执行shutdownNow()时进入该状态。
     * TIDYING：整理。所有任务已经执行完毕，线程池进入该状态会开始进行一些结尾工作，比如及时清理线程池的一些资源。
     * TERMINATED：终止。线程池已经完全停止，所有的状态都已经结束了，线程池处于最终的状态。
     */
    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;

/**
     * 可重入锁，用于加锁往workers里加线程
     */
    private final ReentrantLock mainLock = new ReentrantLock();
    /**
     * mainLock锁的Condition对象，用于线程通信
     * Condition对象：用于线程间通信，通过await()和signal(),signalAll()让线程等待或唤醒。通常用lock锁创建Condition对象，即lock.newCondition();
     */
    private final Condition termination = mainLock.newCondition();
    /**
     * 一个HashSet，存有所有工作线程
     */
    private final HashSet<Worker> workers = new HashSet<Worker>();

/**
     * @Author: vince
     * @CreateTime: 2024/09/09
     * @Description: Worker继承了AQS，目的就是为了控制工作线程的中断。
     * Worker实现了Runnable，它的run()方法调用runWorker()启动线程
     * @Version: 1.0
     */
    private final class Worker extends AbstractQueuedSynchronizer implements Runnable{
        private static final long serialVersionUID = 6138294804551838833L;
        // 当前线程工厂创建的线程(也是执行任务使用的线程)
        final Thread thread;
        // 当前的第一个任务
        Runnable firstTask;
        // 记录执行了多少个任务
        volatile long completedTasks;
        // 构造方法
        Worker(Runnable firstTask) {
            // 将State设置为-1，代表当前不允许中断线程
            setState(-1);
            // 设置任务
            this.firstTask = firstTask;
            // 设置线程
            this.thread = getThreadFactory().newThread(this);
        }
        /**
         * 线程启动执行的方法，调用了runWorker()方法
         */
        public void run() {
            runWorker(this);
        }
        /**
         * 中断工作线程
         */
        void interruptIfStarted() {
            Thread t;
            // 只有Worker中的state >= 0的时候，可以中断工作线程
            if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
                try {
                    // 如果状态正常，并且线程未中断，这边就中断线程
                    t.interrupt();
                } catch (SecurityException ignore) {
                }
            }
        }
    }

/**
     * 是否允许核心线程超时：默认是false，即核心线程永不超时
     * 如果是true，核心线程将根据参数的超时时间存活
     */
    private volatile boolean allowCoreThreadTimeOut;

import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class ThreadPoolExample {
    public static void main(String[] args) {
        // 创建线程池
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                5, // 核心线程数
                200, // 最大线程数量
                10, // 空闲线程存活时间
                TimeUnit.SECONDS, // 时间单位
                new LinkedBlockingDeque<>(100000), // 任务队列，大小100000
                Executors.defaultThreadFactory(), // 线程工厂
                new ThreadPoolExecutor.AbortPolicy() // 拒绝策略
        );
        // 提交任务到线程池
        for (int i = 0; i < 10; i++) {
            threadPoolExecutor.execute(() -> {
                System.out.println(Thread.currentThread().getName() + " 正在执行任务");
            });
        }
        // 关闭线程池
        threadPoolExecutor.shutdown();
    }
}

import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class ThreadPoolExample {
    public static void main(String[] args) {
        // 创建线程池
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                5, // 核心线程数
                200, // 最大线程数量
                10, // 空闲线程存活时间
                TimeUnit.SECONDS, // 时间单位
                new LinkedBlockingDeque<>(3), // 任务队列，大小100000
                Executors.defaultThreadFactory(), // 线程工厂
                new ThreadPoolExecutor.AbortPolicy() // 拒绝策略
        );
        // 提交任务到线程池
        for (int i = 0; i < 10; i++) {
            threadPoolExecutor.execute(() -> {
                System.out.println(Thread.currentThread().getName() + " 正在执行任务");
            });
        }
        // 关闭线程池
        threadPoolExecutor.shutdown();
    }
}

/**
     * 执行任务
     * @param command Runnable对象
     */
    public void execute(Runnable command) {
        // 0.判空：如果当前传过来的任务是null,直接抛出异常即可
        if (command == null)
            throw new NullPointerException();
        int c = ctl.get();
        // 1.启动核心线程数：如果corePoolSize没到最大值就创建核心线程直接执行新任务
        if (workerCountOf(c) < corePoolSize) {
            // Step2:添加一个核心线程
            if (addWorker(command, true)){
                return;
            }
            // 更新一下当前值
            c = ctl.get();
        }
        // 如果走到下面会有两种情况：
        // a、核心线程数满了,需要往阻塞队列里面扔任务
        // b、核心线程数满了,阻塞队列也满了,执行拒绝策略
        // 2.任务放至阻塞队列：判断当前的状态是不是Running的状态(RUNNING可以处理任务，并且处理阻塞队列中的任务)
        // 如果是Running的状态,则尝试将任务放至阻塞队列。队列的offer(E e)：用于非阻塞的插入操作，插入失败会返回false而不是等待
        if (isRunning(c) && workQueue.offer(command)) {
            // 再次更新数值
            int recheck = ctl.get();
            // 再次校验当前的线程池状态是不是Running
            // 如果线程池状态不是Running的话,需要删除掉刚刚放的任务
            if (!isRunning(recheck) && remove(command)){
                // 执行拒绝策略
                reject(command);
            }
            // 如果到这里,说明上面阻塞队列中已经有数据了
            // 如果线程池的个数为0的话,需要创建一个非核心工作线程去执行该任务
            // 不能让人家堵塞着
            else if (workerCountOf(recheck) == 0){
                addWorker(null, false);
            }
        }
        // 此时会有两种情况：
        // a、任务成功放进阻塞队列，正在等待被执行；
        // b、放进阻塞队列失败,等待添加非核心工作线程
        // 3.启动非核心线程数
        //    尝试添加非核心工作线程，若添加非核心工作线程失败,证明已经到达maximumPoolSize的限制,执行拒绝策略
        // 添加一个非核心工作线程
        else if (!addWorker(command, false))
            // 工作队列中添加任务失败,执行拒绝策略
            reject(command);
    }

/**
     * 添加工作线程
     * @param firstTask Runnable对象
     * @param core 是否是核心线程
     * @return boolean
     */
    private boolean addWorker(Runnable firstTask, boolean core) {
        // retry主要是为了结束整个循环
        // retry 是一个标签（label），用来标记循环的某个位置。
        // 这个标签与 for (;;) 循环结合使用，调用continue retry时重新开始整个循环，调用break retry时结束整个循环
        retry:
        for (;;) {
            // 获取当前线程池的ctl：包含线程池状态和数量信息
            int c = ctl.get();
            // 获取线程池状态：根据ctl，获取线程池状态
            // runStateOf:基于&运算的特点，保证只会拿到ctl高三位的值
            int rs = runStateOf(c);
            // 1.校验线程池状态和队列：如果不是运行时，直接退出循环
            // rs >= SHUTDOWN:表示此时不再接收新任务。代表当前线程池状态为：SHUTDOWN、STOP、TIDYING、TERMINATED,线程池状态异常
            // RUNNING < SHUTDOWN < STOP < TIDYING < TERMINATED
            // rs == SHUTDOWN，这时表示关闭状态，不再接受新提交的任务，但却可以继续处理阻塞队列中已保存的任务
            // 但这里SHUTDOWN状态稍许不同(不会接收新任务，正在处理的任务正常进行，阻塞队列的任务也会做完)
            if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && !workQueue.isEmpty())){
                return false;
            }
            for (;;) {
                // 2.校验工作线程数量：如果超过线程数上限直接失败
                // 获取线程数
                int wc = workerCountOf(c);
                 // CAPACITY：是ctl的低29位的最大值（二进制是29个1），返回false；
                 // core：addWorker()方法的第二个参数，如果为true表示根据corePoolSize来比较，
                 // 如果为false则根据maximumPoolSize来比较。
                if (wc >= CAPACITY ||
                        wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
                // 3.尝试给ctl加1：如果成功，则跳出第一个for循环；如果失败，则重新循环
                // 尝试CAS方式自增workCount  其实就是类似锁，同时只有一个线程能成功
                if (compareAndIncrementWorkerCount(c))
                    break retry;
                // 3.1 如果增加workerCount失败，则重新获取ctl的值
                c = ctl.get();
                // 3.2 如果当前的运行状态不等于第一个循环刚开始时的状态，说明状态已被改变，返回第一个for循环继续执行
                if (runStateOf(c) != rs)
                    continue retry;
                // else CAS failed due to workerCount change; retry inner loop
            }
        }
        // 4.线程放进workers：加锁把线程放进workers，workers是一个HashSet，用于保存所有工作线程
        // 代码执行到这里  只是通过不停的循环通过CAS获取到执行任务的“锁”
        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
            // 根据firstTask来创建Worker对象
            w = new Worker(firstTask);
            // 每一个Worker对象都会创建一个线程
            final Thread t = w.thread;
            if (t != null) {
                // 获取锁。mainLock是ThreadPoolExecutor的可重入锁字段，用于保证线程安全
                final ReentrantLock mainLock = this.mainLock;
                // 加锁
                mainLock.lock();
                try {
                    int rs = runStateOf(ctl.get());
                    /*
                     * rs < SHUTDOWN表示是RUNNING状态；
                     * 如果rs是RUNNING状态或者rs是SHUTDOWN状态并且firstTask为null，向线程池中添加线程。
                     * 因为在SHUTDOWN时不会在添加新的任务，但还是会执行workQueue中的任务
                     */
                    if (rs < SHUTDOWN ||
                            (rs == SHUTDOWN && firstTask == null)) {
                        // 校验线程是否存活，不存活直接抛出异常
                        if (t.isAlive())
                            throw new IllegalThreadStateException();
                        // workers是一个HashSet，用于保存所有工作线程
                        workers.add(w);
                        int s = workers.size();
                        // largestPoolSize记录着线程池中出现过的最大线程数量
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        workerAdded = true;
                    }
                } finally {
                    // 解锁
                    mainLock.unlock();
                }
                // 5.启动线程：如果成功加进workers，调用线程的.start()方法，启动线程
                if (workerAdded) {
                    // 启动线程  这个t.start()实际是调用的Worker类中的run方法，Worker本身实现了Runnable接口，所以一个Worker类型的对象也是一个线程。
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }

/**
     * 执行任务：由Worker类的run()调用的唯一一行方法
     * @param w Worker对象
     */
    final void runWorker(Worker w) {
        Thread wt = Thread.currentThread();
        // 1.获取Worker中的runnable任务，拿到后Worker置空、解锁
        // 拿到当前Worker的第一个任务(如果携带的话)
        Runnable task = w.firstTask;
        // 置空
        w.firstTask = null;
        // 解锁
        w.unlock();
        boolean completedAbruptly = true;
        try {
            // 2.如果任务不等于空，则加锁、执行任务、解锁
            while (task != null || (task = getTask()) != null) {
                // 加锁
                w.lock();
                // 如果线程池状态 >= STOP，确保线程中断了
                if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) &&
                        !wt.isInterrupted())
                    wt.interrupt();
                try {
                    // 在任务执行前做一些操作：这个方法内为空，需要我们继承ThreadPoolExecutor重写这个方法
                    beforeExecute(wt, task);
                    Throwable thrown = null;
                    try {
                        // 任务执行：调用线程的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 {
                        // 任务执行后一些操作：这个方法内为空，需要我们继承ThreadPoolExecutor重写这个方法
                        afterExecute(task, thrown);
                    }
                } finally {
                    // 任务置空
                    task = null;
                    // 执行任务+1
                    w.completedTasks++;
                    // 解锁
                    w.unlock();
                }
            }
            completedAbruptly = false;
        } finally {
            // 删除线程的方法
            processWorkerExit(w, completedAbruptly);
        }
    }

/**
     * 获取任务
     * @return {@link Runnable }
     */
    private Runnable getTask() {
        // 声明超时变量：表示上次从阻塞队列中取任务时是否超时，先标记为未超时
        boolean timedOut = false;
        // 1.死循环拿线程对象
        for (;;) {
            // 1.1 校验当前线程池状态
            // 拿到当前的ctl
            int c = ctl.get();
            // 获取当前的线程池状态
            int rs = runStateOf(c);
            // 如果线程池状态是STOP，没有必要处理阻塞队列任务，直接返回null
            // 如果线程池状态是SHUTDOWN，并且阻塞队列是空的，直接返回null
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
                // 线程池中的线程个数减一
                decrementWorkerCount();
                return null;
            }
            // 当前线程池中线程个数
            int wc = workerCountOf(c);
            // 1.2 判断线程是否能超时：如果核心线程允许超时，或者当前线程池中的线程个数大于核心线程数，那么这个线程就有寿命
            // allowCoreThreadTimeOut:是否允许核心线程数超时（开启这个之后），核心线程数也会执行下面超时的逻辑
            // wc > corePoolSize:当前线程池中的线程个数大于核心线程数
            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
            // 1.3 如果线程超时，则销毁线程
            // wc > maximumPoolSize:基本不存在
            // timed && timedOut:第一次肯定是失败的(超时标记为false)
            if ((wc > maximumPoolSize || (timed && timedOut))
                    // 1、线程个数为1
                    // 2、阻塞队列是空的
                    && (wc > 1 || workQueue.isEmpty())) {
                // 线程池的线程个数减一
                if (compareAndDecrementWorkerCount(c)){
                    return null;
                }
                continue;
            }
            // 运行到这里，说明线程没超时
            try {
                // 1.4 从阻塞队列取出线程：当队列没线程时，如果线程允许超时则返回空，如果不允许则阻塞获取
                // poll()：从队列中获取并移除一个任务。如果队列为空，它会返回 null
                // take()：从队列中获取并移除一个任务。如果队列为空，它会阻塞当前线程直到有任务可用为止。
                Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take();
                if (r != null){
                    return r;
                }
                // 到这里，说明没从阻塞队列拿出线程，线程超时了
                timedOut = true;
            } catch (InterruptedException retry) {
                timedOut = false;
            }
        }
    }

/**
     * 关闭线程池
     */
    public void shutdown() {
        // 1.加锁
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            // 2.校验关闭权限：校验调用者是否有权限关闭线程池及其中的所有线程
            checkShutdownAccess();
            // 3.修改线程池状态：将线程池状态修改为SHUTDOWN
            advanceRunState(SHUTDOWN);
            // 4.中断空闲线程：遍历线程池workers中的工作线程，并尝试中断那些处于空闲状态的线程。
            interruptIdleWorkers();
            // 5.钩子函数，实际内容为空，用户继承重写可以编写关闭线程池后要执行的逻辑
            onShutdown(); // hook for ScheduledThreadPoolExecutor
        } finally {
            mainLock.unlock();
        }
        // 6.查看当前线程池是否可以变为TERMINATED状态
        // 从 SHUTDOWN 状态修改为 TIDYING，在修改为 TERMINATED
        tryTerminate();
    }

/**
     * 校验关闭权限：校验调用者是否有权限关闭线程池及其中的所有线程
     */
    private void checkShutdownAccess() {
        // 1. 获取系统的安全管理器
        SecurityManager security = System.getSecurityManager();
        // 2. 如果有安全管理器，需要进行权限检查
        if (security != null) {
            // 3. 首先检查调用者是否有权限关闭线程池（shutdownPerm 是一种特殊的权限）
            security.checkPermission(shutdownPerm);
            // 4. 获取线程池的锁，确保操作的线程安全
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                // 5. 遍历线程池中的每一个工作线程，检查调用者是否有权限中断每个线程
                for (Worker w : workers)
                    security.checkAccess(w.thread);
            } finally {
                // 6. 最后解锁，确保锁的释放以防止死锁
                mainLock.unlock();
            }
        }
    }

/**
     * 修改线程池状态
     * @param targetState 目标状态
     */
    private void advanceRunState(int targetState) {
        // 1.进入直接死循环
        for (;;) {
            // 2.拿到当前的ctl
            int c = ctl.get();
            // 3.修改状态：如果当前状态已经大于等于目标状态，则跳出循环，否则CAS修改线程池状态
            if (runStateAtLeast(c, targetState) || ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))
                break;
        }
    }

/**
     * 中断空闲线程：遍历线程池workers中的工作线程，并尝试中断那些处于空闲状态的线程。
     * 不传参数时，默认参数是false，即中断所有线程，而不是只中断完一个线程就完事
     */
    private void interruptIdleWorkers() {
        interruptIdleWorkers(false);
    }
    /**
     * 中断空闲线程：遍历线程池workers中的工作线程，并尝试中断那些处于空闲状态的线程。
     * @param onlyOne 是否只中断一个：如果参数 onlyOne 为 true，中断一个线程后就直接跳出循环。如果为 false，则继续尝试中断所有空闲线程。
     */
    private void interruptIdleWorkers(boolean onlyOne) {
        // 1.加锁
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            // 2.遍历线程池workers中的工作线程，并尝试中断那些处于空闲状态的线程。
            for (java.util.concurrent.ThreadPoolExecutor.Worker w : workers) {
                Thread t = w.thread;
                if (!t.isInterrupted() && w.tryLock()) {
                    try {
                        t.interrupt();
                    } catch (SecurityException ignore) {
                    } finally {
                        w.unlock();
                    }
                }
                // 如果只中断一个，则中断后直接退出循环
                if (onlyOne)
                    break;
            }
        } finally {
            mainLock.unlock();
        }
    }

/**
     * 尝试设置线程池为TERMINATED状态
     */
    final void tryTerminate() {
        // 1.开始死循环；
        for (;;) {
            // 拿到ctl
            int c = ctl.get();
            // 加锁
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                // 2.尝试CAS将线程池状态设为TIDYING，如果失败则解锁再次循环，如果成功则继续执行：
                if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                    try {
                        // 2.1 执行销毁前方法：根据用户自定义的逻辑，在线程池销毁之前做最后的处理
                        // 该方法是一个钩子函数，是个空方法，允许我们通过继承重写的方式，自己定义该方法逻辑
                        terminated();
                    } finally {
                        // 2.2 线程池状态设为TERMINATED
                        ctl.set(ctlOf(TERMINATED, 0));
                        // 2.3 唤醒所有等待线程池终止的线程
                        // termination是mainLock的 Condition 对象 ，这是与 ReentrantLock 相关联的条件变量，用于线程间的通信。
                        // Condition对象：用于线程间通信，通过await()和signal(),signalAll()让线程等待或唤醒。通常用lock锁创建Condition对象，即lock.newCondition();
                        termination.signalAll();
                    }
                    return;
                }
            } finally {
                // 解锁
                mainLock.unlock();
            }
        }
    }

/**
     * 终止线程池
     * @return {@link List }<{@link Runnable }> 被删除的所有线程
     */
    public List<Runnable> shutdownNow() {
        // 声明tasks，用于收集所有被删除的线程
        List<Runnable> tasks;
        // 1.加锁
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            // 2.校验关闭权限：校验调用者是否有权限关闭线程池及其中的所有线程
            checkShutdownAccess();
            // 3.修改线程池状态：将线程池状态修改为STOP
            advanceRunState(STOP);
            // 4.中断所有线程：将线程池中的线程全部中断
            interruptWorkers();
            // 5.删除所有线程：删除阻塞队列中所有的工作线程，并将这些删除线程收集到tasks
            // drain以为排空
            tasks = drainQueue();
        } finally {
            // 6.解锁
            mainLock.unlock();
        }
        // 7.尝试设置线程池为TERMINATED状态
        // 从 Stop 状态修改为 TIDYING，在修改为 TERMINATED
        tryTerminate();
        return tasks;
    }

/**
     * 清空阻塞队列：删除阻塞队列中所有线程，并返回这些线程
     * @return {@link List }<{@link Runnable }>
     */
    private List<Runnable> drainQueue() {
        // 1.获取阻塞队列
        BlockingQueue<Runnable> q = workQueue;
        // 返回的结果
        ArrayList<Runnable> taskList = new ArrayList<Runnable>();
        // 2.清空阻塞队列并将数据放入taskList中。
        // drainTo 方法是 BlockingQueue 的批量删除方法，它可以将队列中的所有元素批量转移到另一个集合中。
        q.drainTo(taskList);
        // 3.再次遍历清空：因为没加锁，所以需要再次遍历，防止依赖
        if (!q.isEmpty()) {
            for (Runnable r : q.toArray(new Runnable[0])) {
                if (q.remove(r))
                    taskList.add(r);
            }
        }
        // 最终返回即可
        return taskList;
    }

ExecutorService executorService = Executors.newFixedThreadPool(10);
// 源码
// 固定大小线程池: 
new ThreadExcutor(n, n, 0L, ms, new LinkedBlockingQueue<Runable>()
// 缓存线程池: 
new ThreadExcutor(0, max_valuem, 60L, s, new SynchronousQueue<Runnable>());
// 定时任务线程池: 
new ScheduledThreadPoolExecutor(corePoolSize);
// super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
              new DelayedWorkQueue());
// 单线程线程池: 
new ThreadExcutor(1, 1, 0L, ms, new LinkedBlockingQueue<Runable>())

public class Executors {
    /**
     * 创建固定大小的线程池
     *
     * @param nThreads the number of threads in the pool
     * @return the newly created thread pool
     * @throws IllegalArgumentException if {@code nThreads <= 0}
     */
    public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }
...
}

public class Executors {
    /**
     * 创建固定大小的线程池
     *
     * @param nThreads the number of threads in the pool
     * @return the newly created thread pool
     * @throws IllegalArgumentException if {@code nThreads <= 0}
     */
    public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }
    // ...
}

public class Executors {
    /**
     * 缓存线程池
     * @return the newly created thread pool
     */
    public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }
    // ...
}

public class Executors {
    /**
     * 定时任务线程池：可以定期或设定多久后执行任务
     */
    public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
        return new ScheduledThreadPoolExecutor(corePoolSize);
    }
    // ...
}

public class ScheduledThreadPoolExecutor
        extends ThreadPoolExecutor
        implements ScheduledExecutorService {
    /**
     * Creates a new {@code ScheduledThreadPoolExecutor} with the
     * given core pool size.
     *
     * @param corePoolSize the number of threads to keep in the pool, even
     *        if they are idle, unless {@code allowCoreThreadTimeOut} is set
     * @throws IllegalArgumentException if {@code corePoolSize < 0}
     */
    public ScheduledThreadPoolExecutor(int corePoolSize) {
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
              new DelayedWorkQueue());
    }
    // ...
}

public class Executors {
    /**
     * 单线程线程池：仅有一个核心线程和无界队列
     */
    public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }
    // ...
}