一、执行任务
public void execute(Runnable command) {
// 如果任务为空，直接抛异常
if (command == null)
throw new NullPointerException();
/*
* Proceed in 3 steps:
*
* 1. If fewer than corePoolSize threads are running, try to
* start a new thread with the given command as its first
* task. The call to addWorker atomically checks runState and
* workerCount, and so prevents false alarms that would add
* threads when it shouldn't, by returning false.
*
* 2. If a task can be successfully queued, then we still need
* to double-check whether we should have added a thread
* (because existing ones died since last checking) or that
* the pool shut down since entry into this method. So we
* recheck state and if necessary roll back the enqueuing if
* stopped, or start a new thread if there are none.
*
* 3. If we cannot queue task, then we try to add a new
* thread. If it fails, we know we are shut down or saturated
* and so reject the task.
*/
// 获取线程池运行状态
int c = ctl.get();
// 如果运行的线程数小于核心数，添加worker
if (workerCountOf(c) < corePoolSize) {
// 添加worker，并将core设为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);
}
// 走到这里，说明核心线程数用完且任务队列已满，那么启用非核心线程数，如果失败，执行拒绝策略
else if (!addWorker(command, false))
reject(command);
}
- 首先用核心线程执行任务，如果核心线程已满，将任务添加到任务队列；如果队列也满了，那么用非核心线程执行任务
- addWorker(Runnable firstTask, boolean core)第一个参数是执行的任务，第二个参数如果为true，表示用的是核心线程，false表示用的是非核心线程
- 成员变量ctl是AtomicInteger类型，用来表示线程运行状态和线程数，高3位表示运行状态，低29位表示运行线程数
private boolean addWorker(Runnable firstTask, boolean core) {
//retry用来判断是否可以添加任务，并更新线程数
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// Check if queue empty only if necessary.
// 如果线程池已经关闭且没有任务，直接返回false
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
for (;;) {
// 获取工作线程数
int wc = workerCountOf(c);
// 如果工作线程数大于等于最大线程数，直接返回false
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
// 更新线程数，如果成功，跳出retry
if (compareAndIncrementWorkerCount(c))
break retry;
// 走到这，说明更新线程数失败了，重新获取线程池状态
c = ctl.get(); // Re-read ctl
// 如果线程池状态变化了，从retry重新执行；如果线程池状态没有变化，继续for循环
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对象
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)) {
// 因为线程还没启动，所以这里线程是alive，说明是不正常的
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
// 将新创建的worker加入到workers集合
workers.add(w);
int s = workers.size();
// 更新largestPoolSize值
if (s > largestPoolSize)
largestPoolSize = s;
// workerAdded标记为true
workerAdded = true;
}
} finally {
mainLock.unlock();
}
// 如果任务添加完成，启动线程且将workerStarted标记为true
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
// 如果启动线程失败，从workers删除新创建的任务，且执行tryTerminate
if (! workerStarted)
addWorkerFailed(w);
}
// 最后返回线程是否启动成功
return workerStarted;
}
- 首先去更新工作的线程数
- 创建Worker对象，此时会创建Thread类型的成员变量thread，Worker对象会传入到该线程，因为Worker对象实现了Runnable方法，所以启动线程thread时，会执行Worker的run方法
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
二、执行任务
1. 发起任务执行
public void run() {
runWorker(this);
}
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
// 用来表示是否正常执行任务，true表示被打断了，false表示未被打断
boolean completedAbruptly = true;
try {
// 如果worker对象有传入了任务或者任务队列有任务
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();
}
}
// 被中断标记设为false
completedAbruptly = false;
} finally {
// 任务完成后的操作
processWorkerExit(w, completedAbruptly);
}
}
- 首先执行传入worker对象里的任务，如果为空，则从任务队列里获取任务
- 判断是否需要打断线程
- 执行任务
- 任务完成相关的操作
2. 获取任务
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.
// 如果线程池状态关闭且任务队列为空
// 或者
// 线程池状态是停止
// 那么将线程数减1，返回空任务
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
// 获取线程数
int wc = workerCountOf(c);
// Are workers subject to culling?
// 如果allowCoreThreadTimeOut为真，表示核心线程也有超时时间，一般默认为false
// 或者
// 工作线程数超过核心线程数
// 那么将timed设为真，设为真的目的是为了没任务的时候，减少工作线程的数量
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
// 如果工作线程大于最大线程数或者超时了
// 且
// 工作线程数大于1或者任务队列不为空
// 那么工作线程减1
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;
// 如果任务为空，将超时设置为true
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
- 这部分代码能说明线程复用，超时未获取到任务减少线程的原理
- 如果设置了核心线程有超时时间或者线程数超过了核心线程数，那么采用带超时的方式获取任务，如果没有获取到任务，那么线程数会减1；如果不采用带超时的方式获取任务，那么一直等待，知道从任务队列里获取了任务
3. 退出任务执行
private void processWorkerExit(Worker w, boolean completedAbruptly) {
if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
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)) {
// 如果任务执行被中断了，保证至少还有1个工作线程在执行
if (!completedAbruptly) {
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
if (min == 0 && ! workQueue.isEmpty())
min = 1;
if (workerCountOf(c) >= min)
return; // replacement not needed
}
// 相当于线程还在工作
addWorker(null, false);
}
}
- 从任务集合中删除当前任务
- 尝试终止线程池
- 如果线程池状态是小于STOP，保证有工作线程在工作
final void tryTerminate() {
for (;;) {
int c = ctl.get();
// isRunning(c) 表示在运行，不能停止
// runStateAtLeast(c, TIDYING)表示已经停止了，没必要停止
// (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty())表示关闭但是有任务，不能停止
if (isRunning(c) ||
runStateAtLeast(c, TIDYING) ||
(runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
return;
// 如果工作线程数不等于0，停止1个空闲的工作线程，通过tryLock判断是否空闲
if (workerCountOf(c) != 0) { // Eligible to terminate
interruptIdleWorkers(ONLY_ONE);
return;
}
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 走到这，说明工作线程是0了，将状态改为TIDYING
if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
try {
// 终止线程池，空方法，留给子类实现
terminated();
} finally {
// 最后将状态改为TERMINATED，通知等待线程池终止的线程
ctl.set(ctlOf(TERMINATED, 0));
termination.signalAll();
}
return;
}
} finally {
mainLock.unlock();
}
// else retry on failed CAS
}
}
- 如果线程池状态不在运行，且工作线程数为0，那么最终将线程池状态改为TERMINATED
三、拒绝策略
1. 直接抛异常(默认策略)
public static class AbortPolicy implements RejectedExecutionHandler {
/**
* Creates an {@code AbortPolicy}.
*/
public AbortPolicy() { }
/**
* Always throws RejectedExecutionException.
*
* @param r the runnable task requested to be executed
* @param e the executor attempting to execute this task
* @throws RejectedExecutionException always
*/
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
throw new RejectedExecutionException("Task " + r.toString() +
" rejected from " +
e.toString());
}
}
2. 调用者的线程执行任务
public static class CallerRunsPolicy implements RejectedExecutionHandler {
/**
* Creates a {@code CallerRunsPolicy}.
*/
public CallerRunsPolicy() { }
/**
* Executes task r in the caller's thread, unless the executor
* has been shut down, in which case the task is discarded.
*
* @param r the runnable task requested to be executed
* @param e the executor attempting to execute this task
*/
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
r.run();
}
}
}
3. 丢掉最早的任务
public static class DiscardOldestPolicy implements RejectedExecutionHandler {
/**
* Creates a {@code DiscardOldestPolicy} for the given executor.
*/
public DiscardOldestPolicy() { }
/**
* Obtains and ignores the next task that the executor
* would otherwise execute, if one is immediately available,
* and then retries execution of task r, unless the executor
* is shut down, in which case task r is instead discarded.
*
* @param r the runnable task requested to be executed
* @param e the executor attempting to execute this task
*/
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
e.getQueue().poll();
e.execute(r);
}
}
}
4. 空的策略
public static class DiscardPolicy implements RejectedExecutionHandler {
/**
* Creates a {@code DiscardPolicy}.
*/
public DiscardPolicy() { }
/**
* Does nothing, which has the effect of discarding task r.
*
* @param r the runnable task requested to be executed
* @param e the executor attempting to execute this task
*/
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
}
}
四、线程工厂
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-";
}
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;
}
}
五、常见四种线程池
1.newFixedThreadPool
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>(),
threadFactory);
}
- 可自定义线程数和线程工厂，核心线程数与最大线程数相等，这样的话线程一旦创建，就会一直运行
2.newCachedThreadPool
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>(),
threadFactory);
}
- 只可以自定义线程工厂，核心线程数为0，最大线程数为Integer最大值
- 相当于没有限制工作线程数，任务量大的时候，会影响机器性能
- 空任务的时候，会有1个线程在运行
3.newSingleThreadScheduledExecutor
public static ScheduledExecutorService newSingleThreadScheduledExecutor() {
return new DelegatedScheduledExecutorService
(new ScheduledThreadPoolExecutor(1));
}
public static ScheduledExecutorService newSingleThreadScheduledExecutor(ThreadFactory threadFactory) {
return new DelegatedScheduledExecutorService
(new ScheduledThreadPoolExecutor(1, threadFactory));
}
- 只创建一个工作线程，可自定义线程工厂，可以定时执行任务
4.newScheduledThreadPool
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
return new ScheduledThreadPoolExecutor(corePoolSize);
}
public static ScheduledExecutorService newScheduledThreadPool(
int corePoolSize, ThreadFactory threadFactory) {
return new ScheduledThreadPoolExecutor(corePoolSize, threadFactory);
}
- 可定义线程工厂和核心工作线程数，最大工作线程数为Integer.MAX_VALUE
- 可定时执行任务
