【Java】Java线程池ThreadPoolExecutor源码分析

• Z时代
• 2024-01-10
• 分类：技术分享

public interface Executor {
void execute(Runnable command);
}

public interface ExecutorService extends Executor {
void shutdown();
List<Runnable> shutdownNow();
boolean isShutdown();
boolean isTerminated();
boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException;
<T> Future<T> submit(Callable<T> task);
<T> Future<T> submit(Runnable task, T result);
Future<?> submit(Runnable task);
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException;
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
throws InterruptedException;
<T> T invokeAny(Collection<? extends Callable<T>> tasks)
throws InterruptedException, ExecutionException;
<T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException;
}

public abstract class AbstractExecutorService implements ExecutorService {
protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
return new FutureTask<T>(runnable, value);
}
protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
return new FutureTask<T>(callable);
}
public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
RunnableFuture<Void> ftask = newTaskFor(task, null);
execute(ftask);
return ftask;
}
public <T> Future<T> submit(Runnable task, T result) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task, result);
execute(ftask);
return ftask;
}
public <T> Future<T> submit(Callable<T> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task);
execute(ftask);
return ftask;
}
private <T> T doInvokeAny(Collection<? extends Callable<T>> tasks, boolean timed, long nanos)
throws InterruptedException, ExecutionException, TimeoutException {...}
public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
throws InterruptedException, ExecutionException {... }
public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {...}
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
throws InterruptedException {...}
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException {...}
}

public ThreadPoolExecutor(int corePoolSize,                             //核心线程数
int maximumPoolSize,                      //最大线程数
long keepAliveTime,                       //线程存活时间
TimeUnit unit,                            //keepAliveTime的单位
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.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;
}

//记录线程池状态和线程数量（总共32位，前三位表示线程池状态，后29位表示线程数量）
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
//线程数量统计位数29  Integer.SIZE=32
private static final int COUNT_BITS = Integer.SIZE - 3;
//容量 000 11111111111111111111111111111
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;
//运行中 111 00000000000000000000000000000
private static final int RUNNING    = -1 << COUNT_BITS;
//关闭 000 00000000000000000000000000000
private static final int SHUTDOWN   =  0 << COUNT_BITS;
//停止 001 00000000000000000000000000000
private static final int STOP       =  1 << COUNT_BITS;
//整理 010 00000000000000000000000000000
private static final int TIDYING    =  2 << COUNT_BITS;
//终止 011 00000000000000000000000000000
private static final int TERMINATED =  3 << COUNT_BITS;
//获取运行状态（获取前3位）
private static int runStateOf(int c)     { return c & ~CAPACITY; }
//获取线程个数（获取后29位）
private static int workerCountOf(int c)  { return c & CAPACITY; }
private static int ctlOf(int rs, int wc) { return rs | wc; }

RUNNING：接受新任务并且处理阻塞队列里的任务
SHUTDOWN：拒绝新任务但是处理阻塞队列里的任务
STOP：拒绝新任务并且抛弃阻塞队列里的任务同时会中断正在处理的任务
TIDYING：所有任务都执行完（包含阻塞队列里面任务），当前线程池活动线程为0，将要调用terminated方法
TERMINATED：终止状态。terminated方法调用完成以后的状态
线程池状态转换：
RUNNING -> SHUTDOWN：显式调用shutdown()方法, 或者隐式调用了finalize()方法
(RUNNING or SHUTDOWN) -> STOP：显式调用shutdownNow()方法
SHUTDOWN -> TIDYING：当线程池和任务队列都为空的时候
STOP -> TIDYING：当线程池为空的时候
TIDYING -> TERMINATED：当 terminated() hook 方法执行完成时候

public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
RunnableFuture<Void> ftask = newTaskFor(task, null);
execute(ftask);
return ftask;
}
public <T> Future<T> submit(Runnable task, T result) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task, result);
execute(ftask);
return ftask;
}
public <T> Future<T> submit(Callable<T> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task);
execute(ftask);
return ftask;
}

