Java 线程池的一些概念

2021年12月05日 75

java

【强制】线程池不允许使用Executors去创建，而是通过ThreadPoolExecutor的方式，这样的处理方式让写的同学更加明确线程池的运行规则，规避资源耗尽的风险。说明:Executors 返回的线程池对象的弊端如下:
FixedThreadPool 和 SingleThreadPool:
允许的请求队列长度为 Integer.MAX_VALUE，可能会堆积大量的请求，从而导致 OOM。
CachedThreadPool:
允许的创建线程数量为 Integer.MAX_VALUE，可能会创建大量的线程，从而导致 OOM。

我们首先需要了解一些线程的概念

核心线程（corePool）： 线程池最终执行任务的角色是线程，核心线程是在线程中有新任务提交的时候，首先检查核心线程数，如果核心线程都在工作，而且数量也已经达到最大核心线程数，则不会继续新建核心线程，而会将任务放入等待队列。

等待队列（workQueue）： 如果核心线程都已经在工作，线程池持续添加任务，就会放到等待队列中，直到核心线程执行完成后，再去等待队列中拉取任务继续执行。

非核心线程： 由 最大线程数 - 核心线程数 得到，如果等待队列满了，那么就会创建新的线程来执行。

线程活跃保持时间（keepAliveTime）： 非核心线程如果空闲下来，超过这个时间内还未执行任务，则线程结束。

拒绝策略（RejectedExecutionHandler）： 当等待队列已满，线程数也达到最大线程数时，线程池会根据饱和策略来执行后续操作，默认的策略是抛弃要加入的任务。

JDK 为我们提供了什么

创建线程池的方法

newFixedThreadPool

返回固定长度的线程池，线程池中的线程数量是固定的

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

newCachedThreadPool

该方法返回一个根据实际情况来进行调整线程数量的线程池，空余线程存活时间是60s

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

newSingleThreadExecutor

该方法返回一个只有一个线程的线程池

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

等待队列的类型

SynchronousQueue

同步队列本身没有容量的，如果入队的时候没有空余线程则直接执行拒绝策略

ArrayBlockingQueue

基于数组来进行实现的，初始化时必须指定容量参数，如果线程池的线程数量没有超过 maximumPoolSize ，则创建新的线程执行任务，如果超过了 maximumPoolSize 则执行拒绝策略。

LinkedBlockingQueue

它内部是基于链表的形式，默认队列的长度是Integer.MAX_VALUE ，若后续仍有新的任务提交，而没有空闲的线程时，它会不断往队列中入队提交的任务，直到资源耗尽。OOM

PriorityBlockingQueue

public PriorityBlockingQueue(int initialCapacity,
                             Comparator<? super E> comparator) {
  if (initialCapacity < 1)
    throw new IllegalArgumentException();
  this.lock = new ReentrantLock();
  this.notEmpty = lock.newCondition();
  this.comparator = comparator;
  this.queue = new Object[initialCapacity];
}

它是一个无界队列，该队列可以根据 Comparator 顺序先后执行。OOM

ThreadPoolExecutor 内的一些方法

//高3位代表线程池的状态，低29位代表的是线程池的数量
//默认是RUNNING状态，线程池的数量为0
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
//线程个数位数，表示的Integer中除去最高的3位之后剩下的位数表示线程池的个数
private static final int COUNT_BITS = Integer.SIZE - 3;
//线程池的线程的最大数量
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

//线程池的状态
//负数补码 计算机内运算都是补码运算
// 1 = 00000000000000000000000000001
// -1 反码 = 11111111111111111111111111110
// -1 反码 补码 = 11111111111111111111111111111
// -1= 11111111111111111111111111111
// -1 << 29

//11100000000000000000000000000000
//接受新任务并且处理阻塞队列里面任务
private static final int RUNNING    = -1 << COUNT_BITS;
//00000000000000000000000000000000
//拒绝新任务但是处理阻塞队列的任务
private static final int SHUTDOWN   =  0 << COUNT_BITS;
//00100000000000000000000000000000
//拒接新任务并且抛弃阻塞队列里面的任务，同时会中断正在处理的任务
private static final int STOP       =  1 << COUNT_BITS;
//01000000000000000000000000000000
//所有任务都执行完(包括阻塞队列中的任务)后当线程池活动线程数为0，将要调用terminated方法。
private static final int TIDYING    =  2 << COUNT_BITS;
//01100000000000000000000000000000
//终止状态，terminated方法调用完成以后的状态
private static final int TERMINATED =  3 << COUNT_BITS;

