Java由浅入深理解线程池设计和原理2

/**
 * @param corePoolSize 池中要保留的线程数
 * @param maximumPoolSize 中允许的最大线程数,前提是队列先满
 * @param keepAliveTime 当线程数大于核心，这是多余空闲线程的最长时间将在终止之前等待新任务。
 * @param keepAliveTime参数的时间单位
 * @param workQueue 用于保存任务的队列
 * @param threadFactory 执行器创建新线程的工厂
 * @param handler 阻止执行时要使用的处理程序,因为达到了线程边界和队列容量
 */
public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {

//任务提交阶段：（4个if条件路线）
public void execute(Runnable command) {
  if (command == null)
            throw new NullPointerException();
  int c = ctl.get();
  //判断工作数，如果小于coreSize，addWork，注意第二个参数core=true
  if (workerCountOf(c) < corePoolSize) {
      if (addWorker(command, true))
          return;
      c = ctl.get();
  }
  //否则，如果线程池还在运行，offer到队列
  if (isRunning(c) && workQueue.offer(command)) {
      //再检查一下状态
      int recheck = ctl.get();
      //如果线程池已经终止，直接移除任务，不再响应
      if (! isRunning(recheck) && remove(command))
          reject(command);
      //否则，如果没有可用线程的话（比如coreSize=0），创建一个空work
        //该work创建时不会给指派任务（为null），但是会被放入works集合，进而从队列获取任务去执行
      else if (workerCountOf(recheck) == 0)
          addWorker(null, false);
  }
  //队列也满，继续调addWork，但是注意，core=false，开启到maxSize的大门
  //超出max的话，addWork会返回false，进入reject
  else if (!addWorker(command, false))
      reject(command);
}
//线程创建
private boolean addWorker(Runnable firstTask, boolean core) {
    //第一步，计数判断，不符合条件打回false
    retry:
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);
        // Check if queue empty only if necessary.
        for (;;) {
            int wc = workerCountOf(c);
            //判断线程数，注意这里！
            //也就说明线程池的线程数是不可能设置任意大的。
            //最大29位（CAPACITY=29位二进制）
            //超出规定范围，返回false，表示不允许再开启新工作线程，创建worker失败！
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            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
        }
    }
    //第二步，创建新work放入线程集合works（一个HashSet）
    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        //符合条件，创建新的work并包装task
        w = new Worker(firstTask);
        final Thread t = w.thread;
        if (t != null) {
            final ReentrantLock mainLock = this.mainLock;
            //加锁，workers是一个hashset，这里要保障线程安全性
            mainLock.lock();
            try {   
                        //...
                    //在这里！！！
                    workers.add(w);
                    //...                    
                    workerAdded = true;
            } finally {
                mainLock.unlock();
            }
            if (workerAdded) {
                //注意，只要是成功add了新的work，那么将该新work立即启动，任务得到执行
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}
//任务获取与执行
//在worker执行runWorker()的时候，不停循环，先查看自己有没有携带Task，如果有，执行
while (task != null || (task = getTask()) != null)
//如果没用，会调用getTask，从队列获取任务
private Runnable getTask() {
    boolean timedOut = false; // Did the last poll() time out?
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);
        // ...
        int wc = workerCountOf(c);
        // Are workers subject to culling? - 很形象，要不要乖乖的被“捕杀”？
        //判断是不是要超时处理，重点！！！决定了当前线程要不要被释放
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
                //线程数超出max，并且上次循环中poll等待超时了，那么说明该线程已终止
        //将线程队列数量原子性减
        if ((wc > maximumPoolSize || (timed && timedOut))
            && (wc > 1 || workQueue.isEmpty())) {
            //计数器做原子递减，递减成功后，返回null，for被中止
            if (compareAndDecrementWorkerCount(c))
                return null;
            //递减失败，继续下一轮循环，直到成功
            continue;
        }
        try {
            //重点！！！
            //如果线程可被释放，那就poll，释放的时间为：keepAliveTime
            //否则，线程是不会被释放的，take一直被阻塞在这里，直到来了新任务继续工作
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
            if (r != null)
                return r;
            //到这里说明可被释放的线程等待超时，已经销毁，设置该标记，下次循环将线程数减少
            timedOut = true;
        } catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}

//work.runWork():
while (task != null || (task = getTask()) != null)
//work.getTask():
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :workQueue.take();

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {}