您的位置：首页 > 科技 > 能源 > ThreadPoolExecutor状态流转和源码分析

ThreadPoolExecutor状态流转和源码分析

2024/9/24 3:21:31 来源：https://blog.csdn.net/Apandam/article/details/141962369  浏览:  次   关键词：ThreadPoolExecutor状态流转和源码分析

为什么使用线程池

  • 降低资源消耗 ，可以重复利用已创建的线程降低线程创建和销毁造成的消耗。
  • 提高响应速度，当任务到达时，任务可以不需要等到线程创建就能立即执行。
  • 提高线程的可管理性 ，线程是稀缺资源，如果无限制地创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一分配、调优和监控。

阿里巴巴开发手册对线程池使用的建议

【推荐】ThreadPoolExecutor 设置线程存活时间(setKeepAliveTime)，确保空闲时线程能被释放。
【强制】线程池不允许使用Executors 去创建，而是通过ThreadPoolExecutor 的方式，这样的处理方式让写的同学更加明确线程池的运行规则，规避资源耗尽的风险。
Executors 返回的线程池对象的弊端如下：
FixedThreadPool 和SingleThreadPool ：允许的请求队列长度为Integer.MAX_VALUE，可能会堆积大量的请求，从而导致OOM；
CachedThreadPool 和ScheduledThreadPool ：允许的创建线程数量为 Integer.MAX_VALUE，可能会创建大量的线程，从而导致OOM。

ThreadPoolExecutor线程池状态

ThreadPoolExecutor.java
/*** The runState provides the main lifecycle control, taking on values:** RUNNING:  Accept new tasks and process queued tasks(接收新任务并且会处理队列中的任务)* SHUTDOWN: Don't accept new tasks, but process queued tasks(不接受新任务,但是会处理队列中的任务)* STOP:     Don't accept new tasks, don't process queued tasks,*           and interrupt in-progress tasks(不接受新任务并且不会处理队列中的任务,中断正在执行中的任务)* TIDYING:  All tasks have terminated, workerCount is zero,*           the thread transitioning to state TIDYING*           will run the terminated() hook method(所有任务执行完成，workerCount为0,线程转到了状态TIDYING,会执行terminated()钩				子方法)* TERMINATED: terminated() has completed(terminated()方法已经执行完成)* 状态之间的转换如下* RUNNING -> SHUTDOWN*    On invocation of shutdown(), perhaps implicitly in finalize()(调用了shutdown()方法)* (RUNNING or SHUTDOWN) -> STOP*    On invocation of shutdownNow()(调用了shutdownNow()方法)* SHUTDOWN -> TIDYING*    When both queue and pool are empty(当队列和线程池为空)* STOP -> TIDYING*    When pool is empty(当线程池为空)* TIDYING -> TERMINATED*    When the terminated() hook method has completed(当terminated()钩子方法执行完成)*/
  • 线程池的shutdown()方法，将线程池由 RUNNING（运行状态）转换为 SHUTDOWN状态
  • 线程池的shutdownNow()方法，将线程池由RUNNING 或 SHUTDOWN 状态转换为 STOP 状态。

注：SHUTDOWN 状态 和 STOP 状态 先会转变为 TIDYING 状态，最终都会变为 TERMINATED

在这里插入图片描述

状态定义

    //ctl维护两个概念上的参数：workCount和runState。workCount表示有效的线程数量，runState表示线程池的运行状态。private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));private static final int COUNT_BITS = Integer.SIZE - 3;// 低29位表示线程数量private static final int CAPACITY   = (1 << COUNT_BITS) - 1;// runState is stored in the high-order bits 线程池运行状态存储在32位的高3位private static final int RUNNING    = -1 << COUNT_BITS;private static final int SHUTDOWN   =  0 << COUNT_BITS;private static final int STOP       =  1 << COUNT_BITS;private static final int TIDYING    =  2 << COUNT_BITS;private static final int TERMINATED =  3 << COUNT_BITS;

在这里插入图片描述

接下来我们来看看 ThreadPoolExecutor 是如何获取状态和线程数的。