二进制运算

正码
1  | 0000 0000 0000 0000 0000 0000 0000 0001
-1 | 1000 0000 0000 0000 0000 0000 0000 0001

反码
1  | 0000 0000 0000 0000 0000 0000 0000 0001
-1 | 1111 1111 1111 1111 1111 1111 1111 1110
  
补码
1  | 0000 0000 0000 0000 0000 0000 0000 0001
-1 | 1111 1111 1111 1111 1111 1111 1111 1111

public void execute(Runnable command) {
  // 判断提交的任务是不是为空，如果为空则抛出NullPointException异常
  if (command == null)
    throw new NullPointerException();
  // 获取线程池的状态和线程池的数量
  int c = ctl.get();
  // 如果线程池的数量小于corePoolSize，则进行添加线程执行任务
  if (workerCountOf(c) < corePoolSize) {
    // 添加线程修改线程数量并且将command作为第一个任务进行处理
    if (addWorker(command, true))
      return;
    // 获取最新的状态
    c = ctl.get();
  }
  // 如果线程池的状态是RUNNING，将命令添加到队列中
  if (isRunning(c) && workQueue.offer(command)) {
    // 二次检查线程池状态和线程数量
    int recheck = ctl.get();
    // 线程不是RUNNING状态，从队列中移除当前任务，并且执行拒绝策略。
    // 这里说明一点，只有RUNNING状态的线程池才会接受新的任务，其余状态全部拒绝。
    if (! isRunning(recheck) && remove(command))
      reject(command);
    //如果线程池的线程数量为空时，代表线程池是空的，添加一个新的线程。
    else if (workerCountOf(recheck) == 0)
      addWorker(null, false);
  }
  //如果队列是满的，或者是SynchronousQueue队列时，则直接添加新的线程执行任务，如果添加失败则进行拒绝
  //可能线程池的线程数量大于maximumPoolSize则采取拒绝策略。
  else if (!addWorker(command, false))
    reject(command);
}

