java 线程池框架核心代码分析-亚博电竞手机版

前言

多线程编程中，为每个任务分配一个线程是不现实的，线程创建的开销和资源消耗都是很高的。线程池应运而生，成为我们管理线程的利器。java 通过executor接口，提供了一种标准的方法将任务的提交过程和执行过程解耦开来，并用runnable表示任务。

下面，我们来分析一下 java 线程池框架的实现threadpoolexecutor。

下面的分析基于jdk1.7

生命周期

threadpoolexecutor中，使用capacity的高3位来表示运行状态，分别是：

1. running：接收新任务，并且处理任务队列中的任务
2. shutdown：不接收新任务，但是处理任务队列的任务
3. stop：不接收新任务，不出来任务队列，同时中断所有进行中的任务
4. tidying：所有任务已经被终止，工作线程数量为 0，到达该状态会执行terminated()
5. terminated：terminated()执行完毕

状态转换图

threadpoolexecutor中用原子类来表示状态位

private final atomicinteger ctl = new atomicinteger(ctlof(running, 0));

线程池模型

核心参数

• corepoolsize：最小存活的工作线程数量（如果设置allowcorethreadtimeout，那么该值为 0）
• maximumpoolsize：最大的线程数量，受限于capacity
• keepalivetime：对应线程的存活时间，时间单位由timeunit指定
• workqueue：工作队列，存储待执行的任务
• rejectexecutionhandler：拒绝策略，线程池满后会触发

线程池的最大容量：capacity中的前三位用作标志位，也就是说工作线程的最大容量为(2^29)-1

四种模型

• cachedthreadpool：一个可缓存的线程池，如果线程池的当前规模超过了处理需求时，那么将回收空闲的线程，当需求增加时，则可以添加新的线程，线程池的规模不存在任何的限制。
• fixedthreadpool：一个固定大小的线程池，提交一个任务时就创建一个线程，直到达到线程池的最大数量，这时线程池的大小将不再变化。
• singlethreadpool：一个单线程的线程池，它只有一个工作线程来执行任务，可以确保按照任务在队列中的顺序来串行执行，如果这个线程异常结束将创建一个新的线程来执行任务。
• scheduledthreadpool：一个固定大小的线程池，并且以延迟或者定时的方式来执行任务，类似于timer。

执行任务 execute

核心逻辑：

1. 当前线程数量 < corepoolsize，直接开启新的核心线程执行任务addworker(command, true)
2. 当前线程数量 >= corepoolsize，且任务加入工作队列成功
   1. 检查线程池当前状态是否处于running
   2. 如果否，则拒绝该任务
   3. 如果是，判断当前线程数量是否为 0，如果为 0，就增加一个工作线程。
3. 开启普通线程执行任务addworker(command, false)，开启失败就拒绝该任务

从上面的分析可以总结出线程池运行的四个阶段：

1. poolsize < corepoolsize 且队列为空，此时会新建线程来处理提交的任务
2. poolsize == corepoolsize，此时提交的任务进入工作队列，工作线程从队列中获取任务执行，此时队列不为空且未满。
3. poolsize == corepoolsize，并且队列已满，此时也会新建线程来处理提交的任务，但是poolsize < maxpoolsize
4. poolsize == maxpoolsize，并且队列已满，此时会触发拒绝策略

拒绝策略

前面我们提到任务无法执行会被拒绝，rejectedexecutionhandler是处理被拒绝任务的接口。下面是四种拒绝策略。

• abortpolicy：默认策略，终止任务，抛出rejectedexception
• callerrunspolicy：在调用者线程执行当前任务，不抛异常
• discardpolicy： 抛弃策略，直接丢弃任务，不抛异常
• discardolderspolicy：抛弃最老的任务，执行当前任务，不抛异常

线程池中的 worker

worker继承了abstractqueuedsynchronizer和runnable，前者给worker提供锁的功能，后者执行工作线程的主要方法runworker(worker w)（从任务队列捞任务执行）。worker 引用存在workers集合里面，用mainlock守护。

private final reentrantlock mainlock = new reentrantlock(); private final hashset workers = new hashset();

核心函数 runworker

下面是简化的逻辑，注意：每个工作线程的run都执行下面的函数

final void runworker(worker w) {     thread wt = thread.currentthread();     runnable task = w.firsttask;     w.firsttask = null;     while (task != null || (task = gettask()) != null) {         w.lock();         beforeexecute(wt, task);         task.run();         afterexecute(task, thrown);         w.unlock();     }     processworkerexit(w, completedabruptly); }

1. 从gettask()中获取任务
2. 锁住 worker
3. 执行beforeexecute(wt, task)，这是threadpoolexecutor提供给子类的扩展方法
4. 运行任务，如果该worker有配置了首次任务，则先执行首次任务且只执行一次。
5. 执行afterexecute(task, thrown);
6. 解锁 worker
7. 如果获取到的任务为 null，关闭 worker

获取任务 gettask

线程池内部的任务队列是一个阻塞队列，具体实现在构造时传入。

private final blockingqueue workqueue;

gettask()从任务队列中获取任务，支持阻塞和超时等待任务，四种情况会导致返回null，让worker关闭。

1. 现有的线程数量超过最大线程数量
2. 线程池处于stop状态
3. 线程池处于shutdown状态且工作队列为空
4. 线程等待任务超时，且线程数量超过保留线程数量

核心逻辑：根据timed在阻塞队列上超时等待或者阻塞等待任务，等待任务超时会导致工作线程被关闭。

timed = allowcorethreadtimeout || wc > corepoolsize; runnable r = timed ?     workqueue.poll(keepalivetime, timeunit.nanoseconds) :     workqueue.take();

在以下两种情况下等待任务会超时：

1. 允许核心线程等待超时，即allowcorethreadtimeout(true)
2. 当前线程是普通线程，此时wc > corepoolsize

工作队列使用的是blockingqueue，这里就不展开了，后面再写一篇详细的分析。

总结

• threadpoolexecutor基于生产者-消费者模式，提交任务的操作相当于生产者，执行任务的线程相当于消费者。
• executors提供了四种基于threadpoolexecutor构造线程池模型的方法，除此之外，我们还可以直接继承threadpoolexecutor，重写beforeexecute和afterexecute方法来定制线程池任务执行过程。
• 使用有界队列还是无界队列需要根据具体情况考虑，工作队列的大小和线程的数量也是需要好好考虑的。
• 拒绝策略推荐使用callerrunspolicy，该策略不会抛弃任务，也不会抛出异常，而是将任务回退到调用者线程中执行。