13.一文彻底了解线程池

public static void main(String[] args) {
  while (true) {
    new Thread(()-> {
    }).start();
  }
}

public interface Executor {
  
  /**
   * Executes the given command at some time in the future.  The command
   * may execute in a new thread, in a pooled thread, or in the calling
   * thread, at the discretion of the {@code Executor} implementation.
   */
  void execute(Runnable command);
}

public interface ExecutorService extends Executor {
  
  /**
   * Initiates an orderly shutdown in which previously submitted
   * tasks are executed, but no new tasks will be accepted.
   * Invocation has no additional effect if already shut down.
   */
  void shutdown();
  
  /**
   * Attempts to stop all actively executing tasks, halts the
   * processing of waiting tasks, and returns a list of the tasks
   * that were awaiting execution.
   * This method does not wait for actively executing tasks to
   * terminate.  Use {@link #awaitTermination awaitTermination} to
   * do that.
   */
  List<Runnable> shutdownNow();
  
  boolean isShutdown();
  
  boolean isTerminated();
  
  /**
   * Blocks until all tasks have completed execution after a shutdown
   * request, or the timeout occurs, or the current thread is
   * interrupted, whichever happens first.
   */
  boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException;
  
  <T> Future<T> submit(Callable<T> task);
  
  <T> Future<T> submit(Runnable task, T result);
  
  /**
   * Submits a Runnable task for execution and returns a Future
   * representing that task. The Future's {@code get} method will
   * return {@code null} upon <em>successful</em> completion.
   */
  Future<?> submit(Runnable task);
}

public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
  if (corePoolSize < 0 || maximumPoolSize <= 0 || maximumPoolSize < corePoolSize || keepAliveTime < 0) 
    throw new IllegalArgumentException();
  if (workQueue == null || threadFactory == null || handler == null) 
    throw new NullPointerException();
  
  this.corePoolSize = corePoolSize;
  this.maximumPoolSize = maximumPoolSize;
  this.workQueue = workQueue;
  this.keepAliveTime = unit.toNanos(keepAliveTime);
  this.threadFactory = threadFactory;
  this.handler = handler;
}

/**
 * The main pool control state, ctl, is an atomic integer packing
 * two conceptual fields
 *   workerCount, indicating the effective number of threads
 *   runState,    indicating whether running, shutting down etc
 */
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

private static final int COUNT_BITS = Integer.SIZE - 3; // 29
private static final int COUNT_MASK = (1 << COUNT_BITS) - 1;// 0001 1111 1111 1111 1111 1111 1111 1111

private static final int RUNNING    = -1 << COUNT_BITS;// 111 0 0000 0000 0000 0000 0000 0000 0000
private static final int SHUTDOWN   =  0 << COUNT_BITS;// 000 0 0000 0000 0000 0000 0000 0000 0000
private static final int STOP       =  1 << COUNT_BITS;// 001 0 0000 0000 0000 0000 0000 0000 0000
private static final int TIDYING    =  2 << COUNT_BITS;// 010 0 0000 0000 0000 0000 0000 0000 0000
private static final int TERMINATED =  3 << COUNT_BITS;// 011 0 0000 0000 0000 0000 0000 0000 0000

private static int runStateOf(int c)     { return c & ~COUNT_MASK; }
private static int workerCountOf(int c)  { return c & COUNT_MASK; }
private static int ctlOf(int rs, int wc) { return rs | wc; }

ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 4, 10, TimeUnit.SECONDS, new LinkedBlockingQueue<>(6));

threadPoolExecutor.submit(() -> {
  // 业务逻辑
});

public abstract class AbstractExecutorService implements ExecutorService {
  public Future<?> submit(Runnable task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<Void> ftask = newTaskFor(task, null);
    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;
  }
  
  public <T> Future<T> submit(Callable<T> task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<T> ftask = newTaskFor(task);
    execute(ftask);
    return ftask;
  }
}

public void execute(Runnable command) {
  // 检测待执行任务
  if (command == null) {
    throw new NullPointerException();
  }
  
  // 获取主控状态CTL
  int c = ctl.get();
  
  // STEP 1: 当工作线程数量小于核心线程时，执行addWorker方法
  if (workerCountOf(c) < corePoolSize) {
    if (addWorker(command, true)) {
      return;
    }
    c = ctl.get();
  }
  
  // 当工作线程数量大于核心线程数量时
  // STEP 2: 首先判断线程池是否处于运行状态，接着尝试添加到队列中
  if (isRunning(c) && workQueue.offer(command)) {
    // 再次检查线程池状态
    int recheck = ctl.get();
    // 不再处于RUNNING，则从队列中删除当前任务，并执行拒绝策略
    if (!isRunning(recheck) && remove(command)) {
      reject(command);
    } else if (workerCountOf(recheck) == 0) {
      addWorker(null, false);
    }
  }
  
  // STEP 3: 无法添加到队列时，尝试执行addWorker
  else if (!addWorker(command, false))
    // addWorker执行失败，则执行拒绝策略
    reject(command);
}

