一文弄懂Java中线程池原理
在工作中,我们经常使用线程池,但是你真的了解线程池的原理吗?同时,线程池工作原理和底层实现原理也是面试经常问的考题,所以,今天我们一起聊聊线程池的原理吧。
为什么要用线程池
使用线程池主要有以下三个原因:
- 降低资源消耗。通过重复利用已创建的线程降低线程创建和销毁造成的消耗。
- 提升响应速度。当任务到达时,任务可以不需要等到线程创建就能立即执行。
- 可以对线程做统一管理。线程是稀缺资源,如果无限制地创建,不仅会消耗系统资源,还会降低系统稳定性,使用线程池可以进行统一分配、调优和监控。
线程池的原理
Java中的线程池顶层接口是Executor
接口,ThreadPoolExecutor
是这个接口的实现类。
我们先看看ThreadPoolExecutor
类。
ThreadPoolExecutor提供的构造方法
// 七个参数的构造函数
public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler)
复制代码
我们先看看这些参数是什么意思:
- int corePoolSize:该线程池中核心线程数最大值
核心线程:线程池中有两类线程,核心线程和非核心线程。核心线程默认情况下会一直存在于线程池中,即使这个核心线程什么都不干(铁饭碗),而非核心线程如果长时间的闲置,就会被销毁(临时工)。
- int maximumPoolSize:该线程池中线程总数最大值 。
该值等于核心线程数量 + 非核心线程数量。
- long keepAliveTime:非核心线程闲置超时时长。
非核心线程如果处于闲置状态超过该值,就会被销毁。如果设置allowCoreThreadTimeOut(true),则会也作用于核心线程。
- TimeUnit unit:keepAliveTime的单位。
TimeUnit是一个枚举类型。
BlockingQueue workQueue:阻塞队列,维护着等待执行的Runnable任务对象。
常用的几个阻塞队列:
LinkedBlockingQueue:链式阻塞队列,底层数据结构是链表,默认大小是
Integer.MAX_VALUE
,也可以指定大小。ArrayBlockingQueue:数组阻塞队列,底层数据结构是数组,需要指定队列的大小。
SynchronousQueue:同步队列,内部容量为0,每个put操作必须等待一个take操作,反之亦然。
DelayQueue:延迟队列,该队列中的元素只有当其指定的延迟时间到了,才能够从队列中获取到该元素 。
ThreadFactory threadFactory
创建线程的工厂 ,用于批量创建线程,统一在创建线程时设置一些参数,如是否守护线程、线程的优先级等。如果不指定,会新建一个默认的线程工厂。
static class DefaultThreadFactory implements ThreadFactory {// 省略属性// 构造函数DefaultThreadFactory() {SecurityManager s = System.getSecurityManager();group = (s != null) ? s.getThreadGroup() :Thread.currentThread().getThreadGroup();namePrefix = "pool-" +poolNumber.getAndIncrement() +"-thread-";}// 省略
}
复制代码
RejectedExecutionHandler handler
拒绝处理策略,线程数量大于最大线程数就会采用拒绝处理策略,四种拒绝处理的策略为 :
- ThreadPoolExecutor.AbortPolicy:默认拒绝处理策略,丢弃任务并抛出RejectedExecutionException异常。
- ThreadPoolExecutor.DiscardPolicy:丢弃新来的任务,但是不抛出异常。
- ThreadPoolExecutor.DiscardOldestPolicy:丢弃队列头部(最旧的)的任务,然后重新尝试执行程序(如果再次失败,重复此过程)。
- ThreadPoolExecutor.CallerRunsPolicy:由调用线程处理该任务。
ThreadPoolExecutor的策略
线程池本身有一个调度线程,这个线程就是用于管理布控整个线程池里的各种任务和事务,例如创建线程、销毁线程、任务队列管理、线程队列管理等等。
故线程池也有自己的状态。ThreadPoolExecutor
类中使用了一些final int
常量变量来表示线程池的状态 ,分别为RUNNING、SHUTDOWN、STOP、TIDYING 、TERMINATED。
// runState is stored in the high-order bits
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;
复制代码
线程池创建后处于RUNNING状态。
调用shutdown()方法后处于SHUTDOWN状态,线程池不能接受新的任务,清除一些空闲worker,不会等待阻塞队列的任务完成。
调用shutdownNow()方法后处于STOP状态,线程池不能接受新的任务,中断所有线程,阻塞队列中没有被执行的任务全部丢弃。此时,poolsize=0,阻塞队列的size也为0。
当所有的任务已终止,ctl记录的”任务数量”为0,线程池会变为TIDYING状态。接着会执行terminated()函数。
线程池处在TIDYING状态时,执行完terminated()方法之后,就会由 TIDYING -> TERMINATED, 线程池被设置为TERMINATED状态。
线程池主要的任务处理流程
处理任务的核心方法是execute
,我们看看 JDK 1.8 源码中ThreadPoolExecutor
是如何处理线程任务的:
// JDK 1.8
public void execute(Runnable command) {if (command == null)throw new NullPointerException(); int c = ctl.get();// 1.当前线程数小于corePoolSize,则调用addWorker创建核心线程执行任务if (workerCountOf(c) < corePoolSize) {if (addWorker(command, true))return;c = ctl.get();}// 2.如果不小于corePoolSize,则将任务添加到workQueue队列。if (isRunning(c) && workQueue.offer(command)) {int recheck = ctl.get();// 2.1 如果isRunning返回false(状态检查),则remove这个任务,然后执行拒绝策略。if (! isRunning(recheck) && remove(command))reject(command);// 2.2 线程池处于running状态,但是没有线程,则创建线程else if (workerCountOf(recheck) == 0)addWorker(null, false);}// 3.如果放入workQueue失败,则创建非核心线程执行任务,// 如果这时创建非核心线程失败(当前线程总数不小于maximumPoolSize时),就会执行拒绝策略。else if (!addWorker(command, false))reject(command);
}
复制代码
ctl.get()
是获取线程池状态,用int
类型表示。第二步中,入队前进行了一次isRunning
判断,入队之后,又进行了一次isRunning
判断。
为什么要二次检查线程池的状态?
在多线程的环境下,线程池的状态是时刻发生变化的。很有可能刚获取线程池状态后线程池状态就改变了。判断是否将command
加入workqueue
是线程池之前的状态。倘若没有二次检查,万一线程池处于非RUNNING状态(在多线程环境下很有可能发生),那么command
永远不会执行。
总结一下处理流程
- 线程总数量 < corePoolSize,无论线程是否空闲,都会新建一个核心线程执行任务(让核心线程数量快速达到corePoolSize,在核心线程数量 < corePoolSize时)。注意,这一步需要获得全局锁。
- 线程总数量 >= corePoolSize时,新来的线程任务会进入任务队列中等待,然后空闲的核心线程会依次去缓存队列中取任务来执行(体现了线程复用)。
- 当缓存队列满了,说明这个时候任务已经多到爆棚,需要一些“临时工”来执行这些任务了。于是会创建非核心线程去执行这个任务。注意,这一步需要获得全局锁。
- 缓存队列满了, 且总线程数达到了maximumPoolSize,则会采取上面提到的拒绝策略进行处理。
整个过程如图所示:
ThreadPoolExecutor如何做到线程复用的?
我们知道,一个线程在创建的时候会指定一个线程任务,当执行完这个线程任务之后,线程自动销毁。但是线程池却可以复用线程,即一个线程执行完线程任务后不销毁,继续执行另外的线程任务。那么,线程池如何做到线程复用呢?
原来,ThreadPoolExecutor在创建线程时,会将线程封装成工作线程worker,并放入工作线程组中,然后这个worker反复从阻塞队列中拿任务去执行。
这里的addWorker
方法是在上面提到的execute
方法里面调用的,先看看上半部分:
// ThreadPoolExecutor.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.if (rs >= SHUTDOWN &&! (rs == SHUTDOWN &&firstTask == null &&! workQueue.isEmpty()))return false;for (;;) {int wc = workerCountOf(c);if (wc >= CAPACITY ||// 1.如果core是ture,证明需要创建的线程为核心线程,则先判断当前线程是否大于核心线程// 如果core是false,证明需要创建的是非核心线程,则先判断当前线程数是否大于总线程数// 如果不小于,则返回falsewc >= (core ? corePoolSize : maximumPoolSize))return false;if (compareAndIncrementWorkerCount(c))break retry;c = ctl.get(); // Re-read ctlif (runStateOf(c) != rs)continue retry;// else CAS failed due to workerCount change; retry inner loop}}
复制代码
上半部分主要是判断线程数量是否超出阈值,超过了就返回false。我们继续看下半部分:
// ThreadPoolExecutor.addWorker方法源码下半部分boolean workerStarted = false;boolean workerAdded = false;Worker w = null;try {// 1.创建一个worker对象w = new Worker(firstTask);// 2.实例化一个Thread对象final Thread t = w.thread;if (t != null) {// 3.线程池全局锁final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {// Recheck while holding lock.// Back out on ThreadFactory failure or if// shut down before lock acquired.int rs = runStateOf(ctl.get());if (rs < SHUTDOWN ||(rs == SHUTDOWN && firstTask == null)) {if (t.isAlive()) // precheck that t is startablethrow new IllegalThreadStateException();workers.add(w);int s = workers.size();if (s > largestPoolSize)largestPoolSize = s;workerAdded = true;}} finally {mainLock.unlock();}if (workerAdded) {// 4.启动这个线程t.start();workerStarted = true;}}} finally {if (! workerStarted)addWorkerFailed(w);}return workerStarted;
}
复制代码
创建worker
对象,并初始化一个Thread
对象,然后启动这个线程对象。
我们接着看看Worker
类,仅展示部分源码:
// Worker类部分源码
private final class Worker extends AbstractQueuedSynchronizer implements Runnable{final Thread thread;Runnable firstTask;Worker(Runnable firstTask) {setState(-1); // inhibit interrupts until runWorkerthis.firstTask = firstTask;this.thread = getThreadFactory().newThread(this);}public void run() {runWorker(this);}//其余代码略...
}
复制代码
Worker
类实现了Runnable
接口,所以Worker
也是一个线程任务。在构造方法中,创建了一个线程,线程的任务就是自己。故addWorker
方法调用addWorker方法源码下半部分中的第4步t.start
,会触发Worker
类的run
方法被JVM调用。
我们再看看runWorker
的逻辑:
// Worker.runWorker方法源代码
final void runWorker(Worker w) {Thread wt = Thread.currentThread();Runnable task = w.firstTask;w.firstTask = null;// 1.线程启动之后,通过unlock方法释放锁w.unlock(); // allow interruptsboolean completedAbruptly = true;try {// 2.Worker执行firstTask或从workQueue中获取任务,如果getTask方法不返回null,循环不退出while (task != null || (task = getTask()) != null) {// 2.1进行加锁操作,保证thread不被其他线程中断(除非线程池被中断)w.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// 2.2检查线程池状态,倘若线程池处于中断状态,当前线程将中断。 if ((runStateAtLeast(ctl.get(), STOP) ||(Thread.interrupted() &&runStateAtLeast(ctl.get(), STOP))) &&!wt.isInterrupted())wt.interrupt();try {// 2.3执行beforeExecute beforeExecute(wt, task);Throwable thrown = null;try {// 2.4执行任务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 {// 2.5执行afterExecute方法 afterExecute(task, thrown);}} finally {task = null;w.completedTasks++;// 2.6解锁操作w.unlock();}}completedAbruptly = false;} finally {processWorkerExit(w, completedAbruptly);}
}
复制代码
首先去执行创建这个worker时就有的任务,当执行完这个任务后,worker的生命周期并没有结束,在while
循环中,worker会不断地调用getTask
方法从阻塞队列中获取任务然后调用task.run()
执行任务,从而达到复用线程的目的。只要getTask
方法不返回null
,此线程就不会退出。
当然,核心线程池中创建的线程想要拿到阻塞队列中的任务,先要判断线程池的状态,如果STOP或者TERMINATED,返回null
。
最后看看getTask
方法的实现:
// Worker.getTask方法源码
private Runnable getTask() {boolean timedOut = false; // Did the last poll() time out?for (;;) {int c = ctl.get();int rs = runStateOf(c);// Check if queue empty only if necessary.if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {decrementWorkerCount();return null;}int wc = workerCountOf(c);// Are workers subject to culling?// 1.allowCoreThreadTimeOut变量默认是false,核心线程即使空闲也不会被销毁// 如果为true,核心线程在keepAliveTime内仍空闲则会被销毁。 boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;// 2.如果运行线程数超过了最大线程数,但是缓存队列已经空了,这时递减worker数量。 // 如果有设置允许线程超时或者线程数量超过了核心线程数量,// 并且线程在规定时间内均未poll到任务且队列为空则递减worker数量if ((wc > maximumPoolSize || (timed && timedOut))&& (wc > 1 || workQueue.isEmpty())) {if (compareAndDecrementWorkerCount(c))return null;continue;}try {// 3.如果timed为true(想想哪些情况下timed为true),则会调用workQueue的poll方法获取任务.// 超时时间是keepAliveTime。如果超过keepAliveTime时长,// poll返回了null,上边提到的while循序就会退出,线程也就执行完了。// 如果timed为false(allowCoreThreadTimeOut为false// 且wc > corePoolSize为false),则会调用workQueue的take方法阻塞在当前。// 队列中有任务加入时,线程被唤醒,take方法返回任务,并执行。Runnable r = timed ?workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :workQueue.take();if (r != null)return r;timedOut = true;} catch (InterruptedException retry) {timedOut = false;}}
}
复制代码
核心线程的会一直卡在workQueue.take
方法,被阻塞并挂起,不会占用CPU资源,直到拿到Runnable
然后返回(当然如果allowCoreThreadTimeOut设置为true
,那么核心线程就会去调用poll
方法,因为poll
可能会返回null
,所以这时候核心线程满足超时条件也会被销毁)。
非核心线程会workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) ,如果超时还没有拿到,下一次循环判断compareAndDecrementWorkerCount就会返回null
,Worker对象的run()
方法循环体的判断为null
,任务结束,然后线程被系统回收 。
四种常见的线程池
Executors
类中提供的几个静态方法来创建线程池。
newCachedThreadPool
public static ExecutorService newCachedThreadPool() {return new ThreadPoolExecutor(0, Integer.MAX_VALUE,60L, TimeUnit.SECONDS,new SynchronousQueue<Runnable>());
}
复制代码
CacheThreadPool
的运行流程如下:
- 提交任务进线程池。
- 因为corePoolSize为0的关系,不创建核心线程,线程池最大为Integer.MAX_VALUE。
- 尝试将任务添加到SynchronousQueue队列。
- 如果SynchronousQueue入列成功,等待被当前运行的线程空闲后拉取执行。如果当前没有空闲线程,那么就创建一个非核心线程,然后从SynchronousQueue拉取任务并在当前线程执行。
- 如果SynchronousQueue已有任务在等待,入列操作将会阻塞。
当需要执行很多短时间的任务时,CacheThreadPool的线程复用率比较高, 会显著的提高性能。而且线程60s后会回收,意味着即使没有任务进来,CacheThreadPool并不会占用很多资源。
newFixedThreadPool
public static ExecutorService newFixedThreadPool(int nThreads) {return new ThreadPoolExecutor(nThreads, nThreads,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>());
}
复制代码
核心线程数量和总线程数量相等,都是传入的参数nThreads,所以只能创建核心线程,不能创建非核心线程。因为LinkedBlockingQueue的默认大小是Integer.MAX_VALUE,故如果核心线程空闲,则交给核心线程处理;如果核心线程不空闲,则入列等待,直到核心线程空闲。
与CachedThreadPool的区别:
- 因为 corePoolSize == maximumPoolSize ,所以FixedThreadPool只会创建核心线程。 而CachedThreadPool因为corePoolSize=0,所以只会创建非核心线程。
- 在 getTask() 方法,如果队列里没有任务可取,线程会一直阻塞在 LinkedBlockingQueue.take() ,线程不会被回收。 CachedThreadPool会在60s后收回。
- 由于线程不会被回收,会一直卡在阻塞,所以没有任务的情况下, FixedThreadPool占用资源更多。
- 都几乎不会触发拒绝策略,但是原理不同。FixedThreadPool是因为阻塞队列可以很大(最大为Integer最大值),故几乎不会触发拒绝策略;CachedThreadPool是因为线程池很大(最大为Integer最大值),几乎不会导致线程数量大于最大线程数,故几乎不会触发拒绝策略。
newSingleThreadExecutor
public static ExecutorService newSingleThreadExecutor() {return new FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>()));
}
复制代码
有且仅有一个核心线程( corePoolSize == maximumPoolSize=1),使用了LinkedBlockingQueue(容量很大),所以,不会创建非核心线程。所有任务按照先来先执行的顺序执行。如果这个唯一的线程不空闲,那么新来的任务会存储在任务队列里等待执行。
newScheduledThreadPool
创建一个定长线程池,支持定时及周期性任务执行。
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {return new ScheduledThreadPoolExecutor(corePoolSize);
}//ScheduledThreadPoolExecutor():
public ScheduledThreadPoolExecutor(int corePoolSize) {super(corePoolSize, Integer.MAX_VALUE,DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,new DelayedWorkQueue());
}
复制代码
四种常见的线程池基本够我们使用了,但是《阿里巴巴开发手册》不建议我们直接使用Executors类中的线程池,而是通过ThreadPoolExecutor
的方式,这样的处理方式让写的同学需要更加明确线程池的运行规则,规避资源耗尽的风险。
但如果你及团队本身对线程池非常熟悉,又确定业务规模不会大到资源耗尽的程度(比如线程数量或任务队列长度可能达到Integer.MAX_VALUE)时,其实是可以使用JDK提供的这几个接口的,它能让我们的代码具有更强的可读性。
小结
在工作中,很多人因为不了解线程池的实现原理,把线程池配置错误,从而导致各种问题。希望你们阅读完本文,能够学会合理的使用线程池。
对于真正想弄懂java并发编程的小伙伴,网上的文章还有视频缺乏系统性,我建议大家还是买点书籍看看,我推荐两本我看过的书。
《Java并发编程实战》:这本书深入浅出地介绍了Java线程和并发,是一本非常棒的Java并发参考手册。
《Java并发编程艺术》:Java并发编程的概念本来就比较复杂,我们需要的是一本能够把原理解释清楚的书籍,而这本《Java并发编程的艺术》书是国内作者写的Java并发书籍,刚好就比上面那一本更简单易懂,至少我自己看下来是这样的感觉。
一文弄懂Java中线程池原理相关推荐
- deque stack java_一文弄懂java中的Queue家族
简介 java中Collection集合有三大家族List,Set和Queue.当然Map也算是一种集合类,但Map并不继承Collection接口. List,Set在我们的工作中会经常使用,通常用 ...
- 一文弄懂java中的Queue家族
文章目录 简介 Queue接口 Queue的分类 BlockingQueue Deque TransferQueue 总结 java中Queue家族简介 简介 java中Collection集合有三大 ...
- queue double java_一文弄懂java中的Queue家族
java中Queue家族简介 简介 java中Collection集合有三大家族List,Set和Queue.当然Map也算是一种集合类,但Map并不继承Collection接口. List,Set在 ...
- 一文搞懂Java自定义线程池参数
背景: 线程是一个开发经常使用的东西,但是实际工作中很少有新建一个线程去执行任务,线程的创建和销毁都会耗费资源,我们一般都会用线程池来完成线程的创建和销毁. 多说一句 JDK自带的线程池不建议大家去使 ...
- 一文弄懂神经网络中的反向传播法
最近在看深度学习的东西,一开始看的吴恩达的UFLDL教程,有中文版就直接看了,后来发现有些地方总是不是很明确,又去看英文版,然后又找了些资料看,才发现,中文版的译者在翻译的时候会对省略的公式推导过程进 ...
- java中date类型如何赋值_一文读懂java中的Reference和引用类型
简介 java中有值类型也有引用类型,引用类型一般是针对于java中对象来说的,今天介绍一下java中的引用类型.java为引用类型专门定义了一个类叫做Reference.Reference是跟jav ...
- 一文弄懂神经网络中的反向传播法——BackPropagation【转】
本文转载自:https://www.cnblogs.com/charlotte77/p/5629865.html 一文弄懂神经网络中的反向传播法--BackPropagation 最近在看深度学习的东 ...
- Java中线程池,你真的会用吗
转载自 Java中线程池,你真的会用吗 在<深入源码分析Java线程池的实现原理>这篇文章中,我们介绍过了Java中线程池的常见用法以及基本原理. 在文中有这样一段描述: 可以通过Ex ...
- Java中线程池,你真的会用吗?
在<深入源码分析Java线程池的实现原理>这篇文章中,我们介绍过了Java中线程池的常见用法以及基本原理. 在文中有这样一段描述: 可以通过Executors静态工厂构建线程池,但一般不建 ...
最新文章
- java封装对象实体类_Java 接口自动化系列--实体类之entity封装
- 二值化网络如何训练?这篇ICML 2021论文给你答案
- STL初探——构造和析构的基本工具: construct()和destroy()
- python3.6运行界面_Electron+Python界面开发(通过zerorpc)
- 后端python基础
- python链表中删除一个节点数据_python实现单链表中删除倒数第K个节点的方法
- java内部类基础(静态内部类)
- Java中Properties类的学习总结
- mysql shell命令和作用_MySQL的一些功能实用的Linux shell脚本分享
- makefile phony
- Java抽奖抢购算法
- Unity编辑器扩展
- Nmap库ICMP主机探测
- 计算机无法安装cad怎么办,电脑安装AutoCAD提示未正常卸载无法安装如何解决
- 台式电脑怎么组装步骤_台式电脑怎么组装 U大侠分享组装台式机的具体步骤
- linux删除 grub rescue,删除Ubuntu后 开机grub rescue无法进入BIOS
- SQLServer的top 100 percent用法
- python爬虫进程和线程的区别_熬了两个通宵写的!终于把多线程和多进程彻底讲明白了!...
- R语言回归分析-改进异常值
- python---用单个葫芦丝音节合成完整的曲目
热门文章
- 运动用什么耳机最好、推荐六款最好的运动耳机
- Hive中数据类型和隐式转换
- bert-embedding:如何得到BERT训练的词向量
- python文件 py、pyc、pyo、pyd文件
- 第十一届蓝桥杯大赛(2020年)软件类省赛第二场 Python大学组 真题+个人解答
- WPF DevExpress grid 设置自定义筛选栏的条件为包含关系
- AI迷航(全集) 读后感
- Unreal Engine 4 手绘风滤镜(Paint Filter)即 桑原滤镜(Kuwahara Filter)教程(上)
- 联想拯救者Y7000p安装ubuntu18.04历险记
- psql :致命错误:用户postgres Password 认证失败