public void execute(Runnable command) {
//传进来的线程为null，则抛出空指针异常
if (command == null)
throw new NullPointerException();
//获取当前线程池的状态+线程个数变量
int c = ctl.get();
/**
* 3个步骤
*/
//1.判断当前线程池线程个数是否小于corePoolSize,小于则调用addWorker方法创建新线程运行,
//且传进来的Runnable当做第一个任务执行。
//如果调用addWorker方法返回false，则直接返回
if (workerCountOf(c) < corePoolSize) {
//添加一个core线程(核心线程)。此处参数的true，表示添加的线程是core容量下的线程
if (addWorker(command, true))
return;
//刷新数据，乐观锁就是没有锁
c = ctl.get();
}
/*  isRunning方法的定义：
private static boolean isRunning(int c)
{return c < SHUTDOWN;}
2.SHUTDOWN值为0，即如果c小于0，表示在运行；offer用来判断任务是否成功入队*/
if (isRunning(c) && workQueue.offer(command)) {
//二次检查
int recheck = ctl.get();
//如果当前线程池状态不是RUNNING则从队列删除任务，并执行拒绝策略
if (! isRunning(recheck) && remove(command))
//执行拒绝策略
reject(command);
//否则如果当前线程池线程空，则添加一个线程
else if (workerCountOf(recheck) == 0)
//添加一个空线程进线程池，使用非core容量线程
//仅有一种情况，会走这步，core线程数为0，max线程数>0,队列容量>0
//创建一个非core容量的线程，线程池会将队列的command执行
addWorker(null, false);
}
//线程池停止了或者队列已满，添加maximumPoolSize容量工作线程，如果失败，执行拒绝策略
else if (!addWorker(command, false))
reject(command);
}

    private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get(); //获取运行状态和工作数量
int rs = runStateOf(c); //获取当前线程池运行的状态
// Check if queue empty only if necessary.
//条件代表着以下几个场景，直接返回false说明当前工作线程创建失败
//1.rs>SHUTDOWN 此时不再接收新任务，且所有的任务已经执行完毕
//2.rs=SHUTDOWN 此时不再接收新任务，但是会执行队列中的任务
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
for (;;) {
int wc = workerCountOf(c);
//先判断当前活动的线程数是否大于最大值，如果超过了就直接返回false说明线程创建失败
//如果没有超过再根据core的值再进行以下判断
//1. core为true，则判断当前活动的线程数是否大于corePoolSize
//2. core为false，则判断当前活动线程数是否大于maximumPoolSize
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
//比较当前值是否和c相同，如果相同，则改为c+1，并且跳出大循环，直接执行Worker进行线程创建
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get();  // Re-read ctl
//检查下当前线程池的状态是否已经发生改变
//如果已经改变了，则进行外层retry大循环，否则只进行内层的循环
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 {
//Worker的也是Runnable的实现类
w = new Worker(firstTask);
//因为不可以直接在Worker的构造方法中进行线程创建
//所以要把它的引用赋给t方便后面进行线程创建
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.add(w);//将创建的线程添加到workers容器中
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}

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;
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
}

/** Delegates main run loop to outer runWorker  */
public void run() {
//这里执行的是ThreadPoolExecutor中的runWorker
runWorker(this);
}

final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;//获取Worker中的任务
w.firstTask = null; //将Woeker中的任务置空
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
//如果当前任务为空  那么就从getTask中获得任务
/**
* 如果task不为空，执行完task后则将task置空
* 继续进入循环，则从getTask中获取任务
*/
while (task != null || (task = getTask()) != null) {
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);
}
}

    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.
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
// Are workers subject to culling?
//如果设置了allowCoreThreadTimeOut(true)
//或者当前运行的任务数大于设置的核心线程数
// timed = true
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
/** ------------------------以上的操作跟之前类似----------------------- */
/** ------------------------关键在于下面的代码------------------------- */
/** ------------------------从阻塞队列中获取任务----------------------- */
try {
Runnable r = timed ?
//对于阻塞队列，poll(long timeout, TimeUnit unit) 将会在规定的时间内去任务
//如果没取到就返回null
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
//take会一直阻塞，等待任务的添加
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}

    /**
* @param completedAbruptly
*/
private void processWorkerExit(Worker w, boolean completedAbruptly) {
if (completedAbruptly) //如果突然被打断，工作线程数不会被减少
decrementWorkerCount();
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
completedTaskCount += w.completedTasks;
workers.remove(w);
} finally {
mainLock.unlock();
}
tryTerminate();
int c = ctl.get();
//判断运行状态是否在STOP之前
if (runStateLessThan(c, STOP)) {
if (!completedAbruptly) {//正常退出，也就是task == null
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
if (min == 0 && ! workQueue.isEmpty())
min = 1;
if (workerCountOf(c) >= min)
return; // replacement not needed
}
//新增一个工作线程，代替原来的工作线程
addWorker(null, false);
}
}