private boolean addWorker(Runnable firstTask, boolean core) {
  retry:
  for (;;) {
    // 获取线程池的状态和线程池线程的数量
    int c = ctl.get();
    // 单独获取线程池的状态
    int rs = runStateOf(c);

    // 检查队列是否只在必要时为空
    // 当线程池状态是STOP、TIDYING、TERMINATED时，这些状态的时候不需要进行线程的添加和启动操作，因为如果是上面的状态，其实线程池的线程正在进行销毁操作，意味着线程调用了shutdownNow等方法。
   	// 如果线程池状态为SHUTDOWN并且第一个任务不为空时，不接受新的任务，直接返回false，也就是说SHUTDOWN的状态，不会接受新任务，只会针对队列中未完成的任务进行操作。
    // 当线线程池状态为SHUTDOWN并且队列为空时，直接返回不进行任务添加。
    if (rs >= SHUTDOWN &&		// 线程池的状态是SHUTDOWN、STOP、TIDYING、TERMINATED
        ! (rs == SHUTDOWN &&	// 可以看做是rs!=SHUTDOWN,线程池状态为STOP、TIDYING、TERMINATED
           firstTask == null &&	// 可以看做firstTask!=null，并且rs=SHUTDOWN
           ! workQueue.isEmpty()))	// 可以看做rs=SHUTDOWN，并且workQueue.isEmpty()队列为空
      return false;
    //循环CAS增加线程池中线程的个数
    for (;;) {
      //获取线程池中线程个数
      int wc = workerCountOf(c);
      //如果线程池线程数量超过最大线程池数量，则直接返回
      if (wc >= CAPACITY ||
          //如果指定使用corePoolSize作为限制则使用corePoolSize，反之使用maximumPoolSize，最为工作线程最大线程线程数量，如果工作线程大于相应的线程数量则直接返回。
          wc >= (core ? corePoolSize : maximumPoolSize))
        return false;
      //CAS增加线程池中线程的数量
      if (compareAndIncrementWorkerCount(c))
        //跳出增加线程池数量。
        break retry;
      //如果修改失败，则重新获取线程池的状态和线程数量
      c = ctl.get();  // Re-read ctl
      //如果最新的线程池状态和原有线程状态不一样时，则跳转到外层retry中，否则在内层循环重新进行CAS
      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 {
    //创建一个Worker对象
    w = new Worker(firstTask);
    //获取worker中的线程，这里线程是通过ThreadFactory线程工厂创建出来的，
    final Thread t = w.thread;
    //判断线程是否为空
    if (t != null) {
      //添加独占锁，为添加worker进行同步操作，防止其他线程同时进行execute方法。
      final ReentrantLock mainLock = this.mainLock;
      mainLock.lock();
      try {
        //获取线程池的状态
        int rs = runStateOf(ctl.get());
        //如果线程池状态为RUNNING或者是线程池状态为SHUTDOWN并且第一个任务为空时，当线程池状态为SHUTDOWN时，是不允许添加新任务的，所以他会从队列中获取任务。
        if (rs < SHUTDOWN ||
            (rs == SHUTDOWN && firstTask == null)) {
          if (t.isAlive()) // precheck that t is startable
            throw new IllegalThreadStateException();
          //添加worker到集合中
          workers.add(w);
          int s = workers.size();
          //跟踪最大的线程池数量
          if (s > largestPoolSize)
            largestPoolSize = s;
          //添加worker成功
          workerAdded = true;
        }
      } finally {
        mainLock.unlock();
      }
      //如果添加worker成功就启动任务
      if (workerAdded) {
        t.start();
        workerStarted = true;
      }
    }
  } finally {
    //如果没有启动，w不为空就已出worker，并且线程池数量进行减少。
    if (! workerStarted)
      addWorkerFailed(w);
  }
  return workerStarted;
}

// 这里发现它是实现了AQS，是一个不可重入的独占锁模式
// 并且它还集成了Runable接口，实现了run方法。
private final class Worker
  extends AbstractQueuedSynchronizer
  implements Runnable
{
  private static final long serialVersionUID = 6138294804551838833L;

  /** 执行任务的线程，通过ThreadFactory创建 */
  final Thread thread;
  /** 初始化第一个任务*/
  Runnable firstTask;
  /** 每个线程完成任务的数量 */
  volatile long completedTasks;

  /**
     * 首先现将state值设置为-1，因为在AQS中state=0代表的是锁没有被占用，而且在线程池中shutdown方法会判断能否争抢到锁，如果可以获得锁则对线程进行中断操作，如果调用了shutdownNow它会判断state>=0会被中断。
     * firstTask第一个任务，如果为空则会从队列中获取任务，后面runWorker中。
     */
  Worker(Runnable firstTask) {
    setState(-1); // inhibit interrupts until runWorker
    this.firstTask = firstTask;
    this.thread = getThreadFactory().newThread(this);
  }

  /** 委托调用外部的runWorker方法 */
  public void run() {
    runWorker(this);
  }

  //是否独占锁
  protected boolean isHeldExclusively() {
    return getState() != 0;
  }

  protected boolean tryAcquire(int unused) {
    if (compareAndSetState(0, 1)) {
      setExclusiveOwnerThread(Thread.currentThread());
      return true;
    }
    return false;
  }

  protected boolean tryRelease(int unused) {
    setExclusiveOwnerThread(null);
    setState(0);
    return true;
  }

  public void lock()        { acquire(1); }
  public boolean tryLock()  { return tryAcquire(1); }
  public void unlock()      { release(1); }
  public boolean isLocked() { return isHeldExclusively(); }
  //这里就是上面shutdownNow中调用的线程中断的方法，getState()>=0
  void interruptIfStarted() {
    Thread t;
    if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
      try {
        t.interrupt();
      } catch (SecurityException ignore) {
      }
    }
  }
}

可以看到Worker是一个实现了AQS的锁，它是一个不可重入的独占锁，并且他也实现了Runnable接口，实现了run方法，在构造函数中将AQS的state设置为-1，为了避免线程还没有进入runWorker方法前，就调用了shutdown或shutdownNow方法，会被中断，设置为-1则不会被中断。后面我们看到run方法，它调用的是ThreadPoolExecutor的runWorker方法，我们这里回想一下，在addWorker方法中，添加worker到HashSet<Worker>中后，他会将workerAdded设置为true，代表添加worker成功。

整体的逻辑是先进行创建线程，线程将Worker设置为执行程序，并将线程塞到Worker中，然后再addWorker中将Worker中的线程取出来，进行启动操作，启动后他会调用Worker中的run方法，然后run方法中将调用ThreadPoolExecutor的runWorker，然后runWorker又会调用Worker中的任务firstTask，这个fistTask是要真正执行的任务，也是用户自己实现的代码逻辑。

final void runWorker(Worker w) {
  //调用者也就是Worker中的线程
  Thread wt = Thread.currentThread();
  //获取Worker中的第一个任务
  Runnable task = w.firstTask;
  //将Worker中的任务清除代表执行了第一个任务了，后面如果再有任务就从队列中获取。
  w.firstTask = null;
  //这里还记的我们在new Worker的时候将AQS的state状态设置为-1，这里先进行解锁操作，将state设置为0
  w.unlock(); // allow interrupts
  boolean completedAbruptly = true;
  try {
    //循环进行获取任务，如果第一个任务不为空，或者是如果第一个任务为空，从任务队列中获取任务，如果有任务则返回获取的任务信息，如果没有任务可以获取则进行阻塞，阻塞也分两种第一种是阻塞直到任务队列中有内容，第二种是阻塞队列一定时间之后还是没有任务就直接返回null。
    while (task != null || (task = getTask()) != null) {
      //先获取worker的独占锁，防止其他线程调用了shutdown方法。
      w.lock();
      // 如果线程池正在停止，确保线程是被中断的，如果没有则确保线程不被中断操作。
      if ((runStateAtLeast(ctl.get(), STOP) || //如果线程池状态为STOP、TIDYING、TERMINATED直接拒绝任务中断当前线程
           (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;
        //统计当前Worker完成了多少任务
        w.completedTasks++;
        //独占锁释放
        w.unlock();
      }
    }
    completedAbruptly = false;
  } finally {
    //处理Worker的退出操作，执行清理工作。
    processWorkerExit(w, completedAbruptly);
  }
}

我们看到如果Worker是第一次被启动，它会从Worker中获取firstTask任务来执行，然后执行成功后，它会getTask()来从队列中获取任务，这个地方比较有意思，它是分情况进行获取任务的，我们都直到BlockingQueue中提供了几种从队列中获取的方法，这个getTask中使用了两种方式，第一种是使用poll进行获取队列中的信息，它采用的是过一点时间如果队列中仍没有任务时直接返回null，然后还有一个就是take方法，take方法是如果队列中没有任务则将当前线程进行阻塞，等待队列中有任务后，会通知等待的队列线程进行消费任务

private Runnable getTask() {
  boolean timedOut = false; //poll获取超时

  for (;;) {
    //获取线程池的状态和线程数量
    int c = ctl.get();
    //获取线程池的状态
    int rs = runStateOf(c);

    //线程池状态大于等于SHUTDOWN
    //1.线程池如果是大于STOP的话减少工作线程池数量
    //2.如果线程池状态为SHUTDOW并且队列为空时，代表队列任务已经执行完，返回null，线程数量减少1
    if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
      decrementWorkerCount();
      return null;
    }
    //获取线程池数量。
    int wc = workerCountOf(c);

    //如果allowCoreThreadTimeOut为true，则空闲线程在一定时间未获得任务会清除
    //或者如果线程数量大于corePoolSize的时候会进行清除空闲线程
    boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
    //1.如果线程池数量大于最大的线程池数量或者对（空余线程进行清除操作并且poll超时了，意思是队列中没有内容了，导致poll间隔一段时间后没有获取内容超时了。
    //2.如果线程池的数量大于1或者是队列已经是空的
    //总之意思就是当线程池的线程池数量大于corePoolSize，或指定了allowCoreThreadTimeOut为true，当队列中没有数据或者线程池数量大于1的情况下，尝试对线程池的数量进行减少操作，然后返回null，用于上一个方法进行清除操作。
    if ((wc > maximumPoolSize || (timed && timedOut))
        && (wc > 1 || workQueue.isEmpty())) {
      if (compareAndDecrementWorkerCount(c))
        return null;
      continue;
    }

    try {
      //如果timed代表的是清除空闲线程的意思
      Runnable r = timed ?
        workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :	//等待一段时间如果没有获取到返回null。
      workQueue.take();					//阻塞当前线程
      //如果队列中获取到内容则返回
      if (r != null)
        return r;
      //如果没有获取到超时了则设置timeOut状态
      timedOut = true;
    } catch (InterruptedException retry) {
      timedOut = false;
    }
  }
}

工作线程调用getTask从队列中进行获取任务。

如果指定了allowCoreThreadTimeOut或线程池线程数量大于corePoolSize则进行清除空闲多余的线程，调用阻塞队列的poll方法，在指定时间内如果没有获取到任务直接返回false。

如果线程池中线程池数量小于corePoolSize或者allowCoreThreadTimeOut为false默认值，则进行阻塞线程从队列中获取任务，直到队列有任务唤醒线程。

private void processWorkerExit(Worker w, boolean completedAbruptly) {
  if (completedAbruptly) 				// 如果突然完成则调整线程数量
    decrementWorkerCount();		// 减少线程数量1

  final ReentrantLock mainLock = this.mainLock;
  mainLock.lock();														//获取锁,同时只有一个线程获得锁
  try {
    completedTaskCount += w.completedTasks;	//统计整个线程池完成的数量
    workers.remove(w);											//将完成任务的worker从HashSet中移除
  } finally {
    mainLock.unlock();											//释放锁
  }
  //尝试设置线程池状态为TERMINATED
  //1.如果线程池状态为SHUTDOWN并且线程池线程数量与工作队列为空时，修改状态。
  //2.如果线程池状态为STOP并且线程池线程数量为空时，修改状态。
  tryTerminate();								

  // 获取线程池的状态和线程池的数量
  int c = ctl.get();
  // 如果线程池的状态小于STOP，也就是SHUTDOWN或RUNNING状态
  if (runStateLessThan(c, STOP)) {
    //如果不是突然完成，也就是正常结束
    if (!completedAbruptly) {
      //如果指定allowCoreThreadTimeOut=true(默认false)则代表线程池中有空余线程时需要进行清理操作，否则线程池中的线程应该保持corePoolSize
      int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
      //这里判断如果线程池中队列为空并且线程数量最小为0时，将最小值调整为1，因为队列中还有任务没有完成需要增加队列，所以这里增加了一个线程。
      if (min == 0 && ! workQueue.isEmpty())
        min = 1;
      if (workerCountOf(c) >= min)
        return; // replacement not needed
    }
    //如果当前线程数效益核心个数，就增加一个Worker
    addWorker(null, false);
  }

final void tryTerminate() {
  for (;;) {
    // 获取线程池的状态和线程池的数量组合状态
    int c = ctl.get();
    //这里单独下面进行分析，这里说明两个问题，需要反向来想这个问题。
    //1.如果线程池状态STOP则不进入if语句
    //2.如果线程池状态为SHUTDOWN并且工作队列为空时，不进入if语句
    if (isRunning(c) ||
        runStateAtLeast(c, TIDYING) ||
        (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
      return;
    //如果线程池数量不为空时，进行中断操作。
    if (workerCountOf(c) != 0) { // Eligible to terminate
      interruptIdleWorkers(ONLY_ONE);
      return;
    }

    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
      //修改状态为TIDYING，并且将线程池的数量进行清空
      if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
        try {
          //执行一些逻辑，默认是空的
          terminated();
        } finally {
          //修改状态为TERMINATED
          ctl.set(ctlOf(TERMINATED, 0));
          //唤醒调用awaitTermination方法的线程
          termination.signalAll();
        }
        return;
      }
    } finally {
      mainLock.unlock();
    }
    // else retry on failed CAS
  }

public void shutdown() {
  //获取全局锁
  final ReentrantLock mainLock = this.mainLock;
  mainLock.lock();
  try {
    //权限检查
    checkShutdownAccess();
    //设置线程池状态为SHUTDOWN，如果状态已经是大于等于SHUTDOWN则直接返回
    advanceRunState(SHUTDOWN);
    //如果线程没有设置中断标识并且线程没有运行则设置中断标识
    interruptIdleWorkers();
    //空的可以实现的内容
    onShutdown(); // hook for ScheduledThreadPoolExecutor
  } finally {
    mainLock.unlock();
  }
  //尝试修改线程池状态为TERMINATED
  tryTerminate();
}

首先对当前线程进行权限检测，查看是否设置了安全管理器，如果设置了则要看当前调用shutdown的线程有没有权限都关闭线程的权限，如果有权限还要看是否有中断工作现成的权限，如果没有权限则抛出SecurityException或NullPointException异常。

设置线程池状态为SHUTDOWN，如果状态已经是大于等于SHUTDOWN则直接返回

如果线程没有设置中断标识并且线程没有运行则设置中断标识

尝试修改线程池状态为TERMINATED

private void advanceRunState(int targetState) {
  for (;;) {
    //获取线程池状态和线程池的线程数量
    int c = ctl.get();
    if (runStateAtLeast(c, targetState) ||		//如果线程池的状态>=SHUTDOWN
        ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))	//设置线程池状态为SHUTDOWN
      //返回
      break;										
  }
}

当线程池的状态>=SHUTDOWN，直接返回
如果线程池状态为RUNNING，设置线程池状态为SHUTDOWN，设置成功则返回

private void interruptIdleWorkers(boolean onlyOne) {
  //获取全局锁，同时只能有一个线程能够调用shutdown方法
  final ReentrantLock mainLock = this.mainLock;
  mainLock.lock();
  try {
    //遍历工作线程
    for (Worker w : workers) {
      Thread t = w.thread;
      //如果当前线程没有设置中断标志并且可以获取Worker自己的锁
      if (!t.isInterrupted() && w.tryLock()) {
        try {
          //设置中断标志
          t.interrupt();
        } catch (SecurityException ignore) {
        } finally {
          w.unlock();
        }
      }
      //执行一次，清理空闲线程。
      if (onlyOne)
        break;
    }
  } finally {
    mainLock.unlock();
  }
}

我们看到当我们调用shutdown方法的时候，只是将空闲的线程给设置了中断标识，也就是活跃正在执行任务的线程并没有设置中断标识，直到将任务全部执行完后才会逐步清理线程操作

public List<Runnable> shutdownNow() {
  List<Runnable> tasks;
  final ReentrantLock mainLock = this.mainLock;
  mainLock.lock();
  try {
    //权限检查
    checkShutdownAccess();
    //设置线程池状态为STOP，如果状态已经是大于等于STOP则直接返回
    advanceRunState(STOP);
    //这里是和SHUTDOWN区别的地方，这里是强制进行中断操作
    interruptWorkers();
    //将为完成任务复制到list集合中
    tasks = drainQueue();
  } finally {
    mainLock.unlock();
  }
  //尝试修改线程池状态为TERMINATED
  tryTerminate();
  return tasks;
}

shutdownNow`方法将线程池状态设置为`STOP`，而`shutdown`则将状态修改为`SHUTDOWN

shutdownNow方法将工作任务进行中断操作，也就是说如果工作线程在工作也会被中断，而shutdown则是先尝试获取锁如果获得锁成功则进行中断标志设置，也就是中断操作，如果没有获取到锁则等待进行完成后自动退出。

shutdownNow方法返回未完成的任务列表。

拒绝策略

AbortPoliay

该策略会直接抛出异常，阻止系统正常工作

CallerRunsPolicy

只要线程池没有关闭线程池状态是RUNNING状态，该略略直接调用线程中运行当前被丢弃的任务

DiscardOledestPolicy

该策略将丢弃最老的一个请求，也就是即将被执行的第一个任务，并尝试再次提交任务

DiscardPolicy

该策略将丢弃最老的一个请求，也就是即将被执行的第一个任务，并尝试再次提交任务

主线程进行线程池的调用，线程池执行execute方法
线程池通过addWorker进行创建线程，并将线程放入到线程池中，这里我们看到第二步是将线程添加到核心线程中，其实线程池内部不分核心线程和非核心线程，只是根据corePoolSize和maximumPoolSize设置的大小来进行区分，因为超过corePoolSize的线程会被回收，至于回收那些线程，是根据线程获取任务的时候进行判断，当前线程池数量大于corePoolSize，或者指定了allowCoreThreadTimeOut为true，则他等待一定时间后会返回，不会一直等待
当线程池的数量达到corePoolSize时，线程池首先会将任务添加到队列中
当队列中任务也达到了队列设置的最大值时，它会创建新的线程，注意的是此时的线程数量已经超过了corePoolSize，但是没有达到maximumPoolSize最大值。
当线程池的线程数量达到了maximumPoolSize，则会相应拒绝策略。

CountDownLatch aqs

在使用CountDownLatch的时候，是先创建CountDownLatch对象，然后在每次执行完一个任务后，就执行一次countDown()方法。直到通过getCount()获取到的值为0时才算执行完，如果count值不为0可通过await()方法让主线程进行等待，知道所有任务都执行完成，count的值被设为0。

public CountDownLatch(int count) {
  if (count < 0) throw new IllegalArgumentException("count < 0");
  this.sync = new Sync(count);
}

private static final class Sync extends AbstractQueuedSynchronizer {
  private static final long serialVersionUID = 4982264981922014374L;
  Sync(int count) {
    setState(count);
  }
}

public void await() throws InterruptedException {
  sync.acquireSharedInterruptibly(1);// 等待可中断的获取共享资源的方法
}
public final void acquireSharedInterruptibly(int arg)
  throws InterruptedException {
  if (Thread.interrupted()) // 如果线程已经中断，直接抛出异常结束。
    throw new InterruptedException();
  if (tryAcquireShared(arg) < 0)// 尝试获取共享资源，获取失败后，自旋入队列
    doAcquireSharedInterruptibly(arg);// 可中断的入队列过程
}

CountDownLatch中内部类Sync的releaseShared()方法，是使用的AQS的releaseShared()方法。

public void countDown() {
    sync.releaseShared(1);
}

public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {// 尝试释放资源
        doReleaseShared();// 释放资源成功后，唤醒节点。
        return true;
    }
    return false;
}

protected boolean tryReleaseShared(int releases) {
  // Decrement count; signal when transition to zero
  for (;;) {
    int c = getState();
    if (c == 0) // 若state为0，说明已经不需要释放资源了，直接返回false。
      return false;
    int nextc = c-1;
    if (compareAndSetState(c, nextc))// 真正的释放资源，是通过CAS的方式将state的值减1。
      return nextc == 0;
  }
}

demo

package com.zhiqi.thread;

import org.junit.Test;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author czq
 */
public class ThreadDemoTest {

    @Test
    public void fixedDemo() throws InterruptedException {
        CountDownLatch countDownLatch = new CountDownLatch(100);
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        for (int i = 0; i < 100; i++) {
            int finalItem = i;
            executorService.execute(() -> {
                try {
                    this.biz(finalItem);
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    countDownLatch.countDown();
                }
            });
        }
        countDownLatch.await();
        executorService.shutdown();
    }

    @Test
    public void cachedDemo() throws InterruptedException {
        CountDownLatch countDownLatch = new CountDownLatch(100);
        ExecutorService executorService = Executors.newCachedThreadPool();
        for (int i = 0; i < 100; i++) {
            int finalItem = i;
            executorService.execute(() -> {
                try {
                    this.biz(finalItem);
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    countDownLatch.countDown();
                }
            });
        }
        countDownLatch.await();
        executorService.shutdown();
    }

    @Test
    public void singleDemo() throws InterruptedException {
        CountDownLatch countDownLatch = new CountDownLatch(100);
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        for (int i = 0; i < 100; i++) {
            int finalItem = i;
            executorService.execute(() -> {
                try {
                    this.biz(finalItem);
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    countDownLatch.countDown();
                }
            });
        }
        countDownLatch.await();
        executorService.shutdown();
    }

    public void biz(int item) throws InterruptedException {
        System.out.println(Thread.currentThread().getName() + "--" + item);
        Thread.sleep(1000);
    }
}

技术杂谈

docker

k8s