如何进行ThreadPoolExecutor 源码解析

ThreadPoolExecutor 默认线程池 
ScheduledThreadPoolExecutor 定时线程池

//在将来的某个时间执行给定的任务。任务可以是新起一个新线程或者复用现有池线程中的线程去执行
public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        /*
         * 执行过程还是分为 3 步:
         *
         * 1.执行任务： 
         * 如果小于核心线程数，尝试创建一个新线程来执行给定的任务。
         * 方法 addWorker() 就是真正的创建一个新线程来执行任务的方法。
         * addWorker()方法会对 runState 和 workerCount进行原子检查。
         * addWorker()方法会返回一个 boolean 值，通过返回 false 值来防止在不应该添加线程的情况下发出错误警报
         * 
         *
         * 2.添加到阻塞队列：
         * 未能满足条件执行完步骤 1 则添加到阻塞队列。
         * 如果任务可以成功排队，会再次进行检查，检查是否应该添加线程（因为现有线程自上次检查后就死了），
         * 或者自进入此方法以来该池已关闭。因此，需要重新检查状态，并在停止的情况下在必要时回滚队列，如果没有，则启动一个新线程。 
         *
         * 3.拒绝任务： 
         * 如果无法将任务添加至阻塞队列，最大线程数也未达到最大会尝试添加一个新的线程。如果失败，说明线程池已关闭或处于饱和状态，因此拒绝该任务。
         */
         
         //clt记录着runState和workerCount
        int c = ctl.get();
        /*
         * workerCountOf方法取出低29位的值，表示当前活动的线程数;
         * 如果当前活动线程数小于corePoolSize，则新建一个线程放入线程池中;并把任务添加到该线程中。
         */
        if (workerCountOf(c) < corePoolSize) {
            /*
             * addWorker中的第二个参数表示限制添加线程的数量是根据corePoolSize来判断还是maximumPoolSize来判断
             * 如果为true，根据corePoolSize来判断;
             * 如果为false，则根据maximumPoolSize来判断
             */
            if (addWorker(command, true))
                return;
            //如果添加失败，则重新获取ctl值    
            c = ctl.get();
        }
        //执行到此处说明从核心线程里给当前任务分配线程失败
        //如果当前线程池是运行状态并且任务添加到队列成功
        if (isRunning(c) && workQueue.offer(command)) {
            //重新获取ctl值。即使添加队列成功也要再次检查，如果不是运行状态，由于之前已经把任务添加到workerQueue 中了，所以要移除该任务,执行过后通过handler使用拒绝策略对该任务进行处理，整个方法返回
            int recheck = ctl.get();
            if (! isRunning(recheck) && remove(command))
                reject(command);
                /*
                 * 获取线程池中的有效线程数，如果数量是0，则执行addWorker方法 这里传入的参数表示:
                 * 第一个参数为null，表示在线程池中创建一个线程，但不去启动;
                 * 第二个参数为false，将线程池的有限线程数量的上限设置为maximumPolSize，添加线程时根据maximumPoolSize来判断;
                 * 如果判断workerCount大于0，则直接返回，在workQueue中新增的comman 会在将来的某个时刻被执行。
                 */
            //因为 任务已经被添加到workQueue中了，所以worker在执行的时候，会直接从workQueue中 获取任务。     
            else if (workerCountOf(recheck) == 0)
                //执行到这里说明任务已经添加到阻塞队列里了，最大线程数也未饱和，则创建一个新的线程去阻塞队列里拿任务
                //这步操作也就是创建一个线程，但并没有传入任务，因为任务已经被添加到workQueue中了，所以worker在执行的时候，会直接从workQueue中 获取任务。
                //为什么要这样做呢？是为了保证线程池在RUNNING状态下必须要有一个线程来执行任务。
                addWorker(null, false);
        }
        /*
         * 如果执行到这里，有两种情况:
         * 1. 线程池已经不是RUNNING状态;
         * 2. 线程池是RUNNING状态，但workerCount >= corePoolSize并且workQueue已满。
         * 这时，再次调用addWorker方法，但第二个参数传入为false，将线程池的 有限线程数量的上限设置为maximumPoolSize;
         * 如果失败则拒绝该任务
         */
        else if (!addWorker(command, false))
            reject(command);
    }