private boolean addWorker(Runnable firstTask, boolean core)

retry:
// 获取主控状态CTL
for (int c = ctl.get();;) {
  
  // 注释1
  // Java 11相对友好很多，减少了很多!的使用，看起来比较符合人的思维
  // 这部分判断可以分成两部分：
  //   1. 至少为SHUTDOWN状态
  //   2.条件3选1满足：
  //     2-1，至少为STOP状态
  //     2-2，firstTask不为空
  //     2-3，workQueue为空
  if (runStateAtLeast(c, SHUTDOWN) && (runStateAtLeast(c, STOP) || firstTask != null || workQueue.isEmpty())) {
    return false;
  }
  for (;;) {
    // core == true，保证工作线程数量小于核心线程数量
    // core == false，保证线程数量小于最大线程数量
    if (workerCountOf(c) >= ((core ? corePoolSize : maximumPoolSize) & COUNT_MASK)) {
      return false;
    }
    
    // 增加工作线程数量并退出
    if (compareAndIncrementWorkerCount(c)) {
      break retry;
    }
    
    // 如果至少是SHUTDOWN状态，则重新执行
    c = ctl.get();
    if (runStateAtLeast(c, SHUTDOWN)) {
      continue retry;
    }   
  }
}

boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
  // 创建Worker对象
  w = new Worker(firstTask);
  // 从worker对象中获取线程
  final Thread t = w.thread;
  if (t != null) {
    // 上锁
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
      int c = ctl.get();
      // 线程池状态检查
      // RUNNING状态，或者“小于”STOP状态（处理队列中的任务）
      if (isRunning(c) || (runStateLessThan(c, STOP) && firstTask == null)) {
        // 线程状态检查
        if (t.getState() != Thread.State.NEW) {
          throw new IllegalThreadStateException();
        }
        // 将Worker对象添加到workers中
        workers.add(w);
        workerAdded = true;
        int s = workers.size();
        if (s > largestPoolSize) {
          // 记录线程池中出现过的最大线程数
          largestPoolSize = s;
        }
      }
    } finally {
      mainLock.unlock();
    }
    
    if (workerAdded) {
      // 启动线程
      t.start();
      workerStarted = true;
    }
  }
} finally {
  if (! workerStarted) {
    // addWorker执行失败
    // addWorkerFailed中包含工作线程数减1的逻辑
    addWorkerFailed(w);
  }  
}
return workerStarted;

private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
  
  final Thread thread;
  
  Runnable firstTask;
  
  volatile long completedTasks;
  
  Worker(Runnable firstTask) {
    setState(-1);
    this.firstTask = firstTask;
    // 通过默认线程工厂创建线程
    this.thread = getThreadFactory().newThread(this);
  }
  
  public void run() {
    runWorker(this);
  }
}

final void runWorker(Worker w) {
  // 注释1
  Thread wt = Thread.currentThread();
  // Worker对象中获取执行任务
  Runnable task = w.firstTask;
  // 将Worker对象中的任务置空
  w.firstTask = null;
  w.unlock();
  boolean completedAbruptly = true;
  try {
    // 注释2
    while (task != null || (task = getTask()) != null) {
      w.lock();
      // 线程池的部分状态要中断正在执行的任务
      if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) && !wt.isInterrupted()) {
        wt.interrupt();
      }
      try {
        beforeExecute(wt, task);
        try {
          // 执行任务
          task.run();
          afterExecute(task, null);
        } catch (Throwable ex) {
          afterExecute(task, ex);
          throw ex;
        }
      } finally {
        task = null;
        w.completedTasks++;
        w.unlock();
      }
    }
    completedAbruptly = false;
  } finally {
    processWorkerExit(w, completedAbruptly);
  }
}

