线程池
为什么要使用线程池？
- 创建/销毁线程消耗系统资源，线程池可以复用线程
- 控制并发数量。并发过多，资源可能耗尽，造成服务崩溃
- 对线程统一管理
构建方法
- corePoolSize：核心线程数
- maximumPoolSize：最大线程数
- keepAliveTime ：非核心线程数闲置时长
- TimeUnit unit： keepAliveTime 时长单位
BlockingQueue<Runnable> workQueue: 阻塞队列，维护 等待 执行的Runnable- ThreadFactory threadFactory：创建线程的工厂，统一在创建线程是设置一些参数，如是否为守护线程，线程名称等。
- RejectedExecutionHandler handler：拒绝处理策略，线程数量大于最大线程数和阻塞队列长度的拒绝处理策略。
- AbortPolicy：默认策略，丢弃任务抛出 RejectedExecutionException
- DiscardPolicy: 丢弃任务，不抛出异常
- DiscardOldestPolicy: 丢弃头部任务（最旧的任务)，然后尝试重新执行
- CallerRunsPolicy: 由调用线程处理
ThreadPoolExecutor 状态
ThreadPoolExecutor 中控制变量叫做 ctl,它是一个 AtomicInter 类型变量。
控制变量由两部分组成
- runState 高3位，线程状态
- workerCount 线程数量，低29位
线程池有自己的状态，使用 int 的高3位bit表示。
线程池容量为 2^29 - 1,大概5亿多。
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
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
// 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;
// 所有任务结束，workCount = 0, 准备运行 terminated() hook
private static final int TIDYING = 2 << COUNT_BITS;
//terminated() 运行完成
private static final int TERMINATED = 3 << COUNT_BITS;
- 线程池创建后 处于 RUNNING
- 线程池状态 随着时间 只能单调增加
- RUNNING -> SHUTDOWN,调用
shutdown() - (RUNNING or SHUTDOWN)-> STOP， 调用
shutdownNow() - SHUTDOWN -> TIDYING ，queue and pool 同时为空
- STOP -> TIDYING，pool 为空
- TIDYING -> TERMINATED，terminated() hook method 执行完毕
任务处理流程
处理任务方法是 execute
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();
// 1.当前线程数小于corePoolSize,则调用addWorker创建核心线程执行任务
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
// 2.如果不小于corePoolSize，则将任务添加到workQueue队列。
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
// 2.1 如果isRunning返回false(状态检查)，则remove这个任务，然后执行拒绝策略。
if (! isRunning(recheck) && remove(command))
reject(command);
// 2.2 线程池处于running状态，但是没有线程，则创建线程
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
// 3.如果放入workQueue失败，则创建非核心线程执行任务，
// 如果这时创建非核心线程失败(当前线程总数不小于maximumPoolSize时)，就会执行拒绝策略。
else if (!addWorker(command, false))
reject(command);
}
- 为什么入队后要二次检查线程池的状态？
因为多线程环境下，线程池状态时刻发生变化。如果没有二次检查，线程池不是 RUNNING，那么 command 将不会执行。
总结一下
- 如果线程数 < corePoolSize，创建新的核心线程执行任务（需要全局锁）
- 如果线程数 >= corePoolSize，新任务进入任务队列等待, 空闲的核心线程会一次从队列获取任务（线程复用）
- 如果缓存队列满了，创建非核心线程执行任务（需要全局锁）
- 如果缓存队列满了，非核心线程达到 maximunPoolSize, 执行拒绝策略
线程复用原理
- 线程池如何做到线程复用的？
ThreadPoolExecutor 创建线程时，会将线程包装成 Worker，并放到工作线程组中，然后这个 worker反复从阻塞队列中获取任务执行。
我们来看 addWorker
{
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// Check if queue empty only if necessary.
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return 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 {
w = new Worker(firstTask);
// 创建 Thread 对象 就是 work 本身
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);
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 {
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);
}
public void run() {
runWorker(this);
}
}
Worker 实现了 Runnable 接口，构造方法中，创建了一个线程，线程任务就是自己.
addWorker中会调用 start方法启动 Worker.
我们来看一下 runWorker,它是个循环，重复从任务队列获取任务并执行
// Worker.runWorker方法源代码
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
// 1.线程启动之后，通过unlock方法释放锁
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
// 2.Worker执行firstTask或从workQueue中获取任务，如果getTask方法不返回null,循环不退出
while (task != null || (task = getTask()) != null) {
// 2.1进行加锁操作，保证thread不被其他线程中断（除非线程池被中断）
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
// 2.2检查线程池状态，倘若线程池处于中断状态，当前线程将中断。
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
// 2.3执行beforeExecute
beforeExecute(wt, task);
Throwable thrown = null;
try {
// 2.4执行任务
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 {
// 2.5执行afterExecute方法
afterExecute(task, thrown);
}
} finally {
task = null;
w.completedTasks++;
// 2.6解锁操作
w.unlock();
}
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
}
}
首先会去执行 firstTask,执行完成后，Worker并不会结束，他会不断的调用getTask方法从阻塞队列获取任务，然后调用 task.run, 从而达到了线程复用的目的。
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 默认false 表示核心线程 永不销毁
// 如果为 ture 核心线程空闲 超时后 会被销毁
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;
}
}
}
默认情况下，核心线程会卡在 workQueue.take方法，不会占用CPU资源（如果 allowCoreThreadTimeOut 为 true，会调用 poll 方法）
非核心线程数会 workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) ，如果超时还没有拿到任务，下一次循环
常见线程池
Executors 提供了静态方法创建线程池。
newCachedThreadPool
- 核心线程数为0，线程池最大数量为 Integer.MAX_VALUE，线程回收时间为60秒。
当有很多短时间的任务时，CacheThreadPool 线程复用率高。
newFixedThreadPool
- 核心线程数=最大线程数，只会创建核心线程
- 使用
LinkedBlockingQueue,所有任务FIFO
newSingleThreadExecutor
- 只有一个核心线程
- 使用
LinkedBlockingQueue,所有任务FIFO
newScheduledThreadPool
- 支持定时及周期执行任务的线程池