public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//检查权限
checkShutdownAccess();
//CAS 更新线程池状态
advanceRunState(SHUTDOWN);
//中断所有空闲的线程
interruptIdleWorkers();
//关闭，此处是do nothing
onShutdown();
} finally {
mainLock.unlock();
}
//尝试结束，上面代码已分析
tryTerminate();
}

public List<Runnable> shutdownNow() {
List<Runnable> tasks;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
advanceRunState(STOP);
//中断所有线程
interruptWorkers();
tasks = drainQueue();
} finally {
mainLock.unlock();
}
tryTerminate();
return tasks;
}

继承关系

Executor接口

public interface Executor {
void execute(Runnable command);
}

ExecutorService接口

public interface ExecutorService extends Executor {
void shutdown();
List<Runnable> shutdownNow();
boolean isShutdown();
boolean isTerminated();
boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException;
<T> Future<T> submit(Callable<T> task);
<T> Future<T> submit(Runnable task, T result);
Future<?> submit(Runnable task);
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException;
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
throws InterruptedException;
<T> T invokeAny(Collection<? extends Callable<T>> tasks)
throws InterruptedException, ExecutionException;
<T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException;
}

AbstractExecutorService抽象类

public abstract class AbstractExecutorService implements ExecutorService {
protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
return new FutureTask<T>(runnable, value);
}
protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
return new FutureTask<T>(callable);
}
public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
RunnableFuture<Void> ftask = newTaskFor(task, null);
execute(ftask);
return ftask;
}
public <T> Future<T> submit(Runnable task, T result) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task, result);
execute(ftask);
return ftask;
}
public <T> Future<T> submit(Callable<T> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task);
execute(ftask);
return ftask;
}
private <T> T doInvokeAny(Collection<? extends Callable<T>> tasks, boolean timed, long nanos)
throws InterruptedException, ExecutionException, TimeoutException {...}
public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
throws InterruptedException, ExecutionException {... }
public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {...}
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
throws InterruptedException {...}
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException {...}
}

构造函数与线程池状态

public ThreadPoolExecutor(int corePoolSize,                             //核心线程数
int maximumPoolSize,                      //最大线程数
long keepAliveTime,                       //线程存活时间
TimeUnit unit,                            //keepAliveTime的单位
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.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;
}

线程池状态

//记录线程池状态和线程数量（总共32位，前三位表示线程池状态，后29位表示线程数量）
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
//线程数量统计位数29  Integer.SIZE=32
private static final int COUNT_BITS = Integer.SIZE - 3;
//容量 000 11111111111111111111111111111
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;
//运行中 111 00000000000000000000000000000
private static final int RUNNING    = -1 << COUNT_BITS;
//关闭 000 00000000000000000000000000000
private static final int SHUTDOWN   =  0 << COUNT_BITS;
//停止 001 00000000000000000000000000000
private static final int STOP       =  1 << COUNT_BITS;
//整理 010 00000000000000000000000000000
private static final int TIDYING    =  2 << COUNT_BITS;
//终止 011 00000000000000000000000000000
private static final int TERMINATED =  3 << COUNT_BITS;
//获取运行状态（获取前3位）
private static int runStateOf(int c)     { return c & ~CAPACITY; }
//获取线程个数（获取后29位）
private static int workerCountOf(int c)  { return c & CAPACITY; }
private static int ctlOf(int rs, int wc) { return rs | wc; }

RUNNING：接受新任务并且处理阻塞队列里的任务
SHUTDOWN：拒绝新任务但是处理阻塞队列里的任务
STOP：拒绝新任务并且抛弃阻塞队列里的任务同时会中断正在处理的任务
TIDYING：所有任务都执行完（包含阻塞队列里面任务），当前线程池活动线程为0，将要调用terminated方法
TERMINATED：终止状态。terminated方法调用完成以后的状态
线程池状态转换：
RUNNING -> SHUTDOWN：显式调用shutdown()方法, 或者隐式调用了finalize()方法
(RUNNING or SHUTDOWN) -> STOP：显式调用shutdownNow()方法
SHUTDOWN -> TIDYING：当线程池和任务队列都为空的时候
STOP -> TIDYING：当线程池为空的时候
TIDYING -> TERMINATED：当 terminated() hook 方法执行完成时候

