-
功能说明
-
类图
-
预热
-
Executor
-
ExecutorService
-
AbstractExecutorService
-
ThreadPoolExecutor
-
线程池状态
-
ThreadPoolExecutor大致结构
-
核心代码流程
-
execute
-
addWorker
-
Worker
-
runWorker
-
getTask
-
processWorkerExit
-
addWorkerFailed
-
tryTerminate
-
interruptWorkers-interruptIdleWorkers
-
ExecutorService接口实现
-
shutdown
-
shutdownNow
-
awaitTermination
-
isTerminated
-
其他辅助方法
-
finalize
-
提前启动线程
功能说明
使用线程池通过线程的重用，降低线程创建的开销，降低资源消耗，额外还增加了一些线程执行的管理功能，方便对线程执行状态的监控。
类图
先看Executor–ExecutorService–AbstractExecutorService–ThreadPoolExecutor这条线的源码。
预热
Executor
/** 主要是解耦command的提交和执行 */
public interface Executor {
/**
* 提交一个command，执行由实现类自己定义，可以直接执行，也可以新开线程执行等等
*/
void execute(Runnable command);
}
ExecutorService
接口ExecutorService继承Executor接口，提供线程终止的方法，扩展父接口execute()方法，提供future返回的submit方法。
public interface ExecutorService extends Executor {
/**
* 关闭之前的任务会继续执行，但是不会接受新的任务，如果已经关闭，调用不会有任务影响
*/
void shutdown();
/**
* 尝试停止正在执行的任务，暂停等待处理的任务，返回等待执行的任务列表
* 没办法保证一定能够停止执行中的任务
*/
List<Runnable> shutdownNow();
/**
* 执行器是否已经关闭
*/
boolean isShutdown();
/**
* 是否所有任务都已关闭，必须先调用shutdown或shutdownnow
*/
boolean isTerminated();
/**
* 阻塞等待所有任务都执行完毕
*/
boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException;
/**
* 下面3个方法都差不多，都是提交一个任务，返回一个Future，然后通过future.get()方法获取执行的结果
*/
<T> Future<T> submit(Callable<T> task);
<T> Future<T> submit(Runnable task, T result);
Future<?> submit(Runnable task);
/**
* 执行给定的任务集合，当所有任务完成时，返回Future的结果
*/
<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;
}
submit()方法参数，runnable、callable、future3者简单的区别：
runnable没有返回的执行，callable有返回值的，future是可取消的callable。
AbstractExecutorService
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);
}
2个newTask()方法分别将callable、runnable转换为future返回，看下FutureTask类图：
返回的FutureTask既可以当Runnable用，也可以当Future使用。
该抽象类实现ExecutorService的submit()方法会调用newTask()方法做转换：
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;
}
看下invokeAny()的实现：
public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
throws InterruptedException, ExecutionException {
try {
return doInvokeAny(tasks, false, 0);
} catch (TimeoutException cannotHappen) {
assert false;
return null;
}
}
public <T> T invokeAny(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {
return doInvokeAny(tasks, true, unit.toNanos(timeout));
}
/** invokeAny真正处理业务逻辑的地方，spring里面经常看到这样的方式，推荐使用 */
private <T> T doInvokeAny(Collection<? extends Callable<T>> tasks,
boolean timed, long nanos)
throws InterruptedException, ExecutionException, TimeoutException {
if (tasks == null)
throw new NullPointerException();
int ntasks = tasks.size();
if (ntasks == 0)
throw new IllegalArgumentException();
List<Future<T>> futures= new ArrayList<Future<T>>(ntasks);
ExecutorCompletionService<T> ecs =
new ExecutorCompletionService<T>(this); //将this的Executor再封装一层
// For efficiency, especially in executors with limited
// parallelism, check to see if previously submitted tasks are
// done before submitting more of them. This interleaving
// plus the exception mechanics account for messiness of main
// loop.
try {
// Record exceptions so that if we fail to obtain any
// result, we can throw the last exception we got.
ExecutionException ee = null;
long lastTime = timed ? System.nanoTime() : 0;
Iterator<? extends Callable<T>> it = tasks.iterator();
// Start one task for sure; the rest incrementally
futures.add(ecs.submit(it.next())); //先提交一个执行
--ntasks;
int active = 1;
for (;;) {
Future<T> f = ecs.poll(); //从ecs中poll个出来，然后检查结果
if (f == null) { //
if (ntasks > 0) {
--ntasks;
futures.add(ecs.submit(it.next()));
++active;
}
else if (active == 0)
break;
else if (timed) {
f = ecs.poll(nanos, TimeUnit.NANOSECONDS);
if (f == null)
throw new TimeoutException();
long now = System.nanoTime();
nanos -= now - lastTime;
lastTime = now;
}
else
f = ecs.take();
}
if (f != null) { //不为null
--active;
try {
return f.get(); //返回结果
} catch (ExecutionException eex) {
ee = eex;
} catch (RuntimeException rex) {
ee = new ExecutionException(rex);
}
}
}
if (ee == null)
ee = new ExecutionException();
throw ee; //异常存在就抛异常
} finally {
for (Future<T> f : futures)
f.cancel(true); //有异常或者有结果返回前，取消所有提交的任务
}
}
invokeAny()之前说过是执行一组任务，有一个完成就取消其他，这里都调用了doInvokeAny()来处理逻辑，大体流程是：
1、 先构造一个ExecutorCompletionService实例ecs(ecs简单解释，使用给定的执行器执行任务，按照给定的顺序返回执行结果future);；
2、 先提交一个任务，然后for循环检查ecs的future，为null那就分情况，如果还有任务那就再提交一个，或有超时设置，实在不行那就阻塞等待；不为null那就future.get()获取结果返回，如果future.get有异常那就设置异常；
3、 退出for循环的时候检查是否有异常，最后再return或throw异常前取消掉提交的任务；
看下invokeAll()的源码：
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
throws InterruptedException {
if (tasks == null)
throw new NullPointerException();
List<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
boolean done = false;
try {
for (Callable<T> t : tasks) { //执行所有任务
RunnableFuture<T> f = newTaskFor(t);
futures.add(f);
execute(f);
}
for (Future<T> f : futures) { //等待所有任务完成
if (!f.isDone()) {
try {
f.get();
} catch (CancellationException ignore) {
} catch (ExecutionException ignore) {
}
}
}
done = true;
return futures;
} finally {
if (!done) //没执行完，那就取消任务
for (Future<T> f : futures)
f.cancel(true);
}
}
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException {
if (tasks == null || unit == null)
throw new NullPointerException();
long nanos = unit.toNanos(timeout);
List<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
boolean done = false;
try {
for (Callable<T> t : tasks)
futures.add(newTaskFor(t));
long lastTime = System.nanoTime();
// Interleave time checks and calls to execute in case
// executor doesn't have any/much parallelism.
Iterator<Future<T>> it = futures.iterator();
while (it.hasNext()) {
execute((Runnable)(it.next()));
long now = System.nanoTime();
nanos -= now - lastTime;
lastTime = now;
if (nanos <= 0) //超时返回future
return futures;
}
//假如上面超时返回了，future还是有可能存在已经完成的任务
for (Future<T> f : futures) {
if (!f.isDone()) {
if (nanos <= 0)
return futures;
try {
f.get(nanos, TimeUnit.NANOSECONDS);
} catch (CancellationException ignore) {
} catch (ExecutionException ignore) {
} catch (TimeoutException toe) {
return futures;
}
long now = System.nanoTime();
nanos -= now - lastTime;
lastTime = now;
}
}
done = true;
return futures;
} finally {
if (!done)
for (Future<T> f : futures)
f.cancel(true);
}
}
大体的流程基本是1.提交任务；2.等待任务完成。有点不同的是超时的invokeAll()方法，即使超时返回了，future也有可能存在已经完成的任务，估计调用的时候还是要自己检查future是否null再加上future.get()是否null来判断。
ThreadPoolExecutor
线程池状态
/**ctl代表2个含义：高3位为线程池运行状态，低29位线程数量 */
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); //11100000000000000000000000000000
private static final int COUNT_BITS = Integer.SIZE - 3; //29
private static final int CAPACITY = (1 << COUNT_BITS) - 1; //00011111111111111111111111111111
// runState is stored in the high-order bits
/**
* RUNNING: 接收新的任务并且也会处理已经提交等待的任务
* SHUTDOWN: 不会接收新的任务，但会处理已经提交等待的任务
* STOP: 不接受新任务，不处理已经提交等待的任务，而且还会中断处理中的任务
* TIDYING: 所有的任务被终止，workCount为0，为此状态时将会调用terminated()方法
* TERMINATED: terminated()调用完成
*
* 状态流转：
* RUNNING -> SHUTDOWN
* On invocation of shutdown(), perhaps implicitly in finalize()
* (RUNNING or SHUTDOWN) -> STOP
* On invocation of shutdownNow()
* SHUTDOWN -> TIDYING
* When both queue and pool are empty
* STOP -> TIDYING
* When pool is empty
* TIDYING -> TERMINATED
* When the terminated() hook method has completed
*/
private static final int RUNNING = -1 << COUNT_BITS; //11100000000000000000000000000000
private static final int SHUTDOWN = 0 << COUNT_BITS; //00000000000000000000000000000000
private static final int STOP = 1 << COUNT_BITS; //00100000000000000000000000000000
private static final int TIDYING = 2 << COUNT_BITS; //01000000000000000000000000000000
private static final int TERMINATED = 3 << COUNT_BITS; //01100000000000000000000000000000
// Packing and unpacking ctl
private static int runStateOf(int c) { return c & ~CAPACITY; } //&操作比较高3位获取线程池状态
private static int workerCountOf(int c) { return c & CAPACITY; } // &比较低29位获取线程数量
private static int ctlOf(int rs, int wc) { return rs | wc; } //通过rs高3位运行状态|wc低29位线程数量计算最后的值
/*
* Bit field accessors that don't require unpacking ctl.
* These depend on the bit layout and on workerCount being never negative.
*/
private static boolean runStateLessThan(int c, int s) {
return c < s;
}
private static boolean runStateAtLeast(int c, int s) {
return c >= s;
}
private static boolean isRunning(int c) {
return c < SHUTDOWN;
}
/**
* Attempt to CAS-increment the workerCount field of ctl.
*/
private boolean compareAndIncrementWorkerCount(int expect) {
return ctl.compareAndSet(expect, expect + 1);
}
/**
* Attempt to CAS-decrement the workerCount field of ctl.
*/
private boolean compareAndDecrementWorkerCount(int expect) {
return ctl.compareAndSet(expect, expect - 1);
}
/**
* Decrements the workerCount field of ctl. This is called only on
* abrupt termination of a thread (see processWorkerExit). Other
* decrements are performed within getTask.
*/
private void decrementWorkerCount() {
do {} while (! compareAndDecrementWorkerCount(ctl.get()));
}
private void advanceRunState(int targetState) {
for (;;) {
int c = ctl.get();
if (runStateAtLeast(c, targetState) ||
ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))
break;
}
}
主要是ctl高低位分离，之前读写锁的时候也有这种处理。
ThreadPoolExecutor大致结构
/**
* 一般来说，如果线程池大小没到corePoolSize大小，会新增线程运行，如果到了，那就入workQueue队列，等线程运行
*/
private final BlockingQueue<Runnable> workQueue;
/**
* 一些操作需要加锁
*/
private final ReentrantLock mainLock = new ReentrantLock();
/**
* 线程池，所有的线程
*/
private final HashSet<Worker> workers = new HashSet<Worker>();
/**
* awaitTermination时条件
*/
private final Condition termination = mainLock.newCondition();
/**
* 线程池线程最大数量
*/
private int largestPoolSize;
/**
* 已经结束的任务，只有在关闭线程池的时候才累加
*/
private long completedTaskCount;
/**
* 线程池工厂，里面有一个newThread()方法用来产生工作线程，如果构造没提供，默认有一个
*/
private volatile ThreadFactory threadFactory;
/**
* 大部分用在线程池满了以后，新的任务过来，使用那种拒绝策略，默认会提供一个
*/
private volatile RejectedExecutionHandler handler;
/**
* 线程数量大于corePoolSize时，线程可以空闲的时间，如果设置了allowCoreThreadTimeOut，小于corePoolSize时也一样处理，否则就等待任务到来
*/
private volatile long keepAliveTime;
/**
* false，核心线程空闲等待，true的话就是用keepAliveTime超时控制获取任务
*/
private volatile boolean allowCoreThreadTimeOut;
/**
* 核心线程数量
*/
private volatile int corePoolSize;
/**
* 跟largestPoolSize这个不一样，这个用来控制线程池大小
*/
private volatile int maximumPoolSize;
/**
* 默认的拒绝策略
*/
private static final RejectedExecutionHandler defaultHandler =
new AbortPolicy();
构造函数：
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), defaultHandler);
}
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
threadFactory, defaultHandler);
}
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
RejectedExecutionHandler handler) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), handler);
}
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;
}
构造函数基本就是对那些变量的检查赋值。
在看execute()方法前先看下默认提供的线程工厂和拒绝策略。
/** ThreadPoolExecutor构造的时候如果不传入，默认这个，用于产生工作线程 */
static class DefaultThreadFactory implements ThreadFactory {
private static final AtomicInteger poolNumber = new AtomicInteger(1);
private final ThreadGroup group;
private final AtomicInteger threadNumber = new AtomicInteger(1);
private final String namePrefix;
DefaultThreadFactory() {
SecurityManager s = System.getSecurityManager();
group = (s != null) ? s.getThreadGroup() :
Thread.currentThread().getThreadGroup();
namePrefix = "pool-" +
poolNumber.getAndIncrement() +
"-thread-";
}
/** 返回一个线程：name为pool-线程池序列-thread-线程序列，优先级为5，非守护线程 */
public Thread newThread(Runnable r) {
Thread t = new Thread(group, r,
namePrefix + threadNumber.getAndIncrement(),
0);
if (t.isDaemon())
t.setDaemon(false);
if (t.getPriority() != Thread.NORM_PRIORITY)
t.setPriority(Thread.NORM_PRIORITY);
return t;
}
}
线程池构造如果不传参，默认提供了一个拒绝策略AbortPolicy
public static class CallerRunsPolicy implements RejectedExecutionHandler {
/**
* Creates a {@code CallerRunsPolicy}.
*/
public CallerRunsPolicy() { }
/**
* 如果线程池没有关闭，在调用者线程直接运行该任务，否则discard
*/
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
r.run();
}
}
}
public static class AbortPolicy implements RejectedExecutionHandler {
/**
* Creates an {@code AbortPolicy}.
*/
public AbortPolicy() { }
/**
* throws RejectedExecutionException.直接抛出异常
*/
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
throw new RejectedExecutionException("Task " + r.toString() +
" rejected from " +
e.toString());
}
}
public static class DiscardPolicy implements RejectedExecutionHandler {
/**
* Creates a {@code DiscardPolicy}.
*/
public DiscardPolicy() { }
/**
* 什么都不做
*/
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
}
}
public static class DiscardOldestPolicy implements RejectedExecutionHandler {
/**
* Creates a {@code DiscardOldestPolicy} for the given executor.
*/
public DiscardOldestPolicy() { }
/**
* 线程池没有关闭就把任务队列的第一个丢弃然后执行新的
*/
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
e.getQueue().poll();
e.execute(r);
}
}
}
实在不满意提供的，那就自己写个吧。
核心代码流程
按照代码调用流程来分解。
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(); //double-check
if (! isRunning(recheck) && remove(command)) //线程池关闭，remove任务，按照给定的策略reject
reject(command);
else if (workerCountOf(recheck) == 0) //如果线程池还在或者remove失败，但是核心线程已经没有了，那就开一个执行队列中的任务
addWorker(null, false);
}
else if (!addWorker(command, false)) //如果检查的时候线程池就关闭了或者线队列满了入队列失败，那就再开一个线程，如果失败那就reject
reject(command);
}
addWorker
/**
* 基于当前线程池的状态和大小(core:true使用corePoolSize，false使用maximumPoolSize)，能否加入一个新的工作类
*/
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
/** 1.如果为Running状态无所谓，不管新增线程还是接收新任务都可以
* 2.如果为shutdown状态那就只能处理任务，不能接收任务。这时入参firstTask必须null且，队列不为空，可以增加一个线程来处理队列，但不能新增任务
* 3.其他状态直接false
*/
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
for (;;) {
int wc = workerCountOf(c);
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize)) //根据core检验大小
return false;
if (compareAndIncrementWorkerCount(c)) //线程数量加1
break retry;
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 {
final ReentrantLock mainLock = this.mainLock;
w = new Worker(firstTask);
final Thread t = w.thread; //这里产生Thread时候可能为null，workerStarted为false，finally会处理addWorkerFailed
if (t != null) {
mainLock.lock(); //线程池主锁
try {
// Recheck while holding lock.
// Back out on ThreadFactory failure or if
// shut down before lock acquired.
int c = ctl.get();
int rs = runStateOf(c);
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) // precheck that t is startable 刚开始对这里的处理感觉又疑问，后来想到可以自定义线程工厂newThread，存在提前启动线程可能
throw new IllegalThreadStateException();
workers.add(w); //加入线程池
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s; //设置线程池大小
workerAdded = true;
}
} finally {
mainLock.unlock(); //解锁
}
if (workerAdded) { //true说明已经加入线程池，那就start吧
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted) //线程t为null或有异常产生，workerStarted为false
addWorkerFailed(w);
}
return workerStarted;
}
Worker
addWorker中用到了Worker：
w = new Worker(firstTask);
final Thread t = w.thread;
...
t.start();
看下Worker类：
/**
* Class Worker mainly maintains interrupt control state for
* threads running tasks, along with other minor bookkeeping.
* This class opportunistically extends AbstractQueuedSynchronizer
* to simplify acquiring and releasing a lock surrounding each
* task execution. This protects against interrupts that are
* intended to wake up a worker thread waiting for a task from
* instead interrupting a task being run. We implement a simple
* non-reentrant mutual exclusion lock rather than use
* ReentrantLock because we do not want worker tasks to be able to
* reacquire the lock when they invoke pool control methods like
* setCorePoolSize. Additionally, to suppress interrupts until
* the thread actually starts running tasks, we initialize lock
* state to a negative value, and clear it upon start (in
* runWorker).
* 英文保留，便于后面的时候对照，以免歧义：
* Worker实现Runnable接口，所以本身可以作为参数传给Thread运行，继承AQS，实现的是独占模式的api
* 主要是为了控制线程task运行的中断状态，每次运行前后加解锁
* 没有使用可重入锁，主要是为了避免获得锁的情况下去修改线程池的一些加锁的方法
* AQS的初始状态为-1，后面lock为1，unlock为0，可以简单判断state大于等于0值来判断是否运行过
*/
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
/**
* This class will never be serialized, but we provide a
* serialVersionUID to suppress a javac warning.
*/
private static final long serialVersionUID = 6138294804551838833L;
/** 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; //每个线程完成的任务统计
/**
* Creates with given first task and thread from ThreadFactory.
* @param firstTask the first task (null if none)
*/
Worker(Runnable firstTask) {
setState(-1); // 初始为-1
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this); //线程工厂newThread
}
/** Delegates main run loop to outer runWorker */
public void run() {
runWorker(this); //线程运行后最后会调用ThreadPoolExecutor的runWorker，后面看
}
// AQS的独占api实现
//
// 0表示unlocked
// 1表示lock
protected boolean isHeldExclusively() {
return getState() != 0;
}
protected boolean tryAcquire(int unused) {
if (compareAndSetState(0, 1)) {
setExclusiveOwnerThread(Thread.currentThread());
return true;
}
return false;
}
protected boolean tryRelease(int unused) {
setExclusiveOwnerThread(null);
setState(0);
return true;
}
public void lock() { acquire(1); }
public boolean tryLock() { return tryAcquire(1); }
public void unlock() { release(1); }
public boolean isLocked() { return isHeldExclusively(); }
void interruptIfStarted() {
Thread t;
if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) { //检查aqs状态和线程状态，只有运行过采取中断
try {
t.interrupt();
} catch (SecurityException ignore) {
}
}
}
}
runWorker
调用t.start()后，最后调用到runWorker()，这是线程处理任务最重要的方法。runWorker()大概处理流程是：初始firstTask不为null就先处理这个任务，然后循环再从workQueue获取getTask来处理，如果队列也取不到任务了，那就退出了，这时候就要看keepAliveTime和allowCoreThreadTimeOut了决定是，大概先这样理解，细节再说。
/**
* Main worker run loop. Repeatedly gets tasks from queue and
* executes them, while coping with a number of issues:
*
* 1. We may start out with an initial task, in which case we
* don't need to get the first one. Otherwise, as long as pool is
* running, we get tasks from getTask. If it returns null then the
* worker exits due to changed pool state or configuration
* parameters. Other exits result from exception throws in
* external code, in which case completedAbruptly holds, which
* usually leads processWorkerExit to replace this thread.
*
* 2. Before running any task, the lock is acquired to prevent
* other pool interrupts while the task is executing, and
* clearInterruptsForTaskRun called to ensure that unless pool is
* stopping, this thread does not have its interrupt set.
*
* 3. Each task run is preceded by a call to beforeExecute, which
* might throw an exception, in which case we cause thread to die
* (breaking loop with completedAbruptly true) without processing
* the task.
*
* 4. Assuming beforeExecute completes normally, we run the task,
* gathering any of its thrown exceptions to send to
* afterExecute. We separately handle RuntimeException, Error
* (both of which the specs guarantee that we trap) and arbitrary
* Throwables. Because we cannot rethrow Throwables within
* Runnable.run, we wrap them within Errors on the way out (to the
* thread's UncaughtExceptionHandler). Any thrown exception also
* conservatively causes thread to die.
*
* 5. After task.run completes, we call afterExecute, which may
* also throw an exception, which will also cause thread to
* die. According to JLS Sec 14.20, this exception is the one that
* will be in effect even if task.run throws.
*
* The net effect of the exception mechanics is that afterExecute
* and the thread's UncaughtExceptionHandler have as accurate
* information as we can provide about any problems encountered by
* user code.
*
* @param w the worker
*/
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask; //线程初始的task
w.firstTask = null;
w.unlock(); // allow interrupts 这里是worker构造默认的aqs为-1，这里unlock下，容许中断
boolean completedAbruptly = true; //线程退出的原因，true是任务导致，false是线程正常退出
try {
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; //线程正常退出，false
} finally {
processWorkerExit(w, completedAbruptly); //work退出，做收尾工作
}
}
getTask
/**
* 存在阻塞或超时获取任务或由于下列原因返回null:
* 1. 超过设置的线程池大小maximumPoolSize；
* 2. 线程池stop.
* 3. 线程池shutdown，队列为空.
* 4. 线程超时等待任务，This worker timed out waiting for a task, and timed-out
* workers are subject to termination (that is,
* {@code allowCoreThreadTimeOut || workerCount > corePoolSize})
* both before and after the timed wait.
*
* @return task, or null if the worker must exit, in which case
* workerCount is decremented
*/
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out? 控制超时获取poll后是否退出，return null
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// Check if queue empty only if necessary.
// shutdown时如果队列空，那就直接worker减1，返回null
// Stop往后的状态，那就null，这时候也不会去处理任务
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
//到这里的时候线程池状态为running或者shutdown但队列不空
boolean timed; // 是否超时获取task
for (;;) {
int wc = workerCountOf(c);
/**
* 1.容许核心线程超时即allowCoreThreadTimeOut为true
* 2.不容许容许核心线程超时即allowCoreThreadTimeOut为false，检查线程池数量：
* 如果大于核心线程数量，那就超时poll获取，否则核心线程阻塞获取
*/
timed = allowCoreThreadTimeOut || wc > corePoolSize;
//timeout第一次为false，所以肯定会有一次poll或take发生，这里控制的应该是在发生一次后是否还继续超时获取
if (wc <= maximumPoolSize && ! (timedOut && timed))
break;
if (compareAndDecrementWorkerCount(c)) //不需要，那就扣减返回null
return null;
c = ctl.get(); // Re-read ctl
if (runStateOf(c) != rs)
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
try {
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false; //响应中断，重新开始
}
}
}
processWorkerExit
在上面runWorker()中如果getTask()返回null或者在用户task.run过程中有异常出现时需要对线程退出做收尾工作processWorkerExit():
/**
* 线程退出收尾
*/
private void processWorkerExit(Worker w, boolean completedAbruptly) {
if (completedAbruptly) // 为true的时候，用户线程运行异常，需要扣减，false的时候为getTask方法中扣减线程数量
decrementWorkerCount();
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
completedTaskCount += w.completedTasks;
workers.remove(w); //从线程池移除
} finally {
mainLock.unlock();
}
tryTerminate(); //有worker线程移除，可能是最后一个线程退出需要尝试终止线程池
int c = ctl.get();
if (runStateLessThan(c, STOP)) { //线程池可能为running或shutdown状态
if (!completedAbruptly) { //这里是线程正常退出
int min = allowCoreThreadTimeOut ? 0 : corePoolSize; //检查allowCoreThreadTimeOut字段，看是否需要维持核心线程数量
if (min == 0 && ! workQueue.isEmpty()) //如果不需要维持核心线程数量但是队列不空，那至少保持一个线程处理队列
min = 1;
if (workerCountOf(c) >= min) //如果线程数量大于最少数量，直接返回，否则下面至少要addWorker一个
return; // replacement not needed
}
addWorker(null, false); //用户异常退出，或者线程数量小于需要维护的数量，那就add
}
}
addWorkerFailed
最开始的时候addWorker()时产生thread失败或线程已经提前启动时的处理.
/**
* 回滚worker创建
*/
private void addWorkerFailed(Worker w) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
if (w != null)
workers.remove(w); // 1.有可能已经加入线程池，remove
decrementWorkerCount(); //2.线程池数量减1
tryTerminate(); //3.涉及线程的扣钱都有要去尝试终止线程池
} finally {
mainLock.unlock();
}
}
tryTerminate
在addworker失败和work处理任务失败退出都涉及到线程的扣钱，可能就剩下这一个线程，就需要关闭线程池，所以需要尝试下关闭线程池。
/**
* 尝试终止线程池
*/
final void tryTerminate() {
for (;;) {
int c = ctl.get();
//线程池状态为Running\状态已经是TIDYING或TERMINATED说明已经终止了\SHUTDOWN但是队列不空还需要处理任务，不需要终止
if (isRunning(c) ||
runStateAtLeast(c, TIDYING) ||
(runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
return;
// 到这里只剩下，shutdown状态队列空，Stop状态
if (workerCountOf(c) != 0) {
//线程池还有线程，但是队列没有任务了，需要中断唤醒等待任务的线程（runwoker的时候首先就通过w.unlock设置线程可中断，getTask最后面的catch处理中断）
//中断空闲线程传入的true，唤醒一个等待线程来处理就行
interruptIdleWorkers(ONLY_ONE);
return;
}
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) { //线程池状态转为TIDYING
try {
terminated();
} finally {
ctl.set(ctlOf(TERMINATED, 0)); //线程池状态转为TERMINATED
termination.signalAll(); //awaitTermination的termination.awaitNanos(nanos)需要signal
}
return;
}
} finally {
mainLock.unlock();
}
// else retry on failed CAS
}
}
/** 空实现，子类覆盖，最好先调用super.terminated() */
protected void terminated() { }
interruptWorkers-interruptIdleWorkers
/**
* 调用worker的interruptIfStarted，中断已经start的所有线程
*/
private void interruptWorkers() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (Worker w : workers)
w.interruptIfStarted();
} finally {
mainLock.unlock();
}
}
/**
* 根据参数中断一个还是所有空闲线程，注意是空闲线程，interruptWorkers这个方法，不管空闲不空闲都会中断
*/
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()) { //trylock保证是空闲的线程
try {
t.interrupt();
} catch (SecurityException ignore) {
} finally {
w.unlock();
}
}
if (onlyOne)
break;
}
} finally {
mainLock.unlock();
}
}
/**
* 还是调用上面那个有参数的方法，防止忘记true\false含义
*/
private void interruptIdleWorkers() {
interruptIdleWorkers(false);
}
/** 中断空闲线程时只处理一个 */
private static final boolean ONLY_ONE = true;
注意：interruptWorkers中断所有线程，interruptIdleWorkers是中断空闲的线程。
ExecutorService接口实现
shutdown
/** 1.推进状态到shutdown，不接受新任务(addworker时检查状态会控制) 2.中断所有空闲线程，会检查线程池状态和队列是否空，保证已经提交的肯定会执行*/
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
advanceRunState(SHUTDOWN); //推进线程池状态
interruptIdleWorkers(); //中断空闲的线程
onShutdown(); // hook for ScheduledThreadPoolExecutor
} finally {
mainLock.unlock();
}
tryTerminate(); //最后线程池尝试关闭下
}
/**
* 推进状态到targetState
*
* @param targetState the desired state, either SHUTDOWN or STOP
* (but not TIDYING or TERMINATED -- use tryTerminate for that)
*/
private void advanceRunState(int targetState) {
for (;;) {
int c = ctl.get();
if (runStateAtLeast(c, targetState) ||
ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))
break;
}
}
/** 空实现，子类覆盖*/
void onShutdown() {
}
shutdownNow
/** 中断所有线程，不管是执行中还是getTask等待的，返回没执行的task列表 */
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;
}
/**
* 从元队列删除，添加到返回队列
*/
private List<Runnable> drainQueue() {
BlockingQueue<Runnable> q = workQueue;
List<Runnable> taskList = new ArrayList<Runnable>();
q.drainTo(taskList);
if (!q.isEmpty()) {
for (Runnable r : q.toArray(new Runnable[0])) {
if (q.remove(r))
taskList.add(r);
}
}
return taskList;
}
awaitTermination
/** 之前的shutdown和shutdownnow是不等待线程池变为中断状态的，这里等待指定超时时间 */
public boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException {
long nanos = unit.toNanos(timeout);
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (;;) {
if (runStateAtLeast(ctl.get(), TERMINATED))
return true;
if (nanos <= 0)
return false;
nanos = termination.awaitNanos(nanos); //tryTerminate()会termination.signalAll()
}
} finally {
mainLock.unlock();
}
}
isTerminated
public boolean isShutdown() {
return ! isRunning(ctl.get());
}
/**
* 在running和TERMINATED之间的3个状态
*/
public boolean isTerminating() {
int c = ctl.get();
return ! isRunning(c) && runStateLessThan(c, TERMINATED);
}
/** TERMINATED状态 */
public boolean isTerminated() {
return runStateAtLeast(ctl.get(), TERMINATED);
}
其他辅助方法
上面看完基本线程这个类就Ok了，这里看下一些辅助方法。
finalize
/**
* Invokes {@code shutdown} when this executor is no longer
* referenced and it has no threads.
*/
protected void finalize() {
shutdown(); //默认调用shutdown处理已提交的任务
}
提前启动线程
/**
* 启动一个线程
*/
public boolean prestartCoreThread() {
return workerCountOf(ctl.get()) < corePoolSize &&
addWorker(null, true);
}
/**
* 确保肯定启动一个线程
*/
void ensurePrestart() {
int wc = workerCountOf(ctl.get());
if (wc < corePoolSize)
addWorker(null, true);
else if (wc == 0)
addWorker(null, false);
}
/**
* 启动所有核心线程
*/
public int prestartAllCoreThreads() {
int n = 0;
while (addWorker(null, true))
++n;
return n;
}
其他的方法大都是get\set方法，不关心，线程池这个类还需要多看看，有些细节需要深入下，这个看懂了，才能去看其他的。