ThreadPoolExecutor源码学习(一)核心属性及应用

在Executors类中&＃xff0c;jdk提供了四个线程池可以使用&＃xff0c;分别是
newCachedThreadPool&＃xff1a;
corePoolSize是0&＃xff0c;允许创建的最大线程数量是Integer.MAX_VALUE
newScheduledThreadPool&＃xff1a;
允许创建的最大线程数量是Integer.MAX_VALUE
newFixedThreadPool:
任务队列允许的最大长度是Integer.MAX_VALUE
newSingleThreadExecutor:
任务队列允许的最大长度是Integer.MAX_VALUE

这四种是阿里巴巴Java开发规范中所不推荐的&＃xff0c;因为这四个线程池中&＃xff0c;要么是最大线程数无界&＃xff0c;要么阻塞队列无界&＃xff0c;可能会造成OOM

推荐使用自定义的线程池&＃xff0c;这样可以合理控制最大线程数和阻塞队列以及拒绝策略等

ThreadPoolExecutor中有几个重要的属性

自定义线程池

private static void customThreadPool() {ExecutorService threadPool &＃61; new ThreadPoolExecutor(2, 5, 1L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(3),Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy());try {for (int i &＃61; 0; i < 90; i&＃43;&＃43;) {threadPool.execute(() -> {System.out.println(Thread.currentThread().getName() &＃43; "\t 开始处理业务");});}} catch (Exception e) {e.printStackTrace();} finally {threadPool.shutdown();}}


参数说明

核心属性

int corePoolSize&＃xff1a;核心线程数
int maximumPoolSize&＃xff1a;允许创建的最大线程数
long keepAliveTime&＃xff1a;空闲时间
TimeUnit unit&＃xff1a;时间单位
BlockingQueue<Runnable> workQueue&＃xff1a;任务队列
ThreadFactory threadFactory&＃xff1a;线程池工厂
RejectedExecutionHandler handler&＃xff1a;拒绝策略对于一个线程池来说&＃xff0c;允许有 maximumPoolSize &＃43; workQueue.length 个任务提交过来&＃xff0c;超过了这个阈值&＃xff0c;就会触发拒绝策略private final AtomicInteger ctl &＃61; new AtomicInteger(ctlOf(RUNNING, 0));
这个遍历的高三位保存的是线程池的状态&＃xff0c;后29位保存的是线程池中工作线程的数量


线程池状态

/*** 运行状态&＃xff0c;线程池刚创建就是这个状态* 高三位是111&＃xff0c;低29位是0* 该状态会接收新的任务请求* 也会处理在阻塞队列中等待处理的任务*/private static final int RUNNING &＃61; -1 << COUNT_BITS;/*** 停工状态&＃xff0c;不再接收新任务&＃xff0c;但是会处理已经在执行的和阻塞队列中的任务* 高三位为000&＃xff0c;低29位为0&＃xff0c;* 在running状态调用线程池的shutdown()方法&＃xff0c;会从running变更为shutdown*/private static final int SHUTDOWN &＃61; 0 << COUNT_BITS;/*** 停止状态&＃xff0c;不再接收新任务&＃xff0c;已有的任务也会中断,阻塞队列中的任务也不会再运行* 高三位为001&＃xff0c;低29位为0* 在调用shutdownnow()方法之后&＃xff0c;会从running/shutdown变更为stop状态*/private static final int STOP &＃61; 1 << COUNT_BITS;/*** 清空状态&＃xff0c;所有任务都停止了&＃xff0c;工作的线程也全部结束了&＃xff0c;workerCount为0* 高三位为010&＃xff0c;低29位为0* 为此状态时&＃xff0c;还将调用terminated()方法*/private static final int TIDYING &＃61; 2 << COUNT_BITS;/*** 终止状态&＃xff0c;线程池已经销毁&＃xff1b;为此状态时还将调用terminated()方法* 高三位为100&＃xff0c;低29位为0*/private static final int TERMINATED &＃61; 3 << COUNT_BITS;


拒绝策略&＃xff1a;

1、AbortPolicy&＃xff1a;直接抛出异常 RejectedExecutionException
2、CallerRunsPolicy&＃xff1a;在调用方线程中执行任务
3、DiscardOldestPolicy&＃xff1a;丢弃最先进入到队列中的任务&＃xff0c;调用的是queue.offer()方法
4、DiscardPolicy&＃xff1a;不做任务处理&＃xff0c;其实丢弃的是当前任务


运行原理

1. 创建线程池&＃xff0c;等待请求过来2. 当调用了execute()方法添加一个请求任务之后&＃xff0c;线程池会做如下判断:3. 1. 如果当前运行的线程数小于corePoolSize&＃xff0c;那么会创建线程&＃xff0c;运行这个任务2. 如果正在运行的线程数量大于或者等于corePoolSize,那么将这个任务加入队列3. 如果队列也满了&＃xff0c;这时候运行的线程数还小于maximumPoolSize&＃xff0c;就会创建非核心线程来运行这个任务4. 如果队列满了&＃xff0c;且正在运行的线程数达到了maximumPoolSize,那么线程池会启动拒绝策略来执行4. 当一个线程的任务结束之后&＃xff0c;会从队列中取下一个任务来执行5. 当一个线程空闲时间超过了keepAliveTime时&＃xff0c;线程池会做以下判断:6. 如果当前运行的线程数超过了corePoolSize时&＃xff0c;那么这个线程会被销毁&＃xff0c;线程池中所有任务结束以后&＃xff0c;线程数量会缩减到corePoolSize的大小

execute和submit的区别

1.入参不同&＃xff0c;execute的入参只能是Runnable,submit的入参可以是Runnable&＃xff0c;也可以是callable
2.submit有返回值&＃xff0c;submit底层也是调用的execute&＃xff0c;在调用execute的时候&＃xff0c;会把入参包装成RunnableFuture
3.submit方便异常处理&＃xff0c;如果task抛出了异常&＃xff0c;那么通过future.get()可以捕获到异常信息

核心方法

1.Executor是线程池的基本接口&＃xff0c;只有一个execute方法
2.ExecutorService是线程池Executor接口的子接口&＃xff0c;额外提供了一些对线程池的其他操作&＃xff1a;submit、shutdown等
3.AbstractExecutorService抽象类&＃xff0c;实现了executorService的部分接口&＃xff0c;比如&＃xff1a;submit
4.ThreadPoolExecutor继承了AbstractExecutorService

获取线程池状态、工作线程数量

/*** 获取当前线程池的状态* &＃64;param c* &＃64;return*/private static int runStateOf(int c) { return c & ~CAPACITY; }/*** 获取线程池活跃的数量* &＃64;param c* &＃64;return*/private static int workerCountOf(int c) { return c & CAPACITY; }/*** 获取运行状态和活动线程数量的值* &＃64;param rs* &＃64;param wc* &＃64;return*/private static int ctlOf(int rs, int wc) { return rs | wc; }


runWorker()

执行提交的任务&＃xff0c;这是比较核心的方法&＃xff0c;在该方法中
1、先执行当前提交的任务
2、从任务队列中获取排队的任务
3、并执行

getTask()

这是从任务队列中获取排队任务的方法&＃xff0c;在新增一个线程的时候&＃xff0c;会先执行当前提交的任务&＃xff0c;如果该任务执行完了之后&＃xff0c;会从任务队列中尝试获取正在排队的任务&＃xff0c;去依次执行