runStateOf方法

//runStateOf()方法是用于获取线程池状态的方法,其中形参c一般是ctl变量,包含了状态和线程数
private static int runStateOf(int c)     { return c & ~CAPACITY; 
}

在这里插入图片描述

CAPACITY 取反后高三位置1，低29位置0。取反后的值与 ctl 进行 ‘与’ 操作。由于任何值 ‘与’ 1等于原值，‘与’ 0等于0。因此 ‘与’ 操作过后，ctl 的高3位保留原值，低29位置0。这样就将状态值从 ctl 中分离出来。

workerCountOf方法

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

在这里插入图片描述

workerCountOf 方法的分析思路与上述类似，就是把后29位从ctl中分离出来，获得活跃线程数

ctlOf方法

//ctlOf(rs, wc)通过状态值和线程数值计算出ctl值。rs是runState的缩写，wc是workerCount的缩写
private static int ctlOf(int rs, int wc) { return rs | wc; 
}

在这里插入图片描述

rs的后29位为0，wc的前三位为0，两者通过 ‘或’ 操作计算出来的最终值同时保留了rs的前3位和wc的后29位，即 ctl 值。

ThreadPoolExecutor构造函数

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;this.keepAliveTime = unit.toNanos(keepAliveTime);this.threadFactory = threadFactory;this.handler = handler;}
  • corePoolSize ：线程池中核心线程数的最大值
  • maximumPoolSize ：线程池中能拥有最多线程数
  • keepAliveTime ：表示空闲线程的存活时间
  • TimeUnit unit ：表示keepAliveTime的单位

当一个线程空闲，超过一定的时间（keepAliveTime）时，线程池会判断，如果当前运行的线程数大于 corePoolSize，那么这个线程就被回收。所以线程池的所有任务完成后，它最终会收缩到 corePoolSize 的大小。

注：如果线程池设置了allowCoreThreadTimeout参数为true（默认false），那么当空闲线程超过keepaliveTime后直接回收。（不会判断线程数是否大于corePoolSize）即：最终线程数会变为0。

  • workQueue ：用于存放任务的阻塞队列

ThreadPoolExecutor线程池推荐了三种等待队列，它们是：SynchronousQueue 、LinkedBlockingQueue 和 ArrayBlockingQueue。

1）有界队列：

SynchronousQueue ：一个不存储元素的阻塞队列，每个插入操作必须等到另一个线程调用移除操作，否则插入操作一直处于 阻塞状态，吞吐量通常要高于LinkedBlockingQueue，静态工厂方法Executors.newCachedThreadPool 使用了这个队列。

ArrayBlockingQueue：一个由数组支持的有界阻塞队列。此队列按 FIFO（先进先出）原则对元素进行排序。一旦创建了这样的缓存区，就不能再增加其容量。试图向已满队列中放入元素会导致操作受阻塞；试图从空队列中提取元素将导致类似阻塞。

2）无界队列：

LinkedBlockingQueue：基于链表结构的无界阻塞队列，它可以指定容量也可以不指定容量（实际上任何无限容量的队列/栈都是有容量的，这个容量就是Integer.MAX_VALUE）

PriorityBlockingQueue：是一个按照优先级进行内部元素排序的无界阻塞队列。队列中的元素必须实现 Comparable 接口，这样才能通过实现compareTo()方法进行排序。优先级最高的元素将始终排在队列的头部；PriorityBlockingQueue 不会保证优先级一样的元素的排序。

注意：keepAliveTime和maximumPoolSize及BlockingQueue的类型均有关系。如果BlockingQueue是无界的，那么永远不会触发maximumPoolSize，自然keepAliveTime也就没有了意义。

  • threadFactory ：指定创建线程的工厂。（可以不指定）

如果不指定线程工厂时，ThreadPoolExecutor会使用ThreadPoolExecutor.defaultThreadFactory创建线程。默认工厂创建的线程：同属于相同的线程组，具有同为 Thread.NORM_PRIORITY 的优先级，以及名为 “pool-XXX-thread-” 的线程名（XXX为创建线程时顺序序号），且创建的线程都是非守护进程。

  • handler ：表示当 workQueue 已满，且池中的线程数达到 maximumPoolSize 时，线程池拒绝添加新任务时采取的策略。（可以不指定）

