ThreadPoolExecutor
- 初始化
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.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;
}
一个线程池的参数配置。当一个任务提交进来的时候，会出现以下几种情况：
| 提交任务线程池的情况 | 将执行的操作 |
|---|---|
| 线程池中线程小于corePoolSize | 创建核心线程，执行任务 |
| 线程池中的线程已达到corePoolSize，且workQueue未满 | 将任务提交到workQueue |
| workQueue已经满 | 创建临时线程（可创建临时线程的个数maximumPoolSize - corePoolSize）；时间配置keepAliveTime、unit是用于回收这些临时线程； |
| workQueue已满且线程池中线程数已达到maximumPoolSize | 执行配置的拒绝策略handler |
列出主要的成员变量
//int 低29为用于表示worker个数
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
//int 高三位用于表示线程池的状态
private static final int RUNNING = -1 << COUNT_BITS;// 111
private static final int SHUTDOWN = 0 << COUNT_BITS;// 000
private static final int STOP = 1 << COUNT_BITS;// 010
private static final int TIDYING = 2 << COUNT_BITS;// 011
private static final int TERMINATED = 3 << COUNT_BITS;
//ctl中包含了线程池状态和worker个数的信息
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
- 提交任务 execute
public void execute(Runnable command) {
//任务为null，抛出空指针异常
if (command == null)
throw new NullPointerException();
//c的初始值（线程池处于running状态，worker个数为0）
int c = ctl.get();
//线程池中的woker数量小于corePoolSize
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);
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
//尝试启动临时线程来执行任务
else if (!addWorker(command, false))
reject(command);
}
尝试添加一个worker线程并启动
//java.util.concurrent.ThreadPoolExecutor.addWorker(Runnable, boolean)
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
//线程池若状态不是Running，说明不能再提交任务了，直接返回false。
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
//尝试将worker数量加1，若超出了限制则直接返回false
for (;;) {
int wc = workerCountOf(c);
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
if (compareAndIncrementWorkerCount(c))
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 {
//初始化一个Worker，这里会通过线程工厂创建线程（Worker自身是Runnable）
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//持有锁了后重新检查线程池的状态
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);
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
//若添加worker成功后，这里会启动线程。（调用Worker的run方法）
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
//任务启动失败，会移除worker并修改ctl中worker的计数
addWorkerFailed(w);
}
return workerStarted;
}
//java.util.concurrent.ThreadPoolExecutor.Worker
//Worker可以看作一个Runnable
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;
//通过ThreadFactory来创建一个Thread
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() {
runWorker(this);
}
...
}
- 执行任务 runWorker
//java.util.concurrent.ThreadPoolExecutor.runWorker(Worker)
//每个Worker线程的run方法会调用这个方法
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
//这里就体现了线程池的复用，可以执行第一次addWorker是新增的任务。也可以尝试从阻塞队列中获取任务来执行。
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);
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;
//worker无论执行成功还是失败，总是将completedTasks加1
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
//当线程异常退出（抛出异常），completedAbruptly总是true
processWorkerExit(w, completedAbruptly);
}
}
//java.util.concurrent.ThreadPoolExecutor.processWorkerExit(Worker, boolean)
private void processWorkerExit(Worker w, boolean completedAbruptly) {
//当Worker执行抛出异常退出时，completedAbruptly一定为true
if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
//worker减1
decrementWorkerCount();
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
completedTaskCount += w.completedTasks;
workers.remove(w);
} finally {
mainLock.unlock();
}
tryTerminate();
int c = ctl.get();
if (runStateLessThan(c, STOP)) {
if (!completedAbruptly) {
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
if (min == 0 && ! workQueue.isEmpty())
min = 1;
if (workerCountOf(c) >= min)
return; // replacement not needed
}
//线程异常退出，会重新拉起来一个Worker
addWorker(null, false);
}
}
尝试从阻塞队列中获取任务（当此方法返回null会导致Worker线程终止，这也就是变相回收）
//java.util.concurrent.ThreadPoolExecutor.getTask()
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
//shutdown且阻塞队列为空；线程池状态>=STOP（阻塞队列有任务也忽略掉）
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
//判断是否需要清理线程
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
//从阻塞队列中获取任务
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
线程池中的状态
| 状态 | 描述 |
|---|---|
| RUNNING | 运行状态，可以接收新任务，处理阻塞队列中的任务 |
| SHUTDOWN | 不接受新提交的任务，会继续处理阻塞队列中的任务；调用shutdown方法线程池会转移到该状态 |
| STOP | 不接受新提交的任务，也不再处理阻塞队列中的任务，中断正在执行任务的线程；调用shutdownNow方法线程池会转移到该状态 |
| TIDYING | 此时workCount为0，这时会调用terminated方法；当存放任务的阻塞队列和worker线程被清空后，线程池会转移到该状态 |
| TERMINATED | terminated方法结束后，线程池会转移到该状态 |
ScheduledThreadPoolExecutor
scheduleAtFixedRate：周期性的执行任务，任务之间的时间间隔为固定的period
scheduleWithFixedDelay：周期性的执行任务，任务结束和下个任务开始之间会有delay time。
ScheduledThreadPoolExecutor$ScheduledFutureTask#run
public void run() {
boolean periodic = isPeriodic();
if (!canRunInCurrentRunState(periodic))
cancel(false);
else if (!periodic)
ScheduledFutureTask.super.run();
else if (ScheduledFutureTask.super.runAndReset()) {
setNextRunTime();
//这里又将task重新put进workQueue，这样就使得task可以周期性的执行。
reExecutePeriodic(outerTask);
}
}
网友评论