private Runnable getTask() {
  // 上次从队列中获取任务是否超时
  boolean timedOut = false;
  for (;;) {
    // 线程池状态判断，某些状态下不需要处理队列中的任务
    int c = ctl.get();
    if (runStateAtLeast(c, SHUTDOWN) && (runStateAtLeast(c, STOP) || workQueue.isEmpty())) {
      decrementWorkerCount();
      return null;
    }
    
    int wc = workerCountOf(c);
    
    // allowCoreThreadTimeOut是否允许核心线程超时销毁，默认为false
    // 通过allowCoreThreadTimeOut方法设置
    // wc > corePoolSize为true表示启用了非核心线程
    boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
    
    // wc > maximumPoolSize，可能的情况是因为同时执行了setMaximumPoolSize方法
    // timed && timedOut为true时，表示上次获取任务超时，当前需要进行超时控制
    // wc > 1 || workQueue.isEmpty()，工作线程数量大于1或队列为空
    if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) {
      // 减少工作线程数量
      if (compareAndDecrementWorkerCount(c)) {
        return null;
      }
      continue;
    }
    
    try {
      // 注释1
      // 从队列中获取待执行任务
      Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take();
      if (r != null) {
        return r;
      }
      timedOut = true;
    } catch (InterruptedException retry) {
      timedOut = false;
    }
  }
}

private void processWorkerExit(Worker w, boolean completedAbruptly) {
  // runWorker执行失败的场景
  if (completedAbruptly) {
    decrementWorkerCount();
  }
  
  final ReentrantLock mainLock = this.mainLock;
  mainLock.lock();
  try {
    completedTaskCount += w.completedTasks;
    // 从workers中删除Worker
    workers.remove(w);
  } finally {
    mainLock.unlock();
  }
  
  // 根据线程池状态判断是否结束线程池
  tryTerminate();
  
  int c = ctl.get();
  // STOP之下的状态，runWorker正常结束时completedAbruptly == false
  // 保证至少有1个worker，用于处理队列中的任务
  if (runStateLessThan(c, STOP)) {
    if (!completedAbruptly) {
      int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
      if (min == 0 && ! workQueue.isEmpty()) {
        min = 1;
      }
      if (workerCountOf(c) >= min){
        return;
      }
    }
    // runWorker异常退出时，即completedAbruptly == true
    // 或者是workers存活少于1个
    addWorker(null, false);
  }
}

public boolean prestartCoreThread() {
  return workerCountOf(ctl.get()) < corePoolSize && addWorker(null, true);
}

public void shutdown() {
  final ReentrantLock mainLock = this.mainLock;
  mainLock.lock();
  try {
    checkShutdownAccess();
    advanceRunState(SHUTDOWN);
    // 中断空闲线程
    interruptIdleWorkers();
    // ScheduledThreadPoolExecutor的hook
    onShutdown();
  } finally {
    mainLock.unlock();
  }
  tryTerminate();
}

public List<Runnable> shutdownNow() {
  List<Runnable> tasks;
  final ReentrantLock mainLock = this.mainLock;
  mainLock.lock();
  try {
    checkShutdownAccess();
    advanceRunState(STOP);
    // 中断所有线程
    interruptWorkers();
    tasks = drainQueue();
  } finally {
    mainLock.unlock();
  }
  tryTerminate();
  return tasks;
}

public interface RejectedExecutionHandler {
  
  void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
}

public static ExecutorService newFixedThreadPool(int nThreads) {
  return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());
}

public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
  return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(), threadFactory);
}

public static ExecutorService newCachedThreadPool() {
  return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());
}

public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
  return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>(), threadFactory);
}

public static ExecutorService newSingleThreadExecutor() {
  return new FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()));
}

public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) {
  return new FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(), threadFactory));
}

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
  return new ScheduledThreadPoolExecutor(corePoolSize);
}

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize, ThreadFactory threadFactory) {
  return new ScheduledThreadPoolExecutor(corePoolSize, threadFactory);
}

public static ScheduledExecutorService newSingleThreadScheduledExecutor() {
  return new DelegatedScheduledExecutorService(new ScheduledThreadPoolExecutor(1));
}

public static ScheduledExecutorService newSingleThreadScheduledExecutor(ThreadFactory threadFactory) {
  return new DelegatedScheduledExecutorService(new ScheduledThreadPoolExecutor(1, threadFactory));
}

public static ExecutorService newWorkStealingPool(int parallelism) {
  return new ForkJoinPool(parallelism, ForkJoinPool.defaultForkJoinWorkerThreadFactory, null, true);
}

public static ExecutorService newWorkStealingPool() {
  return new ForkJoinPool(Runtime.getRuntime().availableProcessors(), ForkJoinPool.defaultForkJoinWorkerThreadFactory, null, true);
}

public void setThreadFactory(ThreadFactory threadFactory);

public void setRejectedExecutionHandler(RejectedExecutionHandler handler);

public void setCorePoolSize(int corePoolSize);

public void setMaximumPoolSize(int maximumPoolSize);

public void setKeepAliveTime(long time, TimeUnit unit);