JAVA线程池源码全解析

//高3位：表示当前线程池运行状态 除去高3位之后的低位：
    // 表示当前线程池所拥有的线程数量
    private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
    // 表示在ctl中，低COUNT_BITS位 用于存放当前线程数量的位
    private static final int COUNT_BITS = Integer.SIZE - 3;
    //低COUNT_BITS位 所能表达的最大数值
    private static final int CAPACITY = (1 << COUNT_BITS) - 1;

    //表示可接受新任务，且可执行队列中的任务；
    private static final int RUNNING = -1 << COUNT_BITS;
    //表示不接受新任务，但可执行队列中的任务；
    private static final int SHUTDOWN = 0 << COUNT_BITS;
    //表示不接受新任务，且不再执行队列中的任务，且中断正在执行的任务；
    private static final int STOP = 1 << COUNT_BITS;
   // 所有任务已经中止，且工作线程数量为0，最后变迁到这个状态的线程将要执行terminated()钩子方法，只会有一个线程执行这个方法；
    private static final int TIDYING = 2 << COUNT_BITS;
    //中止状态，已经执行完terminated()钩子方法；
    private static final int TERMINATED = 3 << COUNT_BITS;

    //任务队列，当线程池中的线程达到核心线程数量时，再提交任务 就会直接提交到 workQueue
    private final BlockingQueue<Runnable> workQueue;

    //线程池全局锁，增加worker 减少 worker 时需要持有mainLock ， 修改线程池运行状态时，也需要。
    private final ReentrantLock mainLock = new ReentrantLock();

    //线程池中真正存放 worker->thread 的地方。
    private final HashSet<Worker> workers = new HashSet<Worker>();

    private final Condition termination = mainLock.newCondition();

    // 记录线程池生命周期内 线程数最大值
    private int largestPoolSize;

    // 记录线程池所完成任务总数
    private long completedTaskCount;

    // 创建线程会使用线程工厂
    private volatile ThreadFactory threadFactory;
    /**
     * 拒绝策略
     */
    private volatile RejectedExecutionHandler handler;
    //空闲线程存活时间，当allowCoreThreadTimeOut == false 时，会维护核心线程数量内的线程存活，超出部分会被超时。
    //allowCoreThreadTimeOut == true 核心数量内的线程 空闲时 也会被回收。
    private volatile long keepAliveTime;
    //控制核心线程数量内的线程 是否可以被回收。true 可以，false不可以。
    private volatile boolean allowCoreThreadTimeOut;
    // 核心线程池数量
    private volatile int corePoolSize;

    // 线程池最大数量
    private volatile int maximumPoolSize;

 //Worker采用了AQS的独占模式
    //独占模式：两个重要属性  state  和  ExclusiveOwnerThread
    //state：0时表示未被占用 > 0时表示被占用   < 0 时 表示初始状态，这种情况下不能被抢锁。
    //ExclusiveOwnerThread:表示独占锁的线程。
    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;

        // worker内部封装的工作线程
        final Thread thread;
        //假设firstTask不为空，那么当worker启动后（内部的线程启动)会优先执行firstTask，当执行完firstTask后，会到queue中去获取下一个任务。
        Runnable firstTask;
        // 记录当前worker所完成的任务数量
        volatile long completedTasks;

        /**
         * Creates with given first task and thread from ThreadFactory.
         *
         * @param firstTask the first task (null if none)
         */
        Worker(Runnable firstTask) {
            // 设置AQS独占模式为初始化中的状态，这时候不能被抢占
            setState(-1); // inhibit interrupts until runWorker
            this.firstTask = firstTask;
            // 使用线程工厂创建一个线程
            this.thread = getThreadFactory().newThread(this);
        }

    public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        // 获取ctl的值
        int c = ctl.get();
        // 当前线程数小于核心线程池数量，此次提交任务，直接创建一个新的worker
        // 相对应线程池多了一个新的线程
        if (workerCountOf(c) < corePoolSize) {
            // addWorker 即为创建线程的过程，会创建worker对象，并且将command作为firstTask
            // core==true 表示采用核心线程数量限制，false采用maxinumPoolSize
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        // 执行到这里有几种情况？
        // 1.当前线程池数量已经达到corePoolSize
        // 2. addWorker失败


        // 当前线程池处于running状态,尝试将task放入到workQueue中
        if (isRunning(c) && workQueue.offer(command)) {
            // 获取dangqctl
            int recheck = ctl.get();
            // !isRunning()成功，代表当你提交到任务队列后，线程池状态被外部线程给修改，例如调用了shutDown(),shutDownNow()
            // remove成功，提交之后，线程池中的线程还没消费
            // remove 失败，说明在shutDown或者shutDown之前,就被线程池的线程给处理了
            if (!isRunning(recheck) && remove(command))
                reject(command);
                // 当前线程池是running状态，
            else if (workerCountOf(recheck) == 0)
                // 如果当前没有线程，就添加一个线程保证当前至少有一个线程存在
                addWorker(null, false);
        }
        //执行到这里，有几种情况？
        //1.offer失败
        //2.当前线程池是非running状态

        //1.offer失败，需要做什么？ 说明当前queue 满了！这个时候 如果当前线程数量尚未达到maximumPoolSize的话，会创建新的worker直接执行command
        //假设当前线程数量达到maximumPoolSize的话，这里也会失败，也走拒绝策略。

        //2.线程池状态为非running状态，这个时候因为 command != null addWorker 一定是返回false。
        else if (!addWorker(command, false))
            reject(command);
    }

1. 工作线程数量小于核心数量，创建核心线程；

2. 达到核心数量，进入任务队列；

3. 任务队列满了，创建非核心线程；

4. 达到最大数量，执行拒绝策略；
   

  //firstTask 可以为null，表示启动worker之后，worker自动到queue中获取任务.. 如果不是null，则worker优先执行firstTask
    //core 采用的线程数限制 如果为true 采用 核心线程数限制  false采用 maximumPoolSize线程数限制.
    private boolean addWorker(Runnable firstTask, boolean core) {
        // 自旋：判断当前线程池状态是否允许创建线程的事情
        retry:
        for (; ; ) {
            // 获取当前ctl值
            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);
                //判断当前线程是否超过限制，超过限制就无法创建线程
                if (wc >= CAPACITY ||
                        wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
                // 通过cas将线程数量加1，能够成功加1相当于申请到创建线程的令牌
                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 {
            // 创建work
            w = new Worker(firstTask);
            //将新创建的work节点的线程 赋值给t
            final Thread t = w.thread;
            if (t != null) {
                final ReentrantLock mainLock = this.mainLock;
                //持有全局锁，可能会阻塞，直到获取成功为止，同一时刻操纵 线程池内部相关的操作，都必须持锁。
                mainLock.lock();
                try {
                    //获取最新线程池运行状态保存到rs中
                    int rs = runStateOf(ctl.get());
                    //
                    if (rs < SHUTDOWN ||
                            (rs == SHUTDOWN && firstTask == null)) {
                        if (t.isAlive()) // precheck that t is startable
                            throw new IllegalThreadStateException();
                        //将创建的work添加到线程池中
                        workers.add(w);
                        // 获取最新当前线程池线程数量
                        int s = workers.size();
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        workerAdded = true;
                    }
                } finally {
                    // 释放锁
                    mainLock.unlock();
                }
                if (workerAdded) {
                    // 添加work成功后，将创建的线程启动
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            // 启动失败
            if (!workerStarted)
                // 释放令牌
                // 将当前worker清理出workers集合
                addWorkerFailed(w);
        }
        return workerStarted;
    }

这里取任务会根据工作线程的数量判断是使用BlockingQueue的poll(timeout, unit)方法还是take()方法。

poll(timeout, unit)方法会在超时时返回null，如果timeout<=0，队列为空时直接返回null。

take()方法会一直阻塞直到取到任务或抛出中断异常。

   public <T> Future<T> submit(Callable<T> task) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<T> ftask = newTaskFor(task);
        execute(ftask);
        return ftask;
    }
    public Future<?> submit(Runnable task) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<Void> ftask = newTaskFor(task, null);
        execute(ftask);
        return ftask;
    }