线程池:ThreadPoolExecutor源码解读()

  本篇文章为你整理了线程池:ThreadPoolExecutor源码解读()的详细内容,包含有 线程池:ThreadPoolExecutor源码解读,希望能帮助你了解 线程池:ThreadPoolExecutor源码解读。

  关于获取任务超时,会依赖以下条件:

  --1、开启核心线程超时设置 或 线程池线程数大于核心线程数
 

  --2、符合1,且从workqueue获取任务超时。(如果不符合1,则以阻塞方式获取任务,不会超时)

  
线程池最小保留线程数:
 

  --1、如果没有开启核心线程超时配置,则至少保留corePoolSize个线程
 

  --2、如果开启核心线程超时并且当前队列里面还有任务,只需保留1个线程

  
当线程池线程数(ctl低位)少于核心线程数(corePoolSize),创建新线程执行任务

  当线程池线程数大于等于核心线程数,且任务队列未满时,将新任务放入到任务队列中,不创建线程

  当线程池线程数大于等于核心线程数(maximumPoolSize),且任务队列已满
 

  --如果工作线程数少于最大线程数,则创建新线程执行任务
 

  --如果工作线程数等于最大线程数,则抛出异常,拒绝新任务进入
 

  二、运行期间
 

  1、线程启动后,将一直循环获取任务并执行,只有当获取任务超时,或者线程池被终止,才会结束。
 

  2、如果获取任务超时,那么Worker线程自然结束。此时线程池减少了1个线程。
 

  3、在线程结束后,线程池会检查:1、线程池线程数 最少保留线程数 2、任务执行异常结束。如果符合,线程池会自动补充1个Worker

  三、终止阶段
 

  调用shutdown()和shutdownNow()都导致线程池线程数减少。
 

  1、shutdown()方式终止线程池:
 

  --停止提交新的任务,已在队列的任务会继续执行,并且中断空闲的Worker线程(Work.state从0- 1成功),线程池状态变为SHUTDOWN
 

  2、shutdownNow()方式终止线程池:
 

  --关闭线程池,不再接受新的任务,中断已经启动的Worker线程(Work.state 0),线程池状态改为STOP

  线程池创建线程及处理任务过程:
 

  梳理一下大概流程:

  用户线程调用execute()提交Runnable任务

  execute()调用addWork()将任务提交给线程池处理:如果有可用的核心线程,则提交给核心线程处理。反则,将任务先添加到任务队列(workQueen)中。

  addWorker()方法将启动一个worker线程,调用runWorker()来处理任务。

  runWorker()方法将循环获取任务,并运行任务的run()方法来执行真正的业务。如果是以核心线程提交任务,则优先处理该任务,否则,循环调用getTask()来获取任务

  getTask()方法,从任务队列(workQueen)取出任务,并返回。

  getTask()没有拿到任务,则执行线程结束processWorkerExit()

  线程池创建阶段:
 

  1.3 线程池配置的重要参数

  ctl:存储线程池状态以及线程数

  corePoolSize、maximumPoolSize、keepAliveTime、workQueue 参照下面的源码分析说明

  allowCoreThreadTimeOut:是否开启核心线程超时。默认false,不在构造函数设置,需要调用方法设置

  HashSet workers:线程池终止时会从该集合找线程来中断,源码分析有说明

  1.4 shutdown()和shutdownNow()区别

  shutdown() :关闭线程池,不再接受新的任务,已提交执行的任务继续执行;中断所有空闲线程;将线程池状态改为SHUTDOWN

  ShutDownNow():关闭线程池,不再接受新的任务,中断已经启动的Worker线程;将线程池状态改为STOP;返回未完成的任务队列

  1.5 线程池中的两个锁

  mainLock主锁是可重入的锁,用来同步更新的成员变量

  Worker内部实现了一个锁,它是不可重入的,在shutdown()场景中,通过tryLock确保不会中断还没有开始执行或者还在执行中的worker线程。

  2 源码分析过程中的困惑及解惑

  ---什么情况任务会提交失败?
 

  同时符合以下条件,任务才会被提交:

  
线程池状态等于RUNNING状态;

  如果任务队列已经满了,并且线程池线程数 少于 配置的线程池最大线程数(maximumPoolSize) 且小于线程池的最大支持线程数(CAPACITY)时。(如果队列没满,任务将会先加入到队列中)

  特别说明:特殊情况会创建任务为空的Worker线程来帮助队列中的任务跑完

  ---核心线程数的意义?从测试结果看,他决定了工作线程最大并发数,但未代码验证

  
核心线程数决定提交任务什么时候会被放入到队列中:即线程池线程数 =核心线程数时。

  核心线程数大小跟并发执行线程(任务)无关。也就是,它不决定工作线程最大并发数

  核心线程数可以动态修改。(如果增大了,可能会马上创建新的Worker线程)

  