在这里插入图片描述

execute()方法执行流程

public void execute(Runnable command) {if (command == null)throw new NullPointerException();int c = ctl.get();//如果workerCount(工作线程数) < 核心线程数if (workerCountOf(c) < corePoolSize) {//执行addWorker方法。addWorker()方法会在下面进行详细分析，这里可以简单理解为添加工作线程处理任务。这里的true表示：在小于核心线程数时添加worker线程，即添加核心线程。if (addWorker(command, true))return;//添加失败，重新获取ctl的值，防止在添加worker时状态改变c = ctl.get();}//运行到这里表示核心线程数已满，因此下面addWorker中第二个参数为false。//判断线程池是否是运行状态，如果是则尝试将任务添加至任务队列中if (isRunning(c) && workQueue.offer(command)) {//再次获取ctl的值，进行double-checkint recheck = ctl.get();//如果线程池为非运行状态，则尝试从任务队列中移除任务if (! isRunning(recheck) && remove(command))//移除成功后执行拒绝策略reject(command);//如果线程池是运行状态，或者从workQueue中删除任务失败（刚好有一个线程执行完毕，并消耗了这个任务）else if (workerCountOf(recheck) == 0)//创建一个非核心线程，null表示新建线程，但不会启动它addWorker(null, false);}//如果线程池是非运行状态，或者任务队列添加任务失败，再次尝试addWorker()方法else if (!addWorker(command, false))//失败，执行拒绝策略reject(command);}

double-check主要目的是判断刚加入workQueue阻塞队列的task是否能被执行

在这里插入图片描述

addWorker()方法执行流程

    private boolean addWorker(Runnable firstTask, boolean core) {//外层循环,负责判断线程池状态retry:for (;;) {int c = ctl.get();//线程池运行状态int rs = runStateOf(c);// Check if queue empty only if necessary.// 如果 rs >= SHUTDOWN,线程池state已经至少是shutdown状态,此时不允许接收新任务// 如果 rs == SHUTDOWN,firstTask为null,并且工作队列不为空,其中一个条件不满足则				无法继续执行addWorker()// 这里设置firstTask == null是因为:线程池在SHUTDOWN状态下,不允许添加新任务,但允许执行工作队列中剩余的任务。//总结:继续执行addWorker()需要满足要么线程池状态为RUNNING,要么线程池状态为SHUTDOWN,并且要满足队列中有任务以及firstTask==null,即只能添加新线程if (rs >= SHUTDOWN &&! (rs == SHUTDOWN &&firstTask == null &&! workQueue.isEmpty()))return false;内层循环,负责worker数量+1for (;;) {//获取活跃线程数int wc = workerCountOf(c);// 如果活跃线程数>=容量,不允许添加新任务// 如果core为true,表示创建核心线程,如果线程数>=核心线程数,则不允许创建线程// 如果core为false,表示创建空闲线程,如果线程数>=最大线程数,则不允许创建线程if (wc >= CAPACITY ||wc >= (core ? corePoolSize : maximumPoolSize))return false;//CAS操作,使得worker数量+1,成功则跳出retry循环if (compareAndIncrementWorkerCount(c))break retry;//CAS操作worker数量+1失败,再次读取ctlc = ctl.get();  // Re-read ctl// 如果运行状态已经改变,则从重新执行外层死循环// 如果运行状态未改变,继续执行内层死循环if (runStateOf(c) != rs)continue retry;// else CAS failed due to workerCount change; retry inner loop}}// 用于记录worker线程的状态boolean workerStarted = false;boolean workerAdded = false;Worker w = null;try {//new一个新的worker线程,每一个Worker内持有真正执行任务的线程w = new Worker(firstTask);final Thread t = w.thread;if (t != null) {final ReentrantLock mainLock = this.mainLock;//加锁,防止多线程并发环境下, 向workers中添加数据以及获取workers数量这两个过程出现并发的问题。mainLock.lock();try {// Recheck while holding lock.// Back out on ThreadFactory failure or if// shut down before lock acquired.int rs = runStateOf(ctl.get());//如果线程池在运行状态或者线程池已经shutdown,且firstTask==null（可能是workQueue中仍有未执行完成的任务，创建没有初始任务的worker线程执行)if (rs < SHUTDOWN ||(rs == SHUTDOWN && firstTask == null)) {//线程已经启动，抛非法线程状态异常if (t.isAlive()) // precheck that t is startablethrow new IllegalThreadStateException();//workers是一个HashSet<Worker>workers.add(w);//设置最大的池大小largestPoolSize,workerAdded设置为trueint s = workers.size();if (s > largestPoolSize)largestPoolSize = s;workerAdded = true;}} finally {mainLock.unlock();}//如果往HashSet中添加worker成功,启动线程if (workerAdded) {t.start();workerStarted = true;}}} finally {//如果启动线程失败,则处理失败逻辑if (! workerStarted)addWorkerFailed(w);}return workerStarted;}

runWorker()方法执行流程

addWorker()方法执行完成时会调用t.start()方法,这里的t是内部类Worker中成员变量

private final class Workerextends AbstractQueuedSynchronizerimplements Runnable{/** Thread this worker is running in.  Null if factory fails. */final Thread thread;/** Initial task to run.  Possibly null. */Runnable firstTask;/** Per-thread task counter */volatile long completedTasks;//new Worker()对象时会新创建线程newThread(),参数传的则是this,即worker对象Worker(Runnable firstTask) {setState(-1); // inhibit interrupts until runWorkerthis.firstTask = firstTask;this.thread = getThreadFactory().newThread(this);}//在调用t.start()方法时,实际上会调用该worker对象中的run方法public void run() {//执行runWorker方法runWorker(this);}
}

final void runWorker(Worker w) {Thread wt = Thread.currentThread();// 拷贝提交的任务,并将Worker中的firstTask置为 null,便于下一次重新赋值Runnable task = w.firstTask;w.firstTask = null;//new Worker()是state==-1,此处是调用Worker类的tryRelease()方法,将state置为0,而interruptIfStarted()中只有state>=0才允许调用中断w.unlock(); // allow interrupts//是否'突然完成',如果是由于异常导致的进入finally,那么completedAbruptly==true就是突然完成的boolean completedAbruptly = true;try {//如果当前有任务或者通过getTask()从任务队列中获取的任务不为nullwhile (task != null || (task = getTask()) != null) {//上锁,不是为了防止并发执行任务，为了在shutdown()时不终止正在运行的workerw.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//确保只有在线程状态>STOP时，才会被设置中断标示，否则清除中断标示if ((runStateAtLeast(ctl.get(), STOP) ||(Thread.interrupted() &&runStateAtLeast(ctl.get(), STOP))) &&!wt.isInterrupted())//当前线程调用interrupt()中断wt.interrupt();try {// ThreadPoolExecutor的钩子函数,用户可以继承ThreadPoolExecutor,并重写beforeExecute()方法,从而在任务执行前完成用户定制的操作逻辑beforeExecute(wt, task);Throwable thrown = null;try {//执行提交任务的run()方法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 {//同beforeExecute,只不过在任务执行完后执行afterExecute(task, thrown);}} finally {task = null;w.completedTasks++;w.unlock();}}completedAbruptly = false;} finally {//执行到这里表示任务队列中没了任务,或者线程池关闭了,此时需要将worker从缓存冲清除processWorkerExit(w, completedAbruptly);}}

getTask()方法执行流程

private Runnable getTask() {//如果上一次调用poll方法超时,下一次进入循环时，会清除一个Workerboolean timedOut = false; // Did the last poll() time out?for (;;) {int c = ctl.get();int rs = runStateOf(c);//只有以下两种情况下会继续往下执行,否则返回null//1.线程状态为RUNNING  2.线程状态为SHUTDOWN并且workQueue不为空if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {//循环的CAS减少worker数量,直到成功decrementWorkerCount();return null;}int wc = workerCountOf(c);//如果allowCoreThreadTimeOut为true,说明corePoolSize也需要超时检测boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;//如果worker数量大于maximumPoolSize或者允许超时,则尝试worker数量-1,失败则continue继续if ((wc > maximumPoolSize || (timed && timedOut))&& (wc > 1 || workQueue.isEmpty())) {if (compareAndDecrementWorkerCount(c))return null;continue;}try {//如果有超时检测则调用poll方法,否则调用take方法Runnable r = timed ?workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :workQueue.take();if (r != null)return r;//没有返回说明超时,那么在下一次内层循环时会进入worker count减一的步骤timedOut = true;} catch (InterruptedException retry) {timedOut = false;}}}

shutdown()方法执行流程

    public void shutdown() {final ReentrantLock mainLock = this.mainLock;//加全局锁mainLock.lock();try {// 校验是否有关闭线程池的权限checkShutdownAccess();// 修改线程池状态advanceRunState(SHUTDOWN);// 关闭所有空闲线程interruptIdleWorkers();//钩子函数,用户可以继承ThreadPoolExecutor并实现自定义钩子onShutdown(); // hook for ScheduledThreadPoolExecutor} finally {mainLock.unlock();}// 尝试关闭线程池tryTerminate();}

    private void advanceRunState(int targetState) {//自旋for (;;) {int c = ctl.get();// 判断当前线程池状态 >= SHUTDOWN是否成立,如果不成立的话,通过CAS进行修改if (runStateAtLeast(c, targetState) ||ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))break;}}

    private void interruptIdleWorkers() {interruptIdleWorkers(false);}private void interruptIdleWorkers(boolean onlyOne) {final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {//遍历workers集合for (Worker w : workers) {Thread t = w.thread// 判断worker线程是否已经被标记中断了,如果没有则尝试获取worker线程的锁if (!t.isInterrupted() && w.tryLock()) {try {//中断线程t.interrupt();} catch (SecurityException ignore) {} finally {w.unlock();}}// 如果onlyOne为true的话最多中断一个线程if (onlyOne)break;}} finally {mainLock.unlock();}}

shutdownNow()方法执行流程

    public List<Runnable> shutdownNow() {List<Runnable> tasks;final ReentrantLock mainLock = this.mainLock;//加全局锁mainLock.lock();try {// 校验关闭线程池权限checkShutdownAccess();// 修改线程池状态为STOPadvanceRunState(STOP);// 中断所有线程interruptWorkers();// 获取队列中所有正在等待处理的任务列表tasks = drainQueue();} finally {mainLock.unlock();}// 尝试关闭线程池tryTerminate();return tasks;}

    private void interruptWorkers() {final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {//遍历workers集合for (Worker w : workers)//只要Worker启动了就将其中断w.interruptIfStarted();} finally {mainLock.unlock();}}

参考文章

Java线程池实现原理及其在美团业务中的实践

ThreadPoolExecutor源码分析

Java线程池ThreadPoolExecutor使用和分析(二) - execute()原理

版权声明:

本网仅为发布的内容提供存储空间，不对发表、转载的内容提供任何形式的保证。凡本网注明“来源：XXX网络”的作品，均转载自其它媒体，著作权归作者所有，商业转载请联系作者获得授权，非商业转载请注明出处。

我们尊重并感谢每一位作者，均已注明文章来源和作者。如因作品内容、版权或其它问题，请及时与我们联系，联系邮箱：809451989@qq.com，投稿邮箱：809451989@qq.com

最新新闻

热搜词

数据中台-资产管理、数据安全 图说区块链 神一样的金融科技与未来社会 最近在上海浦东新区的租房经历 高效高质量SCI论文撰写及投稿 金山打字通五笔介绍及讲解 网络青年的英雄瞬间 中国青年报 2007年1月8日

声明：本站所有新闻及新闻图片来源于其他网站，如有侵权，请及时联系我们！

客户服务 | 关于我们 | 版权声明

版权所有：

Copyright 2024 尧图网 All Rights Reserved.QQ:809451989