Java线程池原理解析
站长
· 阅读数 39
参考文章：juejin.cn/post/698321…
一、Java线程池的基本使用
在Java中可以通过如下两种方式使用线程池，不过最终都是使用到ThreadPoolExecutor。
1、直接使用ThreadPoolExecutor
可通过ThreadPoolExecutor的构造函数创建出线程池的实例对象，并调用其execute函数进行任务添加。代码实现如下：
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
10,//corePoolSize
20,//maximumPoolSize
120,//keepAliveTime
TimeUnit.SECONDS,//keepAliveTime时间单位
new ArrayBlockingQueue<>(10)//等待队列
);
//将任务添加到该线程池中
threadPoolExecutor.execute(() -> {
//do something
});//注意这里使用时lambda表达式
这里简单说一下lambda表达式，注意遵循如下几个原则即可，具体可参考该篇博客。
（1）需要函数式接口（即一个接口中只有一个函数）；注意一定要是接口；
（2）lambada表达式->前面的是形参，形参可省略掉参数类型；如果只有一个形参则可以省略掉括号；
（3）->后面的是函数体，如果只有一行代码则可省略掉大括号，如果该一行代码是返回语句则省略掉大括号的同时也需要省略掉return关键词。
2、使用Executors
通过该类提供的函数，可以创建如下几种类型的线程池；不过其底层仍然使用的是ThreadPoolExecutor。
//该线程池只包含一个线程，所有任务顺序执行；适用于需要顺序执行任务的场景
ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
//创建固定线程数量线程池，即该线程池中线程数量固定，如果线程池中线程都处于忙碌状态，则新进来的任务需要进入到阻塞队列中，直到线程空闲为止。
//适用于需要限制线程数量的场景
ExecutorService fixedThreadPool = Executors.newFixedThreadPool(3);
//可缓存线程池是一个根据需要自动调整大小的线程池;适用于执行大量的短期异步任务的场景
ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
//定时任务线程池，底层使用的是ScheduledThreadPoolExecutor，不过该类也继承至ThreadPoolExecutor；
//用于执行定时任务和周期性任务
ExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(3);
如下以函数newCachedThreadPool为例子，学习一下Executors创建线程池的底层实现。从其源码可以知道其也是通过ThreadPoolExecutor类创建的线程池对象，只是其中的各个参数已经定义好了而已。
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
二、ThreadPoolExecutor源码
1、ThreadPoolExecutor构造函数
/**
* corePoolSize：线程池中核心线程数量，一般来说核心线程会一直存在，即使处于空闲状态；
* maximumPoolSize：线程池中最大线程数量；当任务量超过核心线程数量并且任务队列已满，线程池还能创建新的线程来处理任务，但是线程数（包括核心线程数量）不会超过该值；
* keepAliveTime：非核心线程在空闲状态下能够存活的时间；
* unit：存活时间单位（ms、s、h等）；
* workQueue：等待执行的任务列表（先进先出）；
* threadFactory：线程创建工厂，可使用默认的，也可以自定义；
* handler：任务超过 maximumPoolSize+任务队列大小回调函数，即无法处理新的任务；也称拒绝策略。
*/
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.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}
2、execute函数
通过调用该函数以提交新的任务到线程池中进行处理。因此以该函数的源码实现作为学习线程池源码的入口点。代码如下：
public void execute(Runnable command) {
//不允许提交的任务为空
if (command == null)
throw new NullPointerException();
//获取当前线程池状态+线程数量（具体可阅读开头的参考文章）
int c = ctl.get();
//判断当前线程数是否小于核心线程数
if (workerCountOf(c) < corePoolSize) {
//如果小于核心线程数则创建新的线程并直接执行
//这里参数true表示当前需要创建的线程是核心线程（用于判断当前允许的最大线程数量）
if (addWorker(command, true))
return;
//任务执行失败，则重新获取线程池状态+当前线程数
c = ctl.get();
}
//首先判断线程池是否处于运行状态，如果是则将当前任务添加到任务队列中
if (isRunning(c) && workQueue.offer(command)) {
//重新获取线程池状态标识
int recheck = ctl.get();
//如果线程池没有运行，则从队列中移除该任务，并执行拒绝策略
if (! isRunning(recheck) && remove(command))
reject(command);
//当前正在运行线程为0
//则需要启动一个核心线程以运行任务队列中的任务
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
//任务队列已满，则启动非核心线程，false标识启动非核心线程
//如果任务运行失败，则执行拒绝策略
else if (!addWorker(command, false))
reject(command);
}
如上代码可缩减为如下几个步骤：
- 判断是否能够创建核心线程以执行任务
- 将任务添加到任务队列
- 创建非核心线程执行任务
- 任务执行失败则执行拒绝策略
3、addWorker函数
该函数主要有两个功能：（1）通过CAS将当前线程数量+1；（2）创建新的线程并添加到工作队列中，同时立马开始执行当前传递进来的任务；
private boolean addWorker(Runnable firstTask, boolean core) {
retry://类似于goto
//获取当前线程池状态+线程数量
for (int c = ctl.get();;) {
//如果当前线程池状态是RUNNING则可以继续执行任务
//否则如果当前线池状态是STOP、TIDYING或者TERMINATED或者待执行任务为空或者工作队列为空，则直接返回false
if (runStateAtLeast(c, SHUTDOWN)
&& (runStateAtLeast(c, STOP)
|| firstTask != null
|| workQueue.isEmpty()))
return false;
//通过CAS将线程数量+1，直到修改成功为止
for (;;) {
//根据当前是否是核心线程选择当前允许的最大线程数量
//如果超过允许的最大线程数量则直接返回false
if (workerCountOf(c)
>= ((core ? corePoolSize : maximumPoolSize) & COUNT_MASK))
return false;
//通过CAS将线程数量+1
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get();
//CAS修改线程数量失败，再次判断线程池状态
if (runStateAtLeast(c, SHUTDOWN))
continue retry;
}
}
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
//该类为ThreadPoolExecutor内部类，继承了AbstractQueuedSynchronizer类并且实现了Runnable类
//并且在其构造函数中会调用所提供的工厂创建新的线程
w = new Worker(firstTask);
//获取创建的新线程
final Thread t = w.thread;
if (t != null) {
//加锁
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
int c = ctl.get();
//判断线程池是否处于运行状态或者可执行任务状态
if (isRunning(c) ||
(runStateLessThan(c, STOP) && firstTask == null)) {
//如果线程状态不是处于新创建状态则直接抛出异常
if (t.getState() != Thread.State.NEW)
throw new IllegalThreadStateException();
//将新创建的worker添加到工作队列中
workers.add(w);
workerAdded = true;
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
}
//释放锁
} finally {
mainLock.unlock();
}
//如果是新创建的线程则直接开始执行任务
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
//没有启动新的任务或者启动失败，则将新的任务从任务队列移除，并将线程数量-1
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
3、Worker类源码
该类比较简单，其构造函数中会存储当前需要运行的任务（可为空）；并通过提供的工厂创建新的线程，同时将当前对象引用作为Runnable任务传递给该线程；当上述addWorker函数中调用t.start()就会调用到该类的run函数中以执行后续的逻辑。
private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
private static final long serialVersionUID = 6138294804551838833L;
@SuppressWarnings("serial") // Unlikely to be serializable
final Thread thread;
@SuppressWarnings("serial") // Not statically typed as Serializable
Runnable firstTask;
volatile long completedTasks;
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
//当前需要执行的任务
this.firstTask = firstTask;
//通过提供的工厂创建新的线程，注意这里将当前类的引用传递给了Thread作为Runnable任务
this.thread = getThreadFactory().newThread(this);
}
public void run() {
runWorker(this);
}
//省略掉部分源码
......
}
4、runWorker函数
如下代码就是当前线程开始执行任务了；（1）首先判断worker是否存在需要执行的任务，如果有则优先执行该任务；（2）否则从队列中获取任务进行执行；
final void runWorker(Worker w) {
//获取当前线程
Thread wt = Thread.currentThread();
//获取当前worker中需要执行的任务（优先级最高）
Runnable task = w.firstTask;
//将worker中需要执行任务设置为空，防止后续重复执行
w.firstTask = null;
w.unlock();
boolean completedAbruptly = true;
try {
//首先判断当前worker中需要执行的任务是否为空，如果不为空则执行该任务
//否则从任务列表中获取任务进行执行，注意getTask会阻塞当前线程
while (task != null || (task = getTask()) != null) {
//加锁
w.lock();
//判断当前线程池以及线程是否能够继续执行任务
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
//该函数未实现
beforeExecute(wt, task);
try {
//执行任务
task.run();
//该函数未实现
afterExecute(task, null);
} catch (Throwable ex) {
afterExecute(task, ex);
throw ex;
}
} finally {
task = null;
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
//结束当前线程
processWorkerExit(w, completedAbruptly);
}
}
5、getTask函数
在该函数中会根据是否允许超时回收线程以设置任务获取阻塞时间，如果是核心线程则会一直阻塞，非核心线程如果超时仍然没有获取到可执行任务则会被回收。
private Runnable getTask() {
boolean timedOut = false;
for (;;) {
int c = ctl.get();
//判断线程池是否处于可执行任务状态，如果不是则返回null
//注意这里SHUTDOWN状态是可以继续执行任务的
if (runStateAtLeast(c, SHUTDOWN)
&& (runStateAtLeast(c, STOP) || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
//获取当前线程数量
int wc = workerCountOf(c);
//allowCoreThreadTimeOut表示是否允许空闲的核心线程超时回收
//或者如果当前线程数量大于核心线程数量，则超过核心线程数量的线程可以超时回收
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
//如果某个线程超时获取task为null，则会执行到该处
//注意这里修改线程数量是通过CAS，也就是同一时刻只有一个线程能够修改成功，其余线程如果同步在修改则会失败
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
//通过CAS将线程数量-1，并直接返回null，即直接结束当前线程
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
//通过阻塞的方式等待队列任务
//如果timed为true则说明可以超时回收线程，即如果在特定时间内没有新的任务，那么当前线程就需要回收
//take函数会一直等待直到有新的任务出现需要处理
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
三、总结
如上代码分析下来，回答如下几个问题则没有什么难度了：
问题1：线程池任务处理基本流程
回答：通过execute函数将任务提交到线程池，如果核心线程数未超过则直接创建核心线程执行任务，否则添加到任务队列；
如果任务队列满了则创建非核心线程执行任务，否则执行拒绝策略。
问题2：如何复用现有线程
回答：从上述代码分析能够知道，每个线程会优先执行当前worker中的task，如果当前worker的task为空，则会去任务队列中获取任务进行执行；
如果任务队列为空则会阻塞当前线程；即典型的生产者-消费者模型，上层业务会向任务队列中添加任务（生产），线程会从任务队列中获取任务进行执行（消费），这样线程就能够不断复用了。
问题3：核心线程与非核心线程是否存在区别
回答：没有啥区别，如果当前既有核心线程也有非核心线程，那么最终被保留下来的线程就是核心线程，具体的可参考上述getTask函数源码。
转载自:https://juejin.cn/post/7255309681547706425