---线程池状态不是RUNNING,或者往workQueue添加worker失败,这是为什么还要提交任务
 

  以下情况会创建任务为空的Worker线程来执行队列中的任务

  当前线程池状态为shutdown,但是任务队列不为空,这时创建Worker线程来帮助执行队列的任务

  当前线程池状态为running, 任务添加到队列后,接着线程池被关闭,并且从队列移除该任务失败,并且线程池线程数为0,这时创建Worker线程来确保刚提交的任务有机会执行。

  ---为什么runWorker()方法在执行任务前后加锁,但是线程依然能够并发?

  worker线程是通过创建Worker对象来创建的,在addWorke()的while循环创建了多个Worker对象,每个Worker对象都有自己的锁,Worker线程通过runWorker()访问的是当前对象的锁,因此Worker线程能够并发;

  锁的意义是限制不能中断执行中的任务,因为主线程调用shutdown()和shutdownNow()方法时,会遍历WorkerSet的Worker对象,调用tryLock(),这时主线程和Worker线程竞争同一个锁。

  3 源码分析

  3.1 类继承关系

  Executo接口:专门提交任务,只有一个execute()方法。Executor 提供了一种将任务的提交和任务的执行两个操作进行解耦的思路:客户端无需关注执行任务的线程是如何创建、运行和回收的,只需要将任务的执行逻辑包装为一个 Runnable 对象传递进来即可,由 Executor 的实现类自己来完成最复杂的执行逻辑

  ExecutorService接口:继承了Executor,扩展执行任务的能力。例如:获取任务的执行结果、取消任务等功能;提供了关闭线程池、停止线程池,以及阻塞等待线程池完全终止的方法,需要ThreadPoolExecutor实现

  AbstractExecutorServic类:实现了 ExecutorService ,是上层的抽象类,负责将任务的执行流程串联起来,从而使得下层的实现类 ThreadPoolExecutor只需要实现一个执行任务的方法即可

  ThreadPoolExecutor:可以看做是基于生产者-消费者模式的一种服务,内部维护的多个线程相当于消费者,提交的任务相当于产品,提交任务的外部就相当于生产者

  3.2 类的常量/成员变量

  

 //--------------------------常量部分------------------------

 

  // 常量29。用在移位计算Integer.SIZE=32)

  private static final int COUNT_BITS = Integer.SIZE - 3; //29

  // 最大支持线程数 2^29-1:000 11111111111111111...

  private static final int CAPACITY = (1 COUNT_BITS) - 1;

  // 以下为线程池的四个状态,用32位中的前三位表示

   // 011 terminated() 方法执行完成后,线程池的状态会转为TERMINATED.

  private static final int TERMINATED = 3 COUNT_BITS;

  // 010 所有任务都销毁了,workCount=0的时候,线程池的装填在转换为TIDYING是,会执行钩子方法terminated()

  private static final int TIDYING = 2 COUNT_BITS; //翻译为整理

  // 001 拒绝新的任务提交,清空在队列中的任务

  private static final int STOP = 1 COUNT_BITS;

  // 000 拒绝新的任务提交,会将队列中的任务执行完,正在执行的任务继续执行.

  private static final int SHUTDOWN = 0 COUNT_BITS;

  // 111 00000 00000000 00000000 00000000 线程运行中 【running状态值为负数最小】

  private static final int RUNNING = -1 COUNT_BITS; //线程池的默认状态

  
