【Java 线程池】【二】ThreadPoolExecutor 参数详解及拒绝策略

public class ThreadPoolExecutor extends AbstractExecutorService {
    // 使用一个Int类型的32位同时表示: 线程池状态、线程数量
    // 其中高3位表示线程池状态，也就是0~2位表示线程池状态
    // 低29位也就是3~31位表示线程数量，最多能容纳2^29 - 1 约为500多万个线程
    private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
    // Integer.SIZE = 32, COUNT_BITS = 32 - 3 = 29
    // 这里的意思就是使用29位表示线程数量
    private static final int COUNT_BITS = Integer.SIZE - 3;
    // 线程池的容量为 CAPACITY = 2^29 - 1 约为500多万，（1 << COUNT_BITS也就是1左移29位也就是2的29次方）
    // 这里CAPACITY对应的32二进制位为 000 11111111111111111111111111111（高3位全为0，低29位全为1）
    private static final int CAPACITY = (1 << COUNT_BITS) - 1;
    
    // runState is stored in the high-order bits
    // -1 左移29位得到 111 00000000000000000000000000000
    // 其中高3位也就是111表示当前线程池为运行状态RUNNING
    private static final int RUNNING = -1 << COUNT_BITS;
    // 0 左移动29位得到 000 00000000000000000000000000000
    // 其中高3位为000表示当前线程池为状态为SHUTDOWN
    private static final int SHUTDOWN = 0 << COUNT_BITS;
    // 1 左移动29位得到 001 00000000000000000000000000000
    // 其中高3位为001表示当前线程池为状态为STOP
    private static final int STOP = 1 << COUNT_BITS;
    // 2 左移动29位得到 010 00000000000000000000000000000
    // 其中高3位为010表示当前线程池为状态为TIDYING
    private static final int TIDYING = 2 << COUNT_BITS;
    // 3 左移动29位得到 011 00000000000000000000000000000
    // 其中高3位为011表示当前线程池为状态为TERMINATED
    private static final int TERMINATED = 3 << COUNT_BITS;
}

private static int runStateOf(int c)     {
    // 1. 这里的c就是传入ctl的值
    // 2. ~CAPACITY 的32位值是 111 00000000000000000000000000000（高3位为1，低29位为0）
    // 这样c & ~ CAPACITY操作就能保留c的高三位值
    // 由于使用ctl的高3位表示线程池状态，这样也就得到了线程池的状态
    return c & ~CAPACITY;
}

private static int workerCountOf(int c)  {
    // CAPACITY的32位为 000 11111111111111111111111111111（高3位全为0，低29位全为1）
    // c与CAPACITY进行与运算，会得到低29位结果，也就是得到当前线程数量
    return c & CAPACITY;
}

private static int ctlOf(int rs, int wc) {
    // rc表示线程池状态（RUNNING、SHUTDOWN、STOP、TIDYING、TERMINATED）
    // wc表示当前线程池大小
    // rc | wc 进行或操作就可以得到ctl的值（相当于得到一个新的数组，高3位就是rc的值，低29位是wc的值）
    return rs | wc;
}

// 核心线程数
private volatile int corePoolSize;
// 最大线程池数
private volatile int maximumPoolSize;
// 创建线程的工厂
private volatile ThreadFactory threadFactory;
// 阻塞队列，用户存放任务的容器
private final BlockingQueue<Runnable> workQueue;
// 线程容器，Worker就是工作者，一个Worker内部封装了一个Thread线程
// 这里使用HashSet进行存储，表示工作者容器，线程容器
private final HashSet<Worker> workers = new HashSet<Worker>();
// 互斥锁，由于上面的HashSet不是并发安全的，所以操作的时候肯定需要上锁
private final ReentrantLock mainLock = new ReentrantLock();
// 等待条件
private final Condition termination = mainLock.newCondition();
// 当前线程数 > corePoolSize时候，这部分多出的线程在空闲多久之后会被销毁掉
private volatile long keepAliveTime;
// 是否允许核心线程超时
// 如果是false，核心线程不会被销毁
// 如果是true，核心线程在keepAliveTime时间后依然空闲，则会被销毁掉
private volatile boolean allowCoreThreadTimeOut;
// 一个标记变量，标记曾经线程池的最大大小
// 举例：corePoolSize = 5, maximumPoolSize = 10
// 线程池最大的时候有过8个线程，那么largestPoolSize = 8，只是标记一些曾经的最大值
private int largestPoolSize;
// 一个统计变量，线程池已经完成的任务数量
private long completedTaskCount;
// 线程池拒绝策略
// 当线程池饱和（阻塞队列满了，存不进新的任务；同时线程池的数量达到了maximumPoolSize无法再增加新线程）的时候提交新任务会触发此策略的执行
// 当线程池状态为SHUTDOWN，表示我要关闭了，不再接收新任务；此时向线程池提交新任务也会触发此策略的执行
private volatile RejectedExecutionHandler handler;

// 默认的拒绝策略，直接抛出异常（Abort是中止的意思）
private static final RejectedExecutionHandler defaultHandler =
    new AbortPolicy();

final void reject(Runnable command) {
    // reject方法是线程池执行拒绝策略的入口
    // 这里的handler就是拒绝策略
    // rejectedExecution方法就是拒绝策略内部的逻辑
    handler.rejectedExecution(command, this);
}

public static class AbortPolicy implements RejectedExecutionHandler {
    
    public AbortPolicy() { }

    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        // 这里直接抛出一个异常，也就是线程池饱和或者线程池SHUTDOWN的时候直接抛出异常
        throw new RejectedExecutionException("Task " + r.toString() +
                                             " rejected from " +
                                             e.toString());
    }
}

public static class DiscardPolicy implements RejectedExecutionHandler {

    public DiscardPolicy() { }
    // r是最新提交的任务
    // 看这里什么都不干，就是将最新提交的任务忽略了，不会执行它，相当于将它丢弃了
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    }
}

public static class DiscardOldestPolicy implements RejectedExecutionHandler {

    public DiscardOldestPolicy() { }

    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        // 如果当前线程池未关闭，也就是正常运行的
        if (!e.isShutdown()) {
            // 从阻塞队列中poll出一个任务，按照先进先出的方式
            // 也就是当前阻塞队列头部的任务，是这个阻塞队列中最先进去的，最老的意思
            e.getQueue().poll();
            // 抛弃最老的任务之后，调用e.execute(r)将当前任务交给线程池
            e.execute(r);
        }
    }

public static class CallerRunsPolicy implements RejectedExecutionHandler {
  
    public CallerRunsPolicy() { }

    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        // 如果当前线程池状态不是SHUTDOWN，也就是正常的
        if (!e.isShutdown()) {
            // 直接执行任务的run方法
            // 注意，这个时候不是使用线程池里面的线程，是直接使用你当前的线程去执行
            r.run();
        }
    }
}