submit方法和execute方法的区别

submit方法

• 调用submit方法，传入Runnable或者Callable对象
• 判断传入的对象是否为null，为null则抛出异常，不为null继续流程
• 将传入的对象转换为RunnableFuture对象
• 执行execute方法，传入RunnableFuture对象
• 返回RunnableFuture对象

public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
RunnableFuture<Void> ftask = newTaskFor(task, null);
execute(ftask);
return ftask;
}
public <T> Future<T> submit(Runnable task, T result) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task, result);
execute(ftask);
return ftask;
}
public <T> Future<T> submit(Callable<T> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task);
execute(ftask);
return ftask;
}

execute方法

public void execute(Runnable command) {
//传进来的线程为null，则抛出空指针异常
if (command == null)
throw new NullPointerException();
//获取当前线程池的状态+线程个数变量
int c = ctl.get();
/**
* 3个步骤
*/
//1.判断当前线程池线程个数是否小于corePoolSize,小于则调用addWorker方法创建新线程运行,
//且传进来的Runnable当做第一个任务执行。
//如果调用addWorker方法返回false，则直接返回
if (workerCountOf(c) < corePoolSize) {
//添加一个core线程(核心线程)。此处参数的true，表示添加的线程是core容量下的线程
if (addWorker(command, true))
return;
//刷新数据，乐观锁就是没有锁
c = ctl.get();
}
/*  isRunning方法的定义：
private static boolean isRunning(int c)
{return c < SHUTDOWN;}
2.SHUTDOWN值为0，即如果c小于0，表示在运行；offer用来判断任务是否成功入队*/
if (isRunning(c) && workQueue.offer(command)) {
//二次检查
int recheck = ctl.get();
//如果当前线程池状态不是RUNNING则从队列删除任务，并执行拒绝策略
if (! isRunning(recheck) && remove(command))
//执行拒绝策略
reject(command);
//否则如果当前线程池线程空，则添加一个线程
else if (workerCountOf(recheck) == 0)
//添加一个空线程进线程池，使用非core容量线程
//仅有一种情况，会走这步，core线程数为0，max线程数>0,队列容量>0
//创建一个非core容量的线程，线程池会将队列的command执行
addWorker(null, false);
}
//线程池停止了或者队列已满，添加maximumPoolSize容量工作线程，如果失败，执行拒绝策略
else if (!addWorker(command, false))
reject(command);
}

ThreadPoolExecutor.addWorker()

    private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get(); //获取运行状态和工作数量
int rs = runStateOf(c); //获取当前线程池运行的状态
// Check if queue empty only if necessary.
//条件代表着以下几个场景，直接返回false说明当前工作线程创建失败
//1.rs>SHUTDOWN 此时不再接收新任务，且所有的任务已经执行完毕
//2.rs=SHUTDOWN 此时不再接收新任务，但是会执行队列中的任务
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
for (;;) {
int wc = workerCountOf(c);
//先判断当前活动的线程数是否大于最大值，如果超过了就直接返回false说明线程创建失败
//如果没有超过再根据core的值再进行以下判断
//1. core为true，则判断当前活动的线程数是否大于corePoolSize
//2. core为false，则判断当前活动线程数是否大于maximumPoolSize
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
//比较当前值是否和c相同，如果相同，则改为c+1，并且跳出大循环，直接执行Worker进行线程创建
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get();  // Re-read ctl
//检查下当前线程池的状态是否已经发生改变
//如果已经改变了，则进行外层retry大循环，否则只进行内层的循环
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 {
//Worker的也是Runnable的实现类
w = new Worker(firstTask);
//因为不可以直接在Worker的构造方法中进行线程创建
//所以要把它的引用赋给t方便后面进行线程创建
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.add(w);//将创建的线程添加到workers容器中
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}

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;
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
}

/** Delegates main run loop to outer runWorker  */
public void run() {
//这里执行的是ThreadPoolExecutor中的runWorker
runWorker(this);
}

ThreadPoolExecutor.runWorker()

final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;//获取Worker中的任务
w.firstTask = null; //将Woeker中的任务置空
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
//如果当前任务为空  那么就从getTask中获得任务
/**
* 如果task不为空，执行完task后则将task置空
* 继续进入循环，则从getTask中获取任务
*/
while (task != null || (task = getTask()) != null) {
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);
}
}

processWorkerExit

• 从While循环体中可以知道，当线程运行时出现异常，那么都会退出循环，进入到processWorkerExit()
• 从getTask()获得结果为null，则也会进到processWorkerExit()

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.
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
// Are workers subject to culling?
//如果设置了allowCoreThreadTimeOut(true)
//或者当前运行的任务数大于设置的核心线程数
// timed = true
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
/** ------------------------以上的操作跟之前类似----------------------- */
/** ------------------------关键在于下面的代码------------------------- */
/** ------------------------从阻塞队列中获取任务----------------------- */
try {
Runnable r = timed ?
//对于阻塞队列，poll(long timeout, TimeUnit unit) 将会在规定的时间内去任务
//如果没取到就返回null
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
//take会一直阻塞，等待任务的添加
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}

ThreadPoolExecutor.processWorkerExit()

    /**
* @param completedAbruptly
*/
private void processWorkerExit(Worker w, boolean completedAbruptly) {
if (completedAbruptly) //如果突然被打断，工作线程数不会被减少
decrementWorkerCount();
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
completedTaskCount += w.completedTasks;
workers.remove(w);
} finally {
mainLock.unlock();
}
tryTerminate();
int c = ctl.get();
//判断运行状态是否在STOP之前
if (runStateLessThan(c, STOP)) {
if (!completedAbruptly) {//正常退出，也就是task == null
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
if (min == 0 && ! workQueue.isEmpty())
min = 1;
if (workerCountOf(c) >= min)
return; // replacement not needed
}
//新增一个工作线程，代替原来的工作线程
addWorker(null, false);
}
}

线程池关闭

shutdown

public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//检查权限
checkShutdownAccess();
//CAS 更新线程池状态
advanceRunState(SHUTDOWN);
//中断所有空闲的线程
interruptIdleWorkers();
//关闭，此处是do nothing
onShutdown();
} finally {
mainLock.unlock();
}
//尝试结束，上面代码已分析
tryTerminate();
}

shutdownNow

public List<Runnable> shutdownNow() {
List<Runnable> tasks;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
advanceRunState(STOP);
//中断所有线程
interruptWorkers();
tasks = drainQueue();
} finally {
mainLock.unlock();
}
tryTerminate();
return tasks;
}

总结

• 线程池优先使用corePoolSize的数量执行工作任务
• 如果超过corePoolSize，队列入队
• 超过队列，使用maximumPoolSize-corePoolSize的线程处理，这部分线程超时不干活就销毁掉。
• 每个线程执行结束的时候，会判断当前的工作线程和任务数，如果任务数多，就会创建空线程从队列拿任务。
• 线程池执行完成，不会自动销毁，需要手工shutdown，修改线程池状态，中断所有线程。

分配线程池大小的依据

• 任务的性质：CPU密集型任务、IO密集型任务和混合型任务。
• 任务的优先级：高、中和低。
• 任务的执行时间：长、中和短。
• 任务的依赖性：是否依赖其他系统资源，如数据库连接。

使用有界队列

线程池监控

• taskCount：线程池需要执行的任务数量。
• completedTaskCount：线程池在运行过程中已完成的任务数量，小于或等于taskCount。
• largestPoolSize：线程池里曾经创建过的最大线程数量。通过这个数据可以知道线程池是否曾经满过。如该数值等于线程池的最大大小， 则表示线程池曾经满过。
• getPoolSize：线程池的线程数量。如果线程池不销毁的话，线程池里的线程不会自动销毁，所以这个大小只增不减。
• getActiveCount：获取活动的线程数。
• 通过扩展线程池进行监控。可以通过继承线程池来自定义线程池，重写线程池的beforeExecute、afterExecute和terminated方法， 也可以在任务执行前、执行后和线程池关闭前执行一些代码来进行监控。例如，监控任务的平均执行时间、最大执行时间和最小执行时间等。 这几个方法在线程池里是空方法。

关注微信公众号:【入门小站】,解锁更多知识点

以上是 【Java】Java线程池ThreadPoolExecutor源码分析 的全部内容， 来源链接： utcz.com/a/108046.html