// ctl存储线程池状态和线程池大小,那么用前3位表示线程池状态,后29位表示:线程池大小,即线程池线程数

  //线程池状态初始值为RUNNING

  private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

  //任务队列

  //保存不能马上执行的Runnable任务。

  //执行shutdownNow()时,会返回还在队列的任务

  private final BlockingQueue Runnable workQueue;

  // 主锁,对workers、largestPoolSize、completedTaskCount的访问都必须先获取该锁

  private final ReentrantLock mainLock = new ReentrantLock();

  // 包含池中的所有工作线程的集合。持有mainLock访问

  // 创建Worker时,添加到集合

  // 线程结束时,从集合移除

  // 调用shutdown()时,从该集合中找到空闲线程并中断

  // 调用shutdownNow()时,从该集合中找到已启动的线程并中断

  private final HashSet Worker workers = new HashSet Worker

  // 线程通信手段, 用于支持awaitTermination方法:等待所有任务完成,并支持设置超时时间,返回值代表是不是超时.

  private final Condition termination = mainLock.newCondition();

  // 记录workers历史以来的最大值。持有mainLock访问

  // 每次增加worker的时候,都会判断当前workers.size()是否大于最大值,大于则更新

  // 用于线程池监控的,作为重要指标

  private int largestPoolSize;

  // 计数所有已完成任务,持有mainLock访问

  // 每个worker都有一个自己的成员变量 completedTasks 来记录当前 worker 执行的任务次数, 当前线worker工作线程终止的时候, 才会将worker中的completedTasks的数量加入到 completedTaskCount 指标中.

  private long completedTaskCount;

  // 线程工厂

  private volatile ThreadFactory threadFactory;

  // 拒绝策略,默认四种AbortPolicy、CallerRunsPolicy、DiscardPolicy、DiscardOldestPolicy,建议自己实现,增加监控指标

  private volatile RejectedExecutionHandler handler;

  // keepAliveTime和allowCoreThreadTimeOut 是关于线程空闲是否会被销毁的配置

  // 关于空闲的说明:

  // 1、线程池在没有关闭之前,会一直向任务队列(workqueue)获取任务执行,如果任务队列是空的,在新任务提交上来之前,就会产生一个等待时间,期间,线程处于空闲状态

  // 2、向任务队列获取任务用:workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS),表示阻塞式获取元素,等待超时,则终止等待并返回false。通过判断poll()方法是true/falle来判定线程是否超时

  // 获取任务的等待时间 ,以下两种情况会使用到该值

  //1、如果启用allowCoreThreadTimeOut,那表示核心线程的空闲时间

  // 2、当线程池内线程数超过corePoolSize,表示线程获取任务的等待时间

  private volatile long keepAliveTime;

  // 核心线程是否开启超时

  // false:表示核心线程一旦启动,会一直运行,直至关闭线程池。默认该值

  // true:表示核心线程处于空闲且时间超过keepAliveTime,核心线程结束后,将不再创建新线程

  // (默认的构造函数没有设置这个属性,需要手工调用allowCoreThreadTimeOut()方法来设置)

  private volatile boolean allowCoreThreadTimeOut;

  //核心线程数量

  //核心线程是指:线程会一直存活在线程池中,不会被主动销毁【如果核心线程开启超时,有可能被被销毁】。

  private volatile int corePoolSize;

  // 配置的线程池最大线程数

  private volatile int maximumPoolSize;

  // 默认拒绝策略 AbortPolicy

  private static final RejectedExecutionHandler defaultHandler = new AbortPolicy();

  // 安全控制访问(主要用于shutdown和 shutdownNow方法

  private static final RuntimePermission shutdownPerm = new RuntimePermission("modifyThread");

  // 在threadPoolExecutor初始化的时候赋值,acc对象是指当前调用上下文的快照,其中包括当前线程继承的AccessControlContext和任何有限的特权范围,使得可以在稍后的某个时间点(可能在另一个线程中)检查此上下文。

  private final AccessControlContext acc;

  

 

  3.3 成员变量访问方法

  