/**
 * 检查是否可以根据当前线程池的状态添加一个新的工作线程去执行任务。
 * addWorker（runnable,true）表示从核心工作线程数中分配线程执行传进来的任务；
 * addWorker（null,false）表示从最大线程数中分配线程执行阻塞队列中的任务。
 * 线程池如果停止或者关闭则直接返回 false，如果线程池创建新线程失败同样也会返回 false。
 * 如果创建线程失败，或者线程工厂返回 null，或者执行当前 addWorker()的线程抛出异常，（注意是当前线程抛出异常，当前线程抛出异常只与当前任务有关，并不影响其他任务的执行），线程池的相关属性会立即回滚
 */
private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    // 外层 for 循环就是为了能给任务分配线程做准备，判断状态--> 原子递增 workerCount
    // 直到线程池状态不符合条件返回 false ，或者自增成功跳出 for 循环
    // 同样的，getTask()从阻塞队列中获取任务的时候也是这么个逻辑，先对 workerCount 原子递减，再去执行任务
    for (;;) {
        //可以看到，每一步操作都会对线程池的状态参数做判断
        int c = ctl.get();
        int rs = runStateOf(c);
        
        //也是对线程池状态，队列状态做检查
        /**
         * 这里的状态判断也很好理解：
         * 线程池状态为SHUTDOWN，不会再接受新的任务了，返回 false
         * 想城池状态不为SHUTDOWN，传进来的任务为空，并且阻塞队列里也没任务，那还执行个锤子任务，同样返回 false 
         */
        if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
               firstTask == null &&
               ! workQueue.isEmpty()))
            return false;

        //前面已经判断过满足为任务分配一个线程去执行任务
        //这个 for 循环就是为了创建任务做准备，先去原子性的递增 workerCount,workerCount 递增成功了才会去真正的为任务分配线程去执行
        for (;;) {
            //当前工作线程数
            int wc = workerCountOf(c);
            //当前工作线程数大于corePoolSize 或者 maximumPoolSize （跟谁比较就是根据传进来的参数 core 判断），
            //说明也没有分配的线程可以执行任务了，返回 false
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            /** 
             * 执行到这里说明满足条件了，可以分配出来线程去执行任务了
             * 尝试增加workerCount，如果成功，则跳出第一个for循环
             * 这里是进行 CAS 自增 ctl 的 workerCount（先把数量自增，再跳出 for 循环创建新的线程去执行任务）
             * 该方法内部也是调用了原子类 AtomicInteger.compareAndSet()方法，保证原子递增
             */    
            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
        }
    }

    //跳出 for 循环之后说明线程池的工作线程数 workerCount 已经调节过了，接下来要做到就是真正的分配线程，执行任务
    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        //Worker 对象是个内部类，其实就是用threatFactory 生成一个新的线程
        //继承 AQS 类，实现Runable 接口，重写 run()方法，重写的 run()方法也很重要，后面会讲
        w = new Worker(firstTask);
        final Thread t = w.thread;
        if (t != null) {
            //加锁保证同步
            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 是个 HashSet，线程池也是通过维护这个 workers 控制任务的执行
                    workers.add(w);
                    int s = workers.size();
                    if (s > largestPoolSize)
                        //largestPoolSize记录着线程池中出现过的最大线程数量
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
            if (workerAdded) {
                //终于，调用线程的 start() 方法
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        //如果创建线程失败，就要回滚线程池的状态了
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}

  /**
    * 用于判断线程是否空闲以及是否可以被中断
    */
   protected boolean tryAcquire(int unused) {
           //cas 修改状态，不可重入
            if (compareAndSetState(0, 1)) {
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            return false;
        }

/**
 * Worker 类实现 Runnable 接口，重写的 run()方法
 */
final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    //允许中断
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {
        //while 循环就是不断的去执行任务，当自己的任务（firstTask）执行完之后依然会从阻塞队列里拿任务去执行，就这样的操作保证了线程的重用
        //task 为空则从阻塞队列中获取任务
        while (task != null || (task = getTask()) != null) {
            w.lock();
            // 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
            /**
             * 如果线程池正在停止，那么要保证当前线程是中断状态,如果不是的话，则要保证当前线程不是中断状态
             * 这里为什么要这么做呢？考虑在执行该if语句期间可能也执行了shutdownNow方法，shutdownNow方法会 把状态设置为STOP，
             * 回顾一下STOP状态:不能接受新任务，也不处理队列中的任务，会中断正在处理任务的线程。在线程池处于 RUNNING 或 SHUTDOWN 状态时，
             * 调用 shutdownNow() 方法会使线程池进入到STOP状态。
             * STOP状态要中断线程池中的所有线程，而这里使用Thread.interrupted()来判断是否中断是为了确保在 RUNNING或者SHUTDOWN状态时线程是非中断状态的，
             * 因为Thread.interrupted()方法会重置中断的状态。
             */
            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);
    }
}

/**
 * 从阻塞队列中获取任务，返回值是 Runnable
 * 线程池状态不满足执行条件时直接返回 null
 */
private Runnable getTask() {
    //timeOut变量的值表示上次从阻塞队列中取任务时是否超时
    boolean timedOut = false; // Did the last poll() time out?
    
    //这里两个 for 循环操作和 addWorker() 方法里的两个 for 循环操作思想一样
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);
        
        //仍然检查线程池状态
        // Check if queue empty only if necessary.
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            decrementWorkerCount();
            return null;
        }

        int wc = workerCountOf(c);

        //timed变量用于判断是否需要进行超时控制
        //allowCoreThreadTimeOut默认是false，也就是核心线程不允许进行超时
        //wc > corePoolSize，表示当前线程池中的线程数量大于核心线程数量
        //对于超过核心线程数量的这些线程，需要进行超时控制
        // Are workers subject to culling?
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

        //和 addWorker() 里流程一样，也是先对线程池中 workerCount 进行控制，再进行后面的执行任务操作
        //满足条件则 workerCount 数量减一
        if ((wc > maximumPoolSize || (timed && timedOut))
            && (wc > 1 || workQueue.isEmpty())) {
            if (compareAndDecrementWorkerCount(c))
                return null;
            continue;
        }

        try {
            //根据timed来判断，如果为true，则通过阻塞队列的poll方法进行超时控制，如果在keepAliveTime时间内没有获取到任务，则返回null
            //否则通过take方法，如果这时队列为空，则take方法会阻塞直到队列不为空
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
            if (r != null)
                return r;
            // 如果 r == null，说明已经超时，timedOut设置为true
            timedOut = true;
        } catch (InterruptedException retry) {
            // 如果获取任务时当前线程发生了中断，则设置timedOut为false并返回
            timedOut = false;
        }
    }
}
processWorkerExit() 
/**
 * getTask方法返回null时，在runWorker方法中会跳出while循环，然后会执行 processWorkerExit方法。
 * 做线程池的善后工作
 */
private void processWorkerExit(Worker w, boolean completedAbruptly) {
    //如果completedAbruptly值为true，则说明线程执行时出现了异常,需要将workerCount减1
    //如果线程执行时没有出现异常，说明在getTask()方法中已经已经对workerCount减1了，这里就不需要再减了
    if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
        decrementWorkerCount();

    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        //统计完成的任务数
        completedTaskCount += w.completedTasks;
        // 从workers中移除，也就表示着从线程池中移除了一个工作线程
        // workers 是前面提到的 HashSet，线程池就是通过维护这个 worker()来保证线程池运作的 
        workers.remove(w);
    } finally {
        mainLock.unlock();
    }
    // 根据线程池状态进行判断是否结束线程池
    tryTerminate();


    /**
     * 当线程池是RUNNING或SHUTDOWN状态时，如果worker是异常结束，那么会直接addWorker;
     * 如果allowCoreThreadTimeOut=true，并且等待队列有任务，至少保留一个worker
     * 如果allowCoreThreadTimeOut=false，workerCount不少于corePoolSize
     */
    int c = ctl.get();
    if (runStateLessThan(c, STOP)) {
        if (!completedAbruptly) {
            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
            if (min == 0 && ! workQueue.isEmpty())
                min = 1;
            if (workerCountOf(c) >= min)
                return; // replacement not needed
        }
        addWorker(null, false);
    }
}