Java Thread Pool
使用线程池的好处：
- 减少资源的浪费：创建、销毁、切换线程需要消耗系统资源，通过使用线程池可以降低消耗。
- 增加可管理度：通过线程池的同一管理，能够实现线程的更好的管理。
- 提高相应速度：当任务到来时，无需在创建线程，直接就能对任务进行反馈
Java线程池的使用
线程池距离实现类public class ThreadPoolExecutor extends AbstractExecutorService,继承了AbstractExecutorService这个类。
线程池的状态
线程池有核心线程和普通线程的区分，并且线程池有自己的状态。这两个是通过一个AtomicInteger类实现多线程的安全，并且通过一个整数实现状态和线程个数的整合。
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); // ctl表示了状态和线程的个数
private static final int COUNT_BITS = Integer.SIZE - 3; // 32 - 3 = 29 高三位表示状态 低二十九位表示线程个数
private static final int COUNT_MASK = (1 << COUNT_BITS) - 1;
// 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;
private static final int TIDYING = 2 << COUNT_BITS;
private static final int TERMINATED = 3 << COUNT_BITS;
提取线程池状态和线程的多少可以同过线程池的方法进行提取：
// Packing and unpacking ctl
private static int runStateOf(int c) { return c & ~COUNT_MASK; } // 运行在那个状态
private static int workerCountOf(int c) { return c & COUNT_MASK; } // 线程的数量
private static int ctlOf(int rs, int wc) { return rs | wc; } // 状态和线程个数的综合，用来设置的
/** 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;
}
/*** Attempts to CAS-increment the workerCount field of ctl.*/
private boolean compareAndIncrementWorkerCount(int expect) { // CASE设置ctl的线程数量加1return ctl.compareAndSet(expect, expect + 1);
}
/*** Attempts to CAS-decrement the workerCount field of ctl.*/
private boolean compareAndDecrementWorkerCount(int expect) { // CAS设置ctl线程数量减1return 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() { // 直接进行原子的减一， 只用在abrubt termination of a threadctl.addAndGet(-1);
}
简单流程
提交任务
现成池提交任务有两个类型，一个是实现了Runnable接口，一个是实现了Callable接口，两个都是用于线程运行的任务。Runnable是没有返回值的，Callable是由返回值的，并且Callable可以抛出异常，Runnable不可以抛出异常。
向线程池提交任务，可以通过ThreadPoolExecutor.submit([Runnable] | [Runnable, Value] | [Callable])进行提交任务。三个重载方法代码如下(源自于ThreadPoolExecutor的父类AbstractExecutrService中的方法)
这种方法主要是提交Callable任务和查看任务有没有完成的。
// submit(Runnable)
public Future<?> submit(Runnable task) {if (task == null) throw new NullPointerException();RunnableFuture<Void> ftask = newTaskFor(task, null);execute(ftask);return ftask;
}
// submit(Runnable, Value)
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;
}
// submit(Callable)
public <T> Future<T> submit(Callable<T> task) {if (task == null) throw new NullPointerException();RunnableFuture<T> ftask = newTaskFor(task);execute(ftask);return ftask;
}
三个方法的整体流程都相同，将Runnable或者Callable送入的任务进行封装，封装为一个RunnableFuture接口的任务,实际的类型就是FutureMask任务。FutureTask中有一个Callable，送入的是Callable无需转换，送入的是Runnable需要通过RunnableAdapter进行转换为Callable类。
private static final class RunnableAdapter<T> implements Callable<T> {private final Runnable task;private final T result;RunnableAdapter(Runnable task, T result) {this.task = task;this.result = result;}public T call() {task.run();return result;}public String toString() {return super.toString() + "[Wrapped task = " + task + "]";}
}
直接提交Runnable任务，可以通过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();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);
}
最终任务的转换流程：
最终会将一个Runnable对象送入execute方法。execute方法是通过ThreadPoolExector进行实现的。
线程池参数
public ThreadPoolExecutor(int corePoolSize, // 核心线程数int maximumPoolSize, // 最大线程数long keepAliveTime, // 存活时间TimeUnit unit, // keepAliveTime的单位BlockingQueue<Runnable> workQueue, // 任务队列ThreadFactory threadFactory, // 线程工厂RejectedExecutionHandler handler) { // 拒绝策略
}
详细分析
不论是通过父类submit还是通过子类ThreadPoolExecutor的execute进行提交，最终的提交都是通过execute进行执行。直接分析这个方法。
主要的任务流程和上述简单分析的一样。分为四个部分。
int c = ctl.get() // 表示为当前线程池的状态
添加核心线程并执行
if (workerCountOf(c) < corePoolSize) {if (addWorker(command, true))return;c = ctl.get();
}
首先判断线程池的个数是否大于了core_size,小于的话就可以执行add_worker方法。
// firstTask 送入的任务 在执行添加coor_thread线程中是提交的任务
//(由于是Runnable接口，所以无法获得这个的执行结果)
//boolean core表示是否添加核心线程 ture
private boolean addWorker(Runnable firstTask, boolean core) {retry:for (int c = ctl.get();;) { // 重新获得当前线程的状态 recheck// Check if queue empty only if necessary.if (runStateAtLeast(c, SHUTDOWN) // 如果当前线程至少运行在shutdown并且//(已经是stop或者送入的任务为空或者任务队列为空) 就直接退出 返回添加失败// 这也就是说 STOP是无论如何都不会接受任务，SHUTDOWN实在workQueue为空是拒绝任务&& (runStateAtLeast(c, STOP)|| firstTask != null|| workQueue.isEmpty()))return false;// 无限循环增加线程数 先增加线程数改变状态并且之后再进行实际的线程创建for (;;) {if (workerCountOf(c)>= ((core ? corePoolSize : maximumPoolSize) & COUNT_MASK))// 满不满足核心和非核心的线程数的要求return false;if (compareAndIncrementWorkerCount(c))// 增加线程数，成功退出循环break retry;c = ctl.get(); // Re-read ctl// 添加线程数失败，有其他线程修改了 重新获取if (runStateAtLeast(c, SHUTDOWN))// 再次比较 表示state也修改了 直接返回最开始continue retry;// else CAS failed due to workerCount change; retry inner loop// 表示workerCount修改了，重新执行for循环}}// 成功修改了状态之后 要开始实际的创建线程boolean workerStarted = false; // worker开始工作flagboolean workerAdded = false; // worker添加成功flagWorker w = null; // try {w = new Worker(firstTask); // 根据当前的任务创建worker ///worker是一个实现了AbstractQueueSynchronized和Runnable的一个工作者// 里面通过经典AQS的状态进行锁的管理，因为这个worker可能被不同的线程启动，所以有一个内部的锁// 里面还有一个变量thread，是通过ThreadFactory创建的线程，也表示将要执行任务的线程,// 一个firstTask，表示这个worker的任务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 c = ctl.get(); // recheckif (isRunning(c) || // 正处于Running状态或者比STOP小于的状态，且任务不为空，那么可以继续执行/// 和之前那个execute的区别是，如果处于SHUTDOWN状态，那么当任务队列为空的时候就拒绝任务了(runStateLessThan(c, STOP) && firstTask == null)) {if (t.getState() != Thread.State.NEW) // 线程状态不为NEW，直接报错throw new IllegalThreadStateException();workers.add(w); // 一个hashset，添加创建的workerworkerAdded = true; // 表示在workers中添加worker成功int s = workers.size();if (s > largestPoolSize)largestPoolSize = s;}} finally {mainLock.unlock();}if (workerAdded) { // 添加成功之后进行运行workert.start(); // 线程直接启动workerStarted = true; // 启动成功的标志}}} finally {if (! workerStarted) // 如果没有启动成功，addWorkerFailed(w);}return workerStarted;
}
上述代码有几个岔路，---- `if` isRunning || LessThan(Stop)-----| || || |->hashset(workers)添加worker成功 更新largerPoolSize| ||-> `else` 直接返回 | | |-->start()成功||->执行addWorkerFailed()
启动成功执行start()方法
众所周知，start方法会执行run方法，那么是谁的Run方法呢？具体的是worker的run方法，
public void run() {runWorker(this);
}
/*** 主工作循环流程。重复的从队列中获取任务并且执行任务。* 在这个过程中，主要处理三个问题(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. * 有初始任务就对任务进行执行，没有任务就从getTask获取任务。当getTask返回null的时候，worker会让他退出，由于pool state的改变或这一些配置的参数的原因。* 如果因为exception而导致了一场，会有boolean completedAbruptly进行捕获，通过processWorkerExit进行处理，替换掉这个线程。
*/
final void runWorker(Worker w) {Thread wt = Thread.currentThread(); // 执行worker的线程，实际上也就是worker中存储的线程，不太清楚通过w.thread和通过Thread.currentThread()这两个方法有什么区别Runnable task = w.firstTask; // 运行的任务 提取出来w.firstTask = null; // 将w的任务设为nullw.unlock(); // allow interrupts // 解锁 实际上并没有人获得锁boolean completedAbruptly = true; // 是否是突然完成的 突然完成会执行这个方法 decrementWorkerCounttry {while (task != null || (task = getTask()) != null) { // 获得任务，送入的task或者通过getTask()进行获得任务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 interruptif ((runStateAtLeast(ctl.get(), STOP) || // STOP状态 或者 线程已经被打断(interrupted()会清楚打断标记)并且重新获取状态状态为STOP状态，// 如果当前线程没有被打断,那么直接进行打断，(Thread.interrupted() &&runStateAtLeast(ctl.get(), STOP))) &&!wt.isInterrupted())wt.interrupt();try {beforeExecute(wt, task); //hooktry {task.run(); // 执行任务的run方法 这个方法是不抛出异常的afterExecute(task, null); // hook} catch (Throwable ex) {afterExecute(task, ex); // hookthrow ex; // 抛出异常}} finally { // 无论是否正常完成，这个任务直接被清空，该worker完成的任务被加一task = null;w.completedTasks++;w.unlock();}}completedAbruptly = false;// 已经没有任务可以完成了，就会设置为false,表示正常完成，如果抛出异常，会设置为true} finally {processWorkerExit(w, completedAbruptly); // 某一个任务执行是抛出了异常或者任务执行时}
}
getTask方法保证了线程池状态的准确性
private Runnable getTask() {// getTask表示一个线程完成任务，并且在等待boolean timedOut = false; // Did the last poll() time out?for (;;) {int c = ctl.get(); // 表示ctl的值// Check if queue empty only if necessary.if (runStateAtLeast(c, SHUTDOWN) // 运行在shutdwon阶段切运行在stop阶段，不再接受任务并执行了，所以worker获取任务，直接返回null并且减少workerCount的数量 也就对应了processExitWorker的中不是异常退出就不减，因为之间在getTask中剪过了&& (runStateAtLeast(c, STOP) || workQueue.isEmpty())) {decrementWorkerCount();return null;}int wc = workerCountOf(c);// Are workers subject to culling?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;}}
}
通过消除引用实现线程的销毁
private void processWorkerExit(Worker w, boolean completedAbruptly) {if (completedAbruptly) // If abrupt, then workerCount wasn't adjusteddecrementWorkerCount(); // 由于异常中断了线程，那么直接进行workerCount的减少，final ReentrantLock mainLock = this.mainLock; // 主锁mainLock.lock(); // 获得锁try {completedTaskCount += w.completedTasks; // 整个线程池的完成的任务数量workers.remove(w); // 从worker中进行移除 worker从wokers中移除 这里remove了，但是并没有减少workerCount 认为在执行方法已经处理过了workerCount,因为在getTask中保证了线程池状态的完整性的完整性} finally {mainLock.unlock(); // 释放锁}tryTerminate(); // 尝试中断，将线程池的状态设为terminate，如果没有worker了，就和设置为terminatedint c = ctl.get(); // 获得状态和线程数if (runStateLessThan(c, STOP)) { // 查看线程状态小于STOP 需要将这个线程进行替换 STOP直接只进行减少，不进行增加if (!completedAbruptly) { //没有被异常中断 当前线程的数量和最小值怎么样，如果小于最小值，那么添加一个worker，// 如果大于最小值，那么就退出 保持了线程数量的稳定int min = allowCoreThreadTimeOut ? 0 : corePoolSize; // 只会保持核心线程数，如果设置为coreTimeOut,那么核心线程也会撤销if (min == 0 && ! workQueue.isEmpty())min = 1;if (workerCountOf(c) >= min)return; // replacement not needed}addWorker(null, false); // 被异常中断了线程 那么会添加一个新的线程，添加worker。如果没有任务了,通过一直创建线程维持core的线程数量。如果没有设置coreTimeOut任务}
}
final void tryTerminate() {for (;;) {int c = ctl.get();if (isRunning(c) ||runStateAtLeast(c, TIDYING) ||(runStateLessThan(c, STOP) && ! workQueue.isEmpty()))return; // 正在运行或者比tidying更高的state或者运行在shutdonw且在任务不为空 无法中断 退出// 运行在SHUTDOWN或者STOP这两个阶段 SHUTDOWN阶段任务列表不为空if (workerCountOf(c) != 0) { // Eligible to terminate 还有worker在进行工作 打断一个线程 interruptIdleWorkers(ONLY_ONE); // 打断一个workerreturn;}// STOP或者SHUTDOWN阶段，work已经什么都没有了final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) { //先设置为TIDYING ,0try {terminated(); } finally {ctl.set(ctlOf(TERMINATED, 0)); // 在最终设置为TERMINATED 0termination.signalAll();}return;}} finally {mainLock.unlock();}// else retry on failed CAS}
}
private void interruptIdleWorkers(boolean onlyOne) {// 运行在STOP或者SHUTDOWN stagefinal ReentrantLock mainLock = this.mainLock;mainLock.lock();try {for (Worker w : workers) {Thread t = w.thread;if (!t.isInterrupted() && w.tryLock()) { // 没有被打断并且没有记性锁定的worker，能够获取锁，表示没有在进行任务的执行try {t.interrupt();} catch (SecurityException ignore) {} finally {w.unlock();}}if (onlyOne)break;}} finally {mainLock.unlock();}
}
向任务队列中添加任务
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);
}
调用offer方法添加任务，重新检查ctl状态，不再运行，直接不接受任务，拒绝。查看workerCountOf为0，添加一个线程
添加非核心线程执行
else if (!addWorker(command, false))
抛弃任务，交由自定义的任务处理策略进行处理
reject(command);
![[深入理解Java虚拟机]高效并发](./%E5%9B%BE%E7%89%87%E9%99%84%E4%BB%B6/12-2.png)