// 获取当前线程池的状态(前3位)

 

  private static int runStateOf(int c) { return c ~CAPACITY; }

  // 获取当前线程池中线程数(后29位)

  private static int workerCountOf(int c){ return c CAPACITY; }

  // 更新状态和数量

  private static int ctlOf(int rs, int wc) { return rs wc; }

  // 小于判断C是不是小于S,比如runStateLessThan(var,STOP),那var就只有可能是(RUNNING,SHUTDOWN)

  private static boolean runStateLessThan(int c, int s) {

   return c

  // 是不是C = S

  private static boolean runStateAtLeast(int c, int s) {

   return c

  // 判断状态是不是RUNNING

  private static boolean isRunning(int c) {

   return c SHUTDOWN;

  

 

  关于-1 29说明:

  

-1 COUNT_BITS

 

  这里是-1往左移29位,稍微有点不一样,-1的话需要我们自己算出补码来

  -1的原码

  10000000 00000000 00000000 00000001

  -1的反码,负数的反码是将原码除符号位以外全部取反

  11111111 11111111 11111111 11111110

  -1的补码,负数的补码就是将反码+1

  11111111 11111111 11111111 11111111

  关键了,往左移29位,所以高3位全是1就是RUNNING状态

  111 00000 00000000 00000000 00000000

  

 

  3.4 构造函数

  

//corePoolSize、maximumPoolSize、keepAliveTime、unit、workQueue 这五个参数必须指定

 

  //最多参构造函数

  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.acc = System.getSecurityManager() == null ?

   null :

   AccessController.getContext();

   this.corePoolSize = corePoolSize;

   this.maximumPoolSize = maximumPoolSize;

   this.workQueue = workQueue;//默认使用SynchronousQueue Runnable

   this.keepAliveTime = unit.toNanos(keepAliveTime); //默认60S

   this.threadFactory = threadFactory; //默认使用DefaultThreadFactory

   this.handler = handler;

  

 

  构造函数总结:
 

  初始化:corePoolSize(核心线程池大小)、maximumPoolSize(线程池容纳最大线程数)、workQueue(任务队列)、threadFactory(线程工厂)、keepAliveTime(空闲线程存活时长)、handler(拒绝策略)AccessControlContext

  3.5 静态内部类Worker

  3.5.1 Worker继承关系

  

private final class Worker extends AbstractQueuedSynchronizer implements Runnable {

 

  

 

  --Worker继承于AbstractQueuedSynchronizer

  Worker继承于AQS 为的就是自定义实现不可重入的特性(所以没有使用 synchronized 或者 ReentrantLock)来辅助判断线程是否处于执行任务的状态:在开始执行任务前进行加锁,在任务执行结束后解锁,以便在后续通过判断 Worker 是否处于锁定状态来得知其是否处于执行阶段

  -- Worker实现Runnable接口

  Worker实现Runnable接口,线程是通过getThreadFactory().newThread(this) 来创建的,即将 Worker 本身作为构造参数传给 Thread 进行初始化,所以在 thread 启动的时候 Worker 的 run() 方法就会被执行。

  关于ThreadFactory说明:

  

public interface ThreadFactory {

 

   Thread newThread(Runnable r);

  

 

  3.5.2 Worker源码分析

  

private final class Worker

 

   extends AbstractQueuedSynchronizer

   implements Runnable

   private static final long serialVersionUID = 6138294804551838833L;

   //线程类型的属性:thread,线程池启动工作线程,就是启动这个thread。

   // 1、通过this.thread=getThreadFactory().newThread(this),初始化了属性thread,this就是指Worker对象

   //2、因为Worker类实现了Runnable接口,所以thread启动后,会运行Worker的run()方法,然后就去执行runWorker(this)方法

   final Thread thread;

   //线程要执行的第1个任务(可能为 null) 它表示这个任务立即执行,不需要放到任务队列。在工作线程数 核心线程数时,这种场景会出现

   Runnable firstTask;

   //保存Worker线程池执行过的任务数,在runWorker()的finally中累加更新。任务执行成功与否都会更新

   volatile long completedTasks;

   Worker(Runnable firstTask) {

   setState(-1); // AQS父类的state。设为-1

   this.firstTask = firstTask; //firstTask赋初值

   this.thread = getThreadFactory().newThread(this); //属性thread赋值

   //Runnable run方法实现

   public void run() {

   runWorker(this); //调用runWorkder方法:将Worker对象传递给调用者,这样就可以访问firstTask、thread等属性以及lock()相关方法

   // state 的值说明

   // -1:worker初始化; 1 :锁被独占; 0:锁空闲

   //是否持有锁 AQS父类方法的实现

   protected boolean isHeldExclusively() {

   return getState() != 0;

   //以独占方式获取锁,将state设为1 AQS父类方法的实现

   protected boolean tryAcquire(int unused) {

   if (compareAndSetState(0, 1)) {

   setExclusiveOwnerThread(Thread.currentThread());

   return true;

   return false; //假如state=1,那么cas失败,返回false,线程就会进入AQS队列等待

   //释放锁。state设为0 AQS父类方法的实现

   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(); }

   //向线程发起中断请求

   // 符合:1、运行中的;2、没有处于中断 才能中断

   void interruptIfStarted() {

   Thread t;

   if (getState() = 0 (t = thread) != null !t.isInterrupted()) {

   try {

   t.interrupt();

   } catch (SecurityException ignore) {

  

 

  Worker类总结:

  所谓的线程池,其实就是正在运行的多个Worker线程。

  Worker作为线程启动后,它实际执行的是通过execute()提交的Runnable任务(实际业务),worker线程通过一个while循环来不断获取并任务,从而达到线程复用的效果

  firstTask:线程要执行的第1个任务(可能为 null) 它表示这个任务立即执行,不需要放到任务队列。在 1、线程数 核心线程数 2、队列已满且线程池不在运行状态 这两个场景下。

  4 重要方法详解

  4.1 execute()方法

  execute()用来提交要运行的任务

  

public void execute(Runnable command) {

 

   if (command == null)

   throw new NullPointerException();

   int c = ctl.get(); // 计算当前线程池的状态及线程数

   // 1、线程池线程数小于配置的核心线程数

   if (workerCountOf(c) corePoolSize) {

   // 将任务提交给核心线程处理

   if (addWorker(command, true))

   return;

   //失败的情况:1、线程池已经被关闭、2、线程池线程数大于等于核心线程数 (不能以true的方式提交了 )

   c = ctl.get(); // 重新获取线程池状态

   // 2、无空闲核心线程,将任务加入队列

   // 再次确认线程池为RUNNING状态,将任务加入队列【非阻塞式,队列满了会立即返回false】

   if (isRunning(c) workQueue.offer(command)) {

   //任务加入队列成功

   int recheck = ctl.get() ;//再次获取当前线程池状态(线程池可能被其它线程关闭了)

   //判断当前线程池状态是不是RUNNING状态,不是就从workQueue中删除command任务

   if (! isRunning(recheck) remove(command))

   reject(command);//执行拒绝策略

   //如果当前线程数是0(那证明还没有其他工作线程去处理这个任务),那么刚刚的任务肯定在阻塞队列里面了,这

   else if (workerCountOf(recheck) == 0)

   addWorker(null, false);//开启一个没有任务的Worker线程去执行队列的任务

   // 3 workQueue添加worker失败,即队列满了

   //创建非核心线程并执行任务

   else if (!addWorker(command, false)) //如果线程创建失败,说明要么是线程池当前状态!=RUNNING,或者是任务队列已满且线程总数达到最大线程数了

   reject(command);//执行拒绝策略.

  

 

  execute()总结

  进行三次addWorker的尝试:

  addWorker(command, true):创建任务并以核心线程执行

  核心线程数达到上限, 创建任务添加到任务队列,不创建线程

  addWorker(null, false) :任务添加到队列后,接着线程池被关闭,并且从队列移除该任务失败,并且线程池线程数为0,这时创建任务并以非核心线程执行

  addWorker(command, false) :任务队列已满,创建非核心线程并执行

  任务提交失败情况:线程池非RUNNING状态 并且 任务队列已满并且线程池线程数达到最大线程数(maximumPoolSize)

  4.2 addWorker()方法

  

//TERMINATED TIDYING STOP SHUTDOWN RUNNING 

 

  //创建新的线程执行当前任务

  //firstTask: 指定新增线程执行的第一个任务或者不执行任务

  private boolean addWorker(Runnable firstTask, boolean core) {

   //外循环:

   retry:

   for (;;) {

   int c = ctl.get();

   int rs = runStateOf(c);

   // 如果线程池状态是SHUTDOWN、STOP、TIDYING、TERMINATED就不允许提交。

   // 后面的特殊情况,线程池的状态是SHUTDOWN并且要要执行的任务为Null并且队列不是空,这种情况下是允许增加一个线程来帮助队列中的任务跑完的,因为shutdown状态下,允许执行完成阻塞队里中的任务

   if (rs = SHUTDOWN

   ! (rs == SHUTDOWN

   firstTask == null //execute()有addWorkder(null,false)的场景

   ! workQueue.isEmpty()))

   return false;

   //内循环:cas修改工作线程数,同时判断能否添加work

   for (;;) {

   int wc = workerCountOf(c);

   //添加任务前,线程池线程数已达到上限,此时不允许添加。上限分这三种情况:

   // 1、最大支持线程数

   // 2、以core=true提交时,配置的核心线程数。(返回false后,会以core=false再提交一次)

   // 3、以core=false提交时,配置的线程池可容纳最大线程数。

   if (wc = CAPACITY

   wc = (core ? corePoolSize : maximumPoolSize)) //使用core则上限为核心线程数,否则最大线程数

   return false;

   //没超过上限,通过CAS的方式增加worker的数量(+1),增加成功就跳出外层循环

   if (compareAndIncrementWorkerCount(c))

   break retry;

   c = ctl.get(); //获取最新的线程池状态,与刚开始的状态比较

   // - 变了,就从外层循环重新执行,重新进行状态的检查。

   // - 没变,从当前循环重新执行,重新执行CAS操作。

   if (runStateOf(c) != rs)

   continue retry;

   boolean workerStarted = false;

   boolean workerAdded = false;

   Worker w = null;

   try {

   //创建Worker,并给firstTask赋初值

   w = new Worker(firstTask);

   final Thread t = w.thread; //拿到属性thread

   if (t != null) {

   final ReentrantLock mainLock = this.mainLock;

   mainLock.lock(); //此处加锁:因为涉及属性:workers、largestPoolSize(可能) 更新

   try {

   int rs = runStateOf(ctl.get()); //获取线程池最新状态

   if (rs SHUTDOWN //如果当前状态是 SHUTDOWN也就是RUNNING状态

   (rs == SHUTDOWN firstTask == null)) { //或者状态是SHUTDOWN并且当前任务是空的(比如前面说的场景:阻塞队里里面还有,但当前已经是不允许提交的状态了)

   if (t.isAlive()) // 检查Worker线程已经开始跑了。(thread.start()变为alive)

   throw new IllegalThreadStateException();

   workers.add(w); //增加worker

   int s = workers.size(); //获取最新worker的总数,比较并更新largestPoolSize

   if (s largestPoolSize)

   largestPoolSize = s;

   workerAdded = true; //表示添加worker成功

   } finally {

   mainLock.unlock();

   if (workerAdded) {

   //启动worker线程。该线程会一直循环执行getTask(),直至返回null,线程才结束

   t.start(); //执行runWorker()

   workerStarted = true; //表示线程已经跑起来了

   } finally {

   if (! workerStarted)

   addWorkerFailed(w);//worker线程没成功启动,进入失败处理逻辑

   return workerStarted;//;返回当前worker是否启动成功。

  

 

  addWorker()总结:

  检查线程池状态以确定能否提交任务

  校验能否以核心线程的方式提交任务

  线程池的状态是SHUTDOWN并且任务队列不是空,允许增加一个线程来帮助队列中的任务跑完,但不会提交任务

  更新线程池线程数

  超过线程池线程数峰值则更新峰值(largestPoolSize)

  加锁(mainLock)来更新

  启动worker线程

  4.3 runWorker()方法

  

//执行任务

 

  final void runWorker(Worker w) {

   Thread wt = Thread.currentThread(); //runWorker()是由Worker.run()调用,因此wt就是worker线程

   Runnable task = w.firstTask; //拿到firstTask并赋值给局部变量task

   w.firstTask = null; //firstTask置空

   w.unlock(); // 将state设置为0。因为构造函数设成-1,在执行任务前置为0。

   boolean completedAbruptly = true;//标识任务是不是立刻就完成了。

   try {

   //循环:先执行firstTask(不为空),后续通过getTask()获取任务。

   while (task != null (task = getTask()) != null) {

   //任务执行前加锁,任务完成后解锁。

   //任何地方可通过判断锁状态来确认worker是否执行中

   w.lock(); //加锁。防止任务在执行过程中被中断。

   //判断目的:确保线程池当状态值大于等于 STOP 时有向线程发起过中断请求【调用了shutdownNow()】

   // 两种情况:

   //1)如果当前线程池的状态是 =Stop的,并且当前线程没有被中断,那么就要执行中断。

   //2)或者当前线程目前是已中断的状态并且线程池的状态也是 =Stop的(注意Thread.interrupted是会擦除中断标识符的),那么因为中断标识符已经被擦除了,那么!wt.isInterrupted()一定返回true,这个时候还是要将当前线程中断。第二次执行runStateAtLeast(ctl.get(), STOP)相当于一个二次检查

   if ((runStateAtLeast(ctl.get(), STOP)

   (Thread.interrupted()

   runStateAtLeast(ctl.get(), STOP)))

   !wt.isInterrupted())

   wt.interrupt();//中断worker线程 。因为线程池将要终止了,所以这里没有从workerSet移除当前线程

   try {

   beforeExecute(wt, task);//前置操作,空方法,可以业务自己实现

   Throwable thrown = null;

   try {

   //执行任务:就是执行通过execute()提交的Runnable

   task.run();//第一个是firstTask,后面的是通过getTask()拿到的任务

   } 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;//最后将task置为null,触发while循环的条件getTask()

   w.completedTasks++; //已完成的任务计数器+1

   w.unlock();//释放当前线程的独占锁

   completedAbruptly = false; //当第一个try的代码块有异常, completedAbruptly = false 不生效。最后completedAbruptly为true表示发生未知异常了

   } finally {

   //getTask返回null时,执行任务退出

   processWorkerExit(w, completedAbruptly);//completedAbruptly=true表示是突然退出的

  

 

  runWorker()总结:

  
执行任务前先判断线程池是否是STOPING状态,是则中断worker线程。

  执行任务:先执行firstTask,再从任务队列获取执行

  如果没有任务,调用processWorkerExit()来执行线程退出的工作。

  只要还有任务,worker线程就一直执行任务,并刷新completedTasks

  
// 1)如果线程池的状态是 =STOP状态,这个时候不再处理队列中的任务,并且减少worker记录数量,返回的任务为null,这个时候在runRWorker方法中会执行processWorkerExit进行worker的退出操作.

   // 2)如果线程池的状态是 =SHUTDOWN并且workQueue为空,就说明处于SHOTdown以上的状态下,且没有任务在等待,那么也属于获取不到任务,getTask返回null.

   if (rs = SHUTDOWN (rs = STOP workQueue.isEmpty())) {

   decrementWorkerCount();//扣减线程池线程数,在processWorkerExit()处理线程退出

   return null;

   int wc = workerCountOf(c);//获取当前wokrer的数量

   //以下涉及空闲线程是否会被线程池销毁的处理逻辑

   // 线程超时处理前置条件:开启核心线程超时 或 线程池线程数大于核心线程数

   boolean timed = allowCoreThreadTimeOut wc corePoolSize;

   //线程超时处理的进一步判断:

   // 线程池线程数超过maximumPoolSize 或者 线程设置允许超时且当前worker取任务超时

   //并且

   // 线程池大小不是零或阻塞队列是空的),这种就返回null,并减少线程池线程计数

   // 1、 (wc maximumPoolSize) (wc 1) 一般情况,线程池线程数会少于配置的最大线程数,但在addWork中 状态=shutdown且队列不为空时,会创建一个Worker,此时可能导致wc maximumPoolSize,这里同时限定wc 1。因此线程池减少1个线程也不影响任务的执行【processWorkerExit()会保证还有任务就至少留有1个worker线程】。

   // 2、 (wc maximumPoolSize) (workQueue.isEmpty()) 没有任务了,扣减更不影响

   // 3 、(timed timedOut) (wc 1) 超时了,先扣减再说

   // 4 、(timed timedOut) (workQueue.isEmpty()) 超时了 队列没有任务,必须要扣减

   if ((wc maximumPoolSize (timed timedOut))

   (wc 1 workQueue.isEmpty())) {

   //这里为啥不用decrementWorkerCount()呢,上面使用decrementWorkerCount()是因为确定不管是什么情况下,数量都要减,多减一次也没事,因为这个时候就是要关闭线程池释放资源

   //这里不一样,线程池的状态可能是RUNNING状态,多减一次,可能导致获取不到worker去跑

   if (compareAndDecrementWorkerCount(c))

   return null; //扣减线程池线程数,在processWorkerExit()处理线程退出

   continue;//扣减失败, 跳出本次循环重新检查

   //从队列中获取任务

   //符合【线程超时处理前置条件】时用poll设置超时时间,不符合就使用take(阻塞直至有返回)

   try {

   Runnable r = timed ?

   workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :

   workQueue.take();

   if (r != null)

   return r; //task不为空,此处返回task

   timedOut = true; // 此处,r == null,肯定是poll操作超时了(注意,不代表队列空了),继续for循环,回到if ((wc maximumPoolSize (timed timedOut)) 这个地方退出循环

   } catch (InterruptedException retry) {

   timedOut = false;

   private void decrementWorkerCount() {

   do {} while (! compareAndDecrementWorkerCount(ctl.get()));

  

 

 

  getTask()总结:

  
从workQueue中获取一个任务并返回

  没有获取到任务就扣减线程池线程数。获取不到任务的四种情况:

  线程池的状态是 =STOP

  线程池的状态是SHUTDOWN并且任务队列为空

  获取任务超时

  线程池线程数大于maximumPoolSize并且队列为空

  
4.5 processWorkerExit()方法

  

//worker线程没有拿到任务,成为空闲线程。该方法对空闲线程进一步处理

 

  private void processWorkerExit(Worker w, boolean completedAbruptly) {

   //如果completedAbruptly为true,则说明线程执行时出现异常,需要将workerCount数量减一

   //如果completedAbruptly为false,说明在getTask方法中已经对workerCount进行减一,这里不用再减

   if (completedAbruptly)

   decrementWorkerCount();

   final ReentrantLock mainLock = this.mainLock;

   mainLock.lock();

   try {

   //更新已完成任务的数量的统计项

   completedTaskCount += w.completedTasks;

   //从worker集合中移除该worker

   workers.remove(w);

   } finally {

   mainLock.unlock();

   //尝试关闭线程池,但如果是正常运行状态,就不会关闭

   tryTerminate();

   int c = ctl.get();

   if (runStateLessThan(c, STOP)) {//1、线程池是SHUTDOWN或RUNNING(如果不是这两个状态,说明线程已经停止了,不做任何操作)

   if (!completedAbruptly) {//2、线程正常结束

   // 如果没有开启核心线程超时配置,则至少保留corePoolSize个线程;

   int min = allowCoreThreadTimeOut ? 0 : corePoolSize;

   if (min == 0 !workQueue.isEmpty())//如果允许核心线程超时并且当前队列里面还有任务没跑,必须留1个线程,不能全死掉.

   min = 1;

   // 如果线程池数量 =最少预留线程数

   if (workerCountOf(c) = min)

   return; // 线程自然结束了,不用补充worker

   // 1、执行任务异常结束的,补充worker

   // 2、如果线程池数量 最少预留线程数,补充worker

   addWorker(null, false);//异常结束 增加worker

   //注: 别问我为啥上面要删除worker,还要再加,不删是不是不用加了. 明确下那个任务已经退出getTask那块的死循环了,永远回不去了,只能新增worker.

  

 

  processWorkerExit()方法总结!!!!!:

  
当Worker线程结束前,完成以下工作:扣减线程池线程数(ctl)、更新已完成任务数(completedTaskCount)、Worker集合中移除一个Worker(workers)、尝试终止线程池、计算线程池的最少保留线程数、根据最少保留线程数来确定是否补充一个Worker。

  关于最少保留线程数:如果没有开启核心线程超时配置,则至少保留corePoolSize个线程;如果开启核心线程超时并且当前队列里面还有任务,只需保留1个线程;

  需要补充worker的两种情况:1、线程池线程数 最少保留线程数 2、任务执行异常结束

  
//RUNNING状态,不能终止线程池

   //线程池状态是TIDYING或TERMINATED说明线程池已经处于正在终止的路上,不用再终止了.

   //状态为SHUTDOWN,但是任务队列不为空,也不能终止线程池

   if (isRunning(c)

   runStateAtLeast(c, TIDYING)

   (runStateOf(c) == SHUTDOWN ! workQueue.isEmpty()))

   return;

   //调用shutdown()或者shutdownNow()方法时,执行以下处理

   //工作线程数量不等于0,中断一个空闲的工作线程并返回

   //这个时候线程池一定是 1、STOP的状态或者 2、SHUTDOW且队列为空 这两种情况中断一个空闲worker

   if (workerCountOf(c) != 0) {

   interruptIdleWorkers(ONLY_ONE);

   return;

   final ReentrantLock mainLock = this.mainLock;

   mainLock.lock();

   try {

   // 设置线程池状态为TIDYING,如果设置成功,则调用terminated()

   if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {

   try {

   terminated(); //钩子方法,子类实现。默认什么都不做

   } finally {

   // 设置状态为TERMINATED

   ctl.set(ctlOf(TERMINATED, 0));

   termination.signalAll(); //唤醒阻塞等待的线程 (future的场景)

   return;

   } finally {

   mainLock.unlock();

   // else retry on failed CAS

  

 

 

  tryTerminate()总结

  
状态是RUNNING,不能直接终止(如果是调用shutdown(),shutdownNow(),会先将状态改为SHUTDOWN)

  状态是TIDYING或者TERMINATED,不能终止(因为已经处于终止过程中)

  状态是SHUTDOWN并且任务队列不为空,不能终止(因为还有任务要处理)

  
符合可以终止线程池的条件下,如果线程池线程数不等于0,那就中断1个Worker线程,不修改线程池状态

  符合可以终止线程池的条件下,并且线程池线程数等于0,那就将线程池状态改为TIDYING,执行完钩子方法terminated()后状态再改为TERMINATED

  


interruptIdleWorkers(ONLY_ONE); 是否好奇为啥这里只中断一个worker呢, 这里就涉及到了线程池的优雅退出了.

 

  当执行到 interruptIdleWorkers(ONLY_ONE) 前面的时候, 线程池只能处于两种状态:

  1) STOP 状态 , 这个时候 workQueue 可能是有值的 , workQueue 在清空的过程中了.

  2) SHUTDOWN 状态并且 workQueue 是空的 .

  这两种状态都是说明, 线程池即将关闭, 或者说空闲的线程此时已经没用了,这个时候随手关一个, 反正要关,早关晚关而已.

  

 

  4.7 interruptIdleWorker()方法

  

//中断一个或多个线程

 

  private void interruptIdleWorkers(boolean onlyOne) {

   final ReentrantLock mainLock = this.mainLock;

   mainLock.lock();

   try {

   //遍历worker,根据onlyOne判断,如果为ture只中断一个线程

   for (Worker w : workers) {

   Thread t = w.thread;

   //线程没有被中断并且线程是空闲状态

   //通过tryLock实现:不能中断还没有开始执行或者还在执行中的worker线程。

   //线程未启动:-1 ,线程正在执行:1 ,trylock:0-

   if (!t.isInterrupted() w.tryLock()) {

   try {

   t.interrupt(); //中断操作,之后该线程就结束了

   } catch (SecurityException ignore) {

   } finally {

   w.unlock();

   if (onlyOne)

   break;

   } finally {

   mainLock.unlock();

  

 

  interruptIdleWorker()总结:

  
从worker集合中遍历并中断worker线程

  只有worker线程状态是0的,才能够中断(不能中断未启动或者还在执行中的Worker线程)

  
4.8 shutdown()方法

  

//初始化一个有序的关闭,之前提交的任务都会被执行,但是新提交的任务则不会被允许放入任务队列中。如果之前被调用过了的话,那么再次调用也没什么用

 

  public void shutdown() {

   final ReentrantLock mainLock = this.mainLock;

   mainLock.lock(); //mainLock是全局变量,加锁确保不会并发关闭线程池

   try {

   checkShutdownAccess();//安全策略判断。方法检查每一个线程池的线程是否有可以ShutDown的权限。

   advanceRunState(SHUTDOWN); //CAS自旋把ctl中的状态从RUNNING变为SHUTDOWN

   interruptIdleWorkers();//中断所有空闲线程

   onShutdown(); // 方法告知子类,线程池要处于ShutDown状态了 ,ScheduledThreadPoolExecutor预留的钩子

   } finally {

   mainLock.unlock();

   tryTerminate();//尝试终止线程池

  

 

  shutdown()方法总结

  
执行shutdown()方法:关闭线程池,不再接受新的任务,已提交执行的任务继续执行。

  调用interruptIdleWorkers()先中断所有空闲线程

  调用tryTerminate()尝试终止线程池

  shutdown()将线程池状态改为SHUTDOWN但不是STOP

  


//关闭线程池,不再接受新的任务,正在执行的任务尝试终止

 

  public List Runnable shutdownNow() {

   List Runnable tasks;

   final ReentrantLock mainLock = this.mainLock;

   mainLock.lock();

   try {

   checkShutdownAccess();

   advanceRunState(STOP);//线程池的状态置为STOP

   interruptWorkers();

   tasks = drainQueue(); //将剩余任务返回

   } finally {

   mainLock.unlock();

   tryTerminate();

   return tasks;

   private void interruptWorkers() {

   final ReentrantLock mainLock = this.mainLock;

   mainLock.lock();

   try {

   for (Worker w : workers) //循环所有的worker

   w.interruptIfStarted();//已经启动的线程直接执行中断

   } finally {

   mainLock.unlock();

   void interruptIfStarted() {

   Thread t;

   //只有刚刚构建的worker的时候,状态state值是-1(这里也能体现刚构建的worker无法被中断),其他情况都是 =0的

   if (getState() = 0 (t = thread) != null !t.isInterrupted()) {

   try {

   t.interrupt();

   } catch (SecurityException ignore) {

  

 

  ShutDownNow()方法总结

  


public boolean prestartCoreThread() {

 

   return workerCountOf(ctl.get()) corePoolSize

   addWorker(null, true);

  

 

  启动一个空闲的线程作为核心线程

  如果核心线程数已到阈值, 会加入失败, 返回false, 如果线程池处于SHUTDOWN以上的状态也返回false

  只有真正这个线程调用start方法跑起来, 才会返回true

  4.12 prestartAllCoreThreads()方法

  启动所有核心线程,使他们等待获取任务

  

public int prestartAllCoreThreads() {

 

   int n = 0;

   while (addWorker(null, true))//null代表空闲线程,true代表是增加的是核心线程

   ++n;//死循环增加空闲 worker 而已

   return n;

  

 

  以上就是线程池:ThreadPoolExecutor源码解读()的详细内容,想要了解更多 线程池:ThreadPoolExecutor源码解读的内容,请持续关注盛行IT软件开发工作室。

郑重声明:本文由网友发布,不代表盛行IT的观点,版权归原作者所有,仅为传播更多信息之目的,如有侵权请联系,我们将第一时间修改或删除,多谢。

留言与评论(共有 条评论)
   
验证码: