网站优化排名易下拉稳定,网站维护 公司简介,万户高端网站建设,中国建设建行网站ThreadPoolExecutor线程池详解
1. 背景
项目最近的迭代中使用到了ThreadPoolExecutor线程池#xff0c;之前都只是知道怎么用#xff0c;没有了解过线程池的底层原理#xff0c;项目刚上线#xff0c;有时间整理一下线程池的用法#xff0c;学习一下线程池的底层实现与工…ThreadPoolExecutor线程池详解
1. 背景
项目最近的迭代中使用到了ThreadPoolExecutor线程池之前都只是知道怎么用没有了解过线程池的底层原理项目刚上线有时间整理一下线程池的用法学习一下线程池的底层实现与工作原理。
2. ThreadPoolExecutor工作原理
2.1 构造方法
public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueueRunnable 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;}
2.2 线程池的使用
worker
/*** author itender* date 2023/8/7 14:41* desc*/
public class Worker implements Runnable {private String command;public Worker(String s) {this.command s;}Overridepublic void run() {System.out.println(Thread.currentThread().getName() command startTie DateUtil.now());processCommand();System.out.println(Thread.currentThread().getName() command endTime DateUtil.now());}private void processCommand() {try {Thread.sleep(1000);System.out.println(Thread.currentThread().getName() command 处理任务逻辑。。。。。。。。);} catch (InterruptedException e) {Thread.currentThread().interrupt();e.printStackTrace();}}Overridepublic String toString() {return this.command;}
}线程池
/*** author itender* date 2023/8/7 14:37* desc*/
public class ThreadPoolExecutorDemo {private static final int CORE_POOL_SIZE 5;private static final int MAX_POOL_SIZE 10;private static final int QUEUE_CAPACITY 100;private static final Long KEEP_ALIVE_TIME 1L;public static void main(String[] args) {ThreadPoolExecutor threadPoolExecutor new ThreadPoolExecutor(// 核心线程数 5CORE_POOL_SIZE,// 最大线程数 10MAX_POOL_SIZE,// 超过核心线程数线程最大存活时间KEEP_ALIVE_TIME,// 时间单位TimeUnit.MINUTES,// 工作队列最大值new ArrayBlockingQueue(QUEUE_CAPACITY),// 线程工厂创建线程的时候使用r - {Thread thread new Thread(r);thread.setName(pool-);return thread;},new ThreadPoolExecutor.CallerRunsPolicy());for (int i 0; i 10; i) {// 创建任务Worker myRunnable new Worker( i);// 执行任务threadPoolExecutor.execute(myRunnable);}// 种植线程池不接受新任务但是有工作线程处理队列中的任务threadPoolExecutor.shutdown();while (!threadPoolExecutor.isTerminated()) {}System.out.println(Finished All Threads!);}
}2.3 核心参数
2.3.1 核心参数详解
corePoolSize核心线程数任务队列没有达到队列最大容量时最大可以同时运行的线程数。maximumPoolSize最大线程数。当任务队列中存储的任务达到队列的容量时当前可以同时运行的线程数量变为最大线程数。keepAliveTime线程池中的线程数量超过corePoolSize时如果没有新任务提交核心线程外的线程不会立即销毁而是等待直到等待的时间超过了keepAliveTime才会被销毁回收。unitkeepAliveTime参数的时间单位。workQueue工作队列。当有新的任务提交的时候会先判断当前运行的线程数是否达到核心线程数如果达到核心线程数则会把新提交的任务放到工作队列中。threadFactory线程工厂创建新的线程时会使用。handler拒绝策略。
2.3.2 拒绝策略
如果当前同时运行的线程数量达到最大线程数量并且队列也已经被放满了任务时ThreadPoolTaskExecutor 定义一些策略:
AbortPolicy默认拒绝策略。抛出RejectExecutionException来拒绝新任务的处理。CallerRunsPolicy调用当前提交任务的线程来执行任务。一般不希望任务丢失会选用这种策略但从实际角度来看原来的异步调用意图会退化为同步调用。DiscardPolicy不处理新任务直接丢弃。DiscardOldestPolicy丢弃最早的未处理的任务。
2.4 执行流程 2.5 线程池状态
2.5.1 线程池核心属性ctl // ctl本质是 Integer 型变量进行了原子性的封装// ctl表示两种状态// 高3位线程池当前的状态// 低29位线程池当前工作线程的数量private final AtomicInteger ctl new AtomicInteger(ctlOf(RUNNING, 0));// COUNT_BITS 的值为 29整型Integer.SIZE 32 位;private static final int COUNT_BITS Integer.SIZE - 3;// CAPACITY (1 29) - 1; 1左移29位减去1即1*2^29-1// 0001 1111 1111 1111 1111 1111 1111 1111// 低29位用来表示线程池的最大线程容量private static final int CAPACITY (1 COUNT_BITS) - 1;// 高3位用来表示线程池5种状态// 111 运行状态private static final int RUNNING -1 COUNT_BITS;// 000 shutdown状态private static final int SHUTDOWN 0 COUNT_BITS;// 001 停止状态private static final int STOP 1 COUNT_BITS;// 010 过渡状态private static final int TIDYING 2 COUNT_BITS;// 011 中介状态private static final int TERMINATED 3 COUNT_BITS;// 根据ctl的值计算当前线程池的状态// 计算方式c 与 非capacityprivate static int runStateOf(int c) { return c ~CAPACITY; }// 根据ctl的值计算线程池当前运行的线程的容量private static int workerCountOf(int c) { return c CAPACITY; }// 通过运行状态和工作线程数计算ctl的值或运算private static int ctlOf(int rs, int wc) { return rs | wc; }private static boolean runStateLessThan(int c, int s) {return c s;private static boolean runStateAtLeast(int c, int s) {return c s;}private static boolean isRunning(int c) {return c SHUTDOWN;}/*** Attempts to CAS-increment the workerCount field of ctl.*/private boolean compareAndIncrementWorkerCount(int expect) {return ctl.compareAndSet(expect, expect 1);}/*** Attempts to CAS-decrement the workerCount field of ctl.*/private boolean compareAndDecrementWorkerCount(int expect) {return ctl.compareAndSet(expect, expect - 1);}/*** Decrements the workerCount field of ctl. This is called only on* abrupt termination of a thread (see processWorkerExit). Other* decrements are performed within getTask.*/private void decrementWorkerCount() {do {} while (! compareAndDecrementWorkerCount(ctl.get()));}2.5.2 状态切换 RUNNING 能接受新提交的任务并且也能处理阻塞队列中的任务。SHUTDOWN关闭状态不再接受新提交的任务但却可以继续处理阻塞队列中已保存的任务。在线程池处于 RUNNING 状态时调用 shutdown() 方法会使线程池进入到该状态。finalize() 方法在执行过程中也会调用 shutdown() 方法进入该状态。STOP不能接受新任务也不处理队列中的任务会中断正在处理任务的线程。在线程池处于 RUNNING 或 SHUTDOWN 状态时调用 shutdownNow() 方法会使线程池进入到该状态。TIDYING如果所有的任务都已终止了workerCount (有效线程数) 为0线程池进入该状态后会调用 terminated() 方法进入 TERMINATED 状态。TERMINATED在terminated()方法执行完后进入该状态默认 terminated() 方法中什么也没有做。
3. 源码分析
3.1 execute方法
源码
public void execute(Runnable command) {// 判断任务是否为空,如果任务为空抛出空指针异常if (command null)throw new NullPointerException();// 获取ctl属性int c ctl.get();// 判断当前工作线程数量是否小于核心线程的数量if (workerCountOf(c) corePoolSize) {// 工作线程数小于核心线程数创建一个核心线程执行command任务if (addWorker(command, true))// 创建核心线程成功直接返回return;// 并发情况下添加核心线程失败需要重新获取ctl属性c ctl.get();}// 创建核心线程失败当前工作线程数量大于或等于核心线程数量corePoolSize// 判断线程池的状态是否为running如果是添加任务到工作队列中放入任务失败返回falseif (isRunning(c) workQueue.offer(command)) {// 任务添加到队列成功再次获取ctl属性int recheck ctl.get();// 二次检查判断线程池的状态是否为running如果不是队列中移除刚刚添加的任务if (!isRunning(recheck) remove(command))// 执行拒绝策略reject(command);// 1.任务添加到队列// 2.线程池可能是running状态// 3.传入的任务可能从任务队列中移除失败移除失败的唯一可能就是任务已经被执行了// 判断工作线程数量是否为0else if (workerCountOf(recheck) 0)// 工作线程数量为0// 工作队列中有任务在排队添加一个空任务创建非核心线程执行队列中等待的任务addWorker(null, false);}// 创建核心线程失败// 线程池状态不是running状态// 线程池可能是running状态但是任务队列已经满了// 添加任务到工作队列失败创建非核心线程执行任务else if (!addWorker(command, false))// 创建非核心线程失败执行拒绝策略reject(command);}第一点核心通过execute方法源码可以看出线程池具体的执行流程以及一些避免并发情况的判断。
第二点核心线程池为什么会添加空任务非核心线程到线程池。
这里是一个疑惑点为什么需要二次检查线程池的运行状态当前工作线程数量为0尝试创建一个非核心线程并且传入的任务对象为null这个可以看API注释
如果一个任务成功加入任务队列我们依然需要二次检查是否需要添加一个工作线程因为所有存活的工作线程有可能在最后一次检查之后已经终结或者执行当前方法的时候线程池是否已经shutdown了。所以我们需要二次检查线程池的状态必须时把任务从任务队列中移除或者在没有可用的工作线程的前提下新建一个工作线程。
3.2 addWorker方法
源码
private boolean addWorker(Runnable firstTask, boolean core) {// for循环标识// 对线程池当前状态和当前工作线程数量的判断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);// 1. 如果传入的core为true表示将要创建核心线程通过wc和corePoolSize判断如果wc corePoolSize则返回false表示创建核心线程失败// 2. 如果传入的core为false表示将要创非建核心线程通过wc和maximumPoolSize判断如果wc maximumPoolSize则返回false表示创建非核心线程失败// core参数为false说明工作队列已经满了线程池大小变为maximumPoolSize最大线程数if (wc CAPACITY ||wc (core ? corePoolSize : maximumPoolSize))return false;// CAS更新工作线程数wc原子操作将workCount的数量加1更新成功则直接跳出最外层循环if (compareAndIncrementWorkerCount(c))break retry;// CAS更新工作线程数失败判断线程池的状态是否从running编程shutdown如果线程池的状态改变了在执行上面的操作c ctl.get(); // Re-read ctl// 如果线程池状态已经变成shutdown跳过最外层本次循环执行下一次循环if (runStateOf(c) ! rs)continue retry;// 如果线程池状态依然是RUNNINGCAS更新工作线程数wc失败说明有可能是并发更新导致的失败则在内层循环重试即可 // else CAS failed due to workerCount change; retry inner loop}}// 工作线程是否启动成功boolean workerStarted false;// 工作线程是否创建成功boolean workerAdded false;Worker w null;try {w new Worker(firstTask);final Thread t w.thread;if (t ! null) {// 加锁因为会改变一些指标值和非线程安全的集合final ReentrantLock mainLock this.mainLock;// 加锁mainLock.lock();try {// 获取线程池状态int rs runStateOf(ctl.get());//rs SHUTDOWN 如果线程池状态依然为RUNNING,并且线程的状态是存活的话就会将工作线程添加到工作线程集合中//(rsSHUTDOWN firstTask null)如果线程池状态小于STOP也就是RUNNING或者SHUTDOWN状态下同时传入的任务实例firstTask为null则需要添加到工作线程集合和启动新的Worker// 对于2换言之如果线程池处于SHUTDOWN状态下同时传入的任务实例firstTask不为null则不会添加到工作线程集合和启动新的Worker// 这一步其实有可能创建了新的Worker实例但是并不启动临时对象没有任何强引用这种Worker有可能成功下一轮GC被收集的垃圾对象// firstTask null证明只新建线程而不执行任务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();}// 如果成功添加工作线程则调用Worker内部的线程实例t的Thread#start()方法启动真实的线程实例if (workerAdded) {// 启动线程标识线程启动成功t.start();workerStarted true;}}} finally {// 线程启动失败需要从工作线程中移除对应的Workerif (!workerStarted)addWorkerFailed(w);}return workerStarted;
}4. 线程池常见问题
4.1 execute() 和submit()的区别
execute()方法用于提交不需要返回值的任务所以无法判断任务是否被线程池执行成功与否submit()方法用于提交需要返回值的任务。线程池会返回一个 Future 类型的对象通过这个 Future 对象可以判断任务是否执行成功并且可以通过 Future 的 get()方法来获取返回值get()方法会阻塞当前线程直到任务完成而使用 getlong timeoutTimeUnit unit方法的话如果在 timeout 时间内任务还没有执行完就会抛出 java.util.concurrent.TimeoutException。
4.3 阻塞队列的作用 一般的队列只能是有限长度的缓冲区一旦超出缓冲长度就无法保留了。阻塞队列通过阻塞可以保留住当前想要继续入队的任务。 阻塞队列可以在队列中没有任务时阻塞想要获取任务的线程使其进入wait状态释放cpu资源。 阻塞队列带有阻塞和唤醒的功能不需要额外处理无任务执行时线程池利用阻塞队列的take方法挂起从而维持核心线程的存活不至于一直占用cpu资源。
4.2 为什么先添加队列而不是先创建最大线程 在创建新线程的时候是要获取全局锁的这时候其他线程会被阻塞影响整体效率。 在核心线程已满时如果任务继续增加那么放在队列中等队列满了而任务还在增加那么就要创建临时线程了这样代价低。
5. 参考文章
https://www.throwx.cn/2020/08/23/java-concurrency-thread-pool-executor/
https://javaguide.cn/java/concurrent/java-thread-pool-summary.html#%E7%BA%BF%E7%A8%8B%E6%B1%A0%E5%8E%9F%E7%90%86%E5%88%86%E6%9E%90
