线程池源码探究
纳兰小依 人气:11.线程池简介
使用线程池，一般会使用JDK提供的几种封装类型，即：newFixedThreadPool、newSingleThreadExecutor、newCachedThreadPool等，这些线程池的定义在Executors类中，来看看相关的源码：
public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>(),
threadFactory);
}
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
这些方法内部都使用了ThreadPoolExecutor的构造方法，区别只是传入的参数不同。ThreadPoolExecutor有四个重载的构造方法，最终调用的是由7个参数的构造器，其源码如下：
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;
}
参数解释：
corePoolSize:核心池大小，默认情况下，线程池启动之后，并不会立即创建线程，而是要等到任务到来之后，才创建线程去执行任务（除非设置了allowCoreThreadTimeOut参数，该参数会在线程池启动之后立马创建核心池数量的线程）。随着任务的不断增加，现有线程无法满足要求，就会不断的创建新线程，直到线程数达到corePoolSize的值，后续新来的任务会放入阻塞队列；maximumPoolSize: 最大池大小，当任务太多，阻塞队列满了之后，如果线程数量还没有超过该参数的值，就会继续创建新线程，直到线程数达到该参数规定的值，后续再来的任务会使用拒绝策略进行处理；keepAliveTime: 如果线程数超过corePoolSize的值，那么多余的线程在空闲keepAliveTime时间后会被销毁；unit:keepAliveTime参数的单位；workQueue: 阻塞队列；threadFactory: 线程工厂，创建线程时需要使用到该工厂；handler: 拒绝策略。
2.核心字段
ThreadPoolExecutor的核心字段如下：
//ctl低29位表示线程的数量，高3位表示线程池状态，因此当前线程池允许的最大线程数量是2^29-1
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
//固定值29
private static final int COUNT_BITS = Integer.SIZE - 3;
//线程最大容量
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
// 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;
3.线程池状态
线程池的状态有5种，状态之间的转换关系如下图：
初始情况下，线程池创建完毕后会处于RUNNING状态，可以正常的接受新任务；当调用shutdown()时，线程池变成SHUTDOWN状态，此时无法接受新任务，但是会继续执行阻塞队列中的任务；当调用shutdownNow()时，线程由RUNNING状态变成STOP状态，此时不能接受新任务，并且会中断正在执行的任务；当线程池中的线程数减少为0时，就会转成TIDYING状态；在TIDYING状态会自动调用terminated()使线程池转为TERMINATED状态。
shutdown()
shutdown()方法的逻辑分别由5个不同的方法来实现，这里将这些方法整理在一起，如下：
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//检查security manager是否允许调用方执行此方法
checkShutdownAccess();
//将线程池状态更新为SHUTDOWN
advanceRunState(SHUTDOWN);
//中断空闲线程
interruptIdleWorkers();
//这是一个空实现，允许子类进行重写
onShutdown(); // hook for ScheduledThreadPoolExecutor
} finally {
mainLock.unlock();
}
tryTerminate();
}
private void advanceRunState(int targetState) {
for (;;) {
int c = ctl.get();
//如果线程池已经处在targetState及之后的状态则直接结束循环，否则使用CAS操作将线程池状态更新为targetState
if (runStateAtLeast(c, targetState) ||
ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))
break;
}
}
private void interruptIdleWorkers() {
interruptIdleWorkers(false);
}
//onlyOne表示是否只终止一个空闲线程
private void interruptIdleWorkers(boolean onlyOne) {
final ReentrantLock mainLock = this.mainLock;
//加可重入锁
mainLock.lock();
try {
for (Worker w : workers) {
Thread t = w.thread;
//如果线程没有被中断，则尝试获取锁，获取成功后将线程中断
if (!t.isInterrupted() && w.tryLock()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
} finally {
//释放锁
w.unlock();
}
}
if (onlyOne)
break;
}
} finally {
mainLock.unlock();
}
}
final void tryTerminate() {
//自旋
for (;;) {
int c = ctl.get();
//线程池还在运行，或者已经是TIDYING或TERMINATED状态，或者阻塞队列不为空，这几种情况不再继续执行
if (isRunning(c) ||
runStateAtLeast(c, TIDYING) ||
(runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
return;
//线程数不为0时，终止一个空闲线程
if (workerCountOf(c) != 0) { // Eligible to terminate
interruptIdleWorkers(ONLY_ONE);
return;
}
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//将线程池设置为DIDYING状态
if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
//设置成功后，执行terminated()方法
try {
//这也是一个空实现，子类可以根据需要进行重写
terminated();
} finally {
//将线程池设置为TERMINATED状态
ctl.set(ctlOf(TERMINATED, 0));
termination.signalAll();
}
return;
}
} finally {
mainLock.unlock();
}
// else retry on failed CAS
}
}
shutdownNow()
public List<Runnable> shutdownNow() {
List<Runnable> tasks;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//检查security manager是否允许调用方执行此方法
checkShutdownAccess();
//将线程池状态更新为STOP
advanceRunState(STOP);
//与shutdown的区别是，这里会中断所有线程，而不仅仅是空闲线程
interruptWorkers();
//将任务从workQueue中移除，转移到一个ArrayList中，此操作后，workQueue为空，已有的任务无法继承执行
tasks = drainQueue();
} finally {
mainLock.unlock();
}
tryTerminate();
return tasks;
}
//中断所有线程
private void interruptWorkers() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (Worker w : workers)
w.interruptIfStarted();
} finally {
mainLock.unlock();
}
}
4.执行任务
线程池通过execute()方法执行任务，其源码如下：
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
int c = ctl.get();
//如果当前活跃线程小于核心池大小，就尝试创建新的线程
if (workerCountOf(c) < corePoolSize) {
//如果成功创建新线程并且启动成功，直接返回
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，则创建一个非核心线程，并将firstTask设置为null
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
//如果添加非核心线程失败，则执行拒绝策略
else if (!addWorker(command, false))
reject(command);
}
//获取活跃的线程数
private static int workerCountOf(int c) { return c & CAPACITY; }
//获取线程池运行状态
private static int runStateOf(int c) { return c & ~CAPACITY; }
来看看addWorker()方法的实现：
//core表示要创建的是否是核心线程，true表示创建核心线程，false表示创建非核心线程
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
//获取线程池状态
int rs = runStateOf(c);
// Check if queue empty only if necessary.
//rs >= SHUTDOWN，表示线程池已终止
//rs>=SHUTDOWN，说明已经调用了shutdown()或者shutdownNow()方法，在此条件满足的情况下，第二项条件等同于
//rs!=SHUTDOWN || firstTask != null || workQueue.isEmpty()，满足这三个条件的任何一个都不会再添加新任务
//rs!=SHUTDOWN，说明是STOP、TIDYING、TERMINATE这三种
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
//执行到这里说明：
//① rs<SHUTDOWN，即线程池是运行状态
//② rs=SHUTDOWN，farstTask=null, 并且阻塞队列不为空
for (;;) {
int wc = workerCountOf(c);
//有三种情况会返回false：1)线程数达到最大值；2)当前创建核心线程，但是线程数已经达到核心池大小；
//3)当前创建非核心线程，并且线程数达到最大池大小
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
//如果使用CAS操作成功将ctl的值加1，则跳出最外层循环
if (compareAndIncrementWorkerCount(c))
break retry;
//走到这里说明无法使用CAS更新ctl的值，说明此时发生了多线程竞争，需要重新查看线程池的状态
c = ctl.get(); // Re-read ctl
if (runStateOf(c) != rs)
continue retry;
// 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 {
// 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 startable
throw new IllegalThreadStateException();
//workers是个HashSet类型，只在重入锁代码中被访问
workers.add(w);
//更新当前活跃线程的最大值
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
//启动线程，调用Worker类的run()方法
t.start();
workerStarted = true;
}
}
} finally {
//成功创建新线程时，才会设置workerStarted=true，这里处理没有创建新线程的情况
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
addWorker()方法中用到了Worker类，这是ThreadPoolExecutor的内部类，对线程进行了包装，线程池创建或者启动的线程，实际都是Worker类型的实例，其源码如下（省略了无关代码）：
private final class Worker extends AbstractQueuedSynchronizer implements Runnable
{
/** Thread this worker is running in. Null if factory fails. */
final Thread thread;
/** Initial task to run. Possibly null. */
Runnable firstTask;
/** Per-thread task counter */
volatile long completedTasks;
Worker(Runnable firstTask) {
setState(-1);
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
/** Delegates main run loop to outer runWorker */
public void run() {
runWorker(this);
}
当启动Worker线程时，会调用runWorker()方法，每一个启动的线程都会在该方法的while循环中不断获取任务去执行，该方法源码如下：
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
//如果能够成功拿到任务，则执行下面的代码块，如果getTask()方法返回null，当前线程就会执行退出逻辑
while (task != null || (task = getTask()) != null) {
//如果能将state字段设置为1，表示成功拿到锁，就接着向下执行，否则线程会加入等待队列，不再继续执行
//注意这里是在成功拿到新任务之后才会加锁，结合shutdown()方法的逻辑
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
//如果线程池正在关闭，需要中断当前线程
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
//前置钩子
beforeExecute(wt, task);
Throwable thrown = null;
try {
//执行任务
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 {
//后置钩子
afterExecute(task, thrown);
}
} finally {
task = null;
w.completedTasks++;
//释放锁
w.unlock();
}
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
}
}
beforeExecute()和afterExecute()是protected类型，并且默认是空实现，很明显是留给子类去实现钩子逻辑。上面的代码使用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.
//线程池正在关闭，或者阻塞队列空了，就减少线程数，并返回null
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
// Are workers subject to culling?
//在设置了allowCoreThreadTimeOut参数后，超过给定的时间，会将空闲的核心线程清理掉
//或者线程数量超过了核心池数量，会在一定时间后清理掉多余的线程
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
//1)线程数量超过最大池数量，或者超时; 2)线程数大于1，或者阻塞队列为空; 这两个条件都成立时，就将ctl值减1
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
//如果设置了超时状态，则使用poll方法取任务，超过keepAliveTime还没有任务到来就返回true
//否则使用take取任务，在阻塞队列为空时会一直等待
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
//线程有可能在等待新任务的到来而阻塞，但是在等待的过程中调用shutdownNow()关闭线程时，线程会抛出中断异常，在这里被捕获
timedOut = false;
}
}
}
现在来整理一下runWorker()方法的思路：每一个新创建的线程都会在runWorker()方法里通过while循环不断地从阻塞队列中获取任务，取到任务之后就执行任务的run()方法，取不到任务就会一直阻塞，或者等待一定的时间之后，空闲线程超时需要回收，就会执行processWorkerExit()方法。
5.线程池是如何关闭的
shutdown()
在介绍shutdown()方法时有一个疑问，该方法只会中断空闲线程，但是非空闲的线程不会被中断，即使该线程被阻塞，因此该方法有可能无法关闭那些一直处在等待状态的非空闲线程，这一点在使用时需要注意。在runWorker()方法中，while循环会在成功拿到任务后才会加锁，因此那些由于阻塞队列为空拿不到任务而阻塞的线程也会被shutdown()方法中断
while (task != null || (task = getTask()) != null) {
//如果能将state字段设置为1，表示成功拿到锁，就接着向下执行，否则线程会加入等待队列，不再继续执行
//注意这里是在成功拿到新任务之后才会加锁，结合shutdown()方法的逻辑
w.lock();
//忽略其他代码
}
shutdownNow()
shutdownNow()会中断所有的存活线程，不论这些线程是否空闲，因此可能会导致任务在执行的过程中抛出异常，这点需要注意。
不论是调用哪个方法来关闭线程池，最终线程的退出是要根据getTask()方法来决定。当getTask()方法返回null，即当前阻塞队列已经没有任务时，线程会退出，并且在getTask()方法的自旋代码会首先检查线程池的状态，如下：
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
在调用shutdownNow()方法关闭线程池后，rs >= STOP逻辑成立，直接返回null，而shutdown()方法会继续执行阻塞队列中的任务，直到workQueue.isEmpty()条件为真，getTask()返回null导致线程一个个结束，不论是哪种情况，最终线程池中的线程数量都会变成0。
加载全部内容