共计 13975 个字符，预计需要花费 35 分钟才能阅读完成。
继承关系
Executor 接口
public interface Executor {void execute(Runnable command);
}ExecutorService 接口
public interface ExecutorService extends Executor {void shutdown();
List<Runnable> shutdownNow();
boolean isShutdown();
boolean isTerminated();
boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException;
<T> Future<T> submit(Callable<T> task);
<T> Future<T> submit(Runnable task, T result);
Future<?> submit(Runnable task);
<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;
}ExecutorService 接口继承 Executor 接口，并减少了 submit、shutdown、invokeAll 等等一系列办法。
AbstractExecutorService 抽象类
public abstract class AbstractExecutorService implements ExecutorService {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);
}
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;
}
private <T> T doInvokeAny(Collection<? extends Callable<T>> tasks, boolean timed, long nanos)
throws InterruptedException, ExecutionException, TimeoutException {...}
public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
throws InterruptedException, ExecutionException {...}
public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {...}
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
throws InterruptedException {...}
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException {...}
}AbstractExecutorService 抽象类实现 ExecutorService 接口，并且提供了一些办法的默认实现，例如 submit 办法、invokeAny 办法、invokeAll 办法。
像 execute 办法、线程池的敞开办法（shutdown、shutdownNow 等等）就没有提供默认的实现。
构造函数与线程池状态
public ThreadPoolExecutor(int corePoolSize, // 外围线程数
int maximumPoolSize, // 最大线程数
long keepAliveTime, // 线程存活工夫
TimeUnit unit, //keepAliveTime 的单位
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;
}线程池状态
// 记录线程池状态和线程数量（总共 32 位，前三位示意线程池状态，后 29 位示意线程数量）private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
// 线程数量统计位数 29 Integer.SIZE=32
private static final int COUNT_BITS = Integer.SIZE - 3;
// 容量 000 11111111111111111111111111111
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
// 运行中 111 00000000000000000000000000000
private static final int RUNNING = -1 << COUNT_BITS;
// 敞开 000 00000000000000000000000000000
private static final int SHUTDOWN = 0 << COUNT_BITS;
// 进行 001 00000000000000000000000000000
private static final int STOP = 1 << COUNT_BITS;
// 整顿 010 00000000000000000000000000000
private static final int TIDYING = 2 << COUNT_BITS;
// 终止 011 00000000000000000000000000000
private static final int TERMINATED = 3 << COUNT_BITS;
// 获取运行状态（获取前 3 位）private static int runStateOf(int c) {return c & ~CAPACITY;}
// 获取线程个数（获取后 29 位）private static int workerCountOf(int c) {return c & CAPACITY;}
private static int ctlOf(int rs, int wc) {return rs | wc;}int 是 4 个字节,32 位
RUNNING：承受新工作并且解决阻塞队列里的工作
SHUTDOWN：回绝新工作然而解决阻塞队列里的工作
STOP：回绝新工作并且摈弃阻塞队列里的工作同时会中断正在解决的工作
TIDYING：所有工作都执行完（蕴含阻塞队列外面工作），以后线程池流动线程为 0，将要调用 terminated 办法
TERMINATED：终止状态。terminated 办法调用实现当前的状态
线程池状态转换：RUNNING -> SHUTDOWN：显式调用 shutdown()办法, 或者隐式调用了 finalize()办法
(RUNNING or SHUTDOWN) -> STOP：显式调用 shutdownNow()办法
SHUTDOWN -> TIDYING：当线程池和工作队列都为空的时候
STOP -> TIDYING：当线程池为空的时候
TIDYING -> TERMINATED：当 terminated() hook 办法执行实现时候submit 办法和 execute 办法的区别
submit 办法
- 调用 submit 办法，传入 Runnable 或者 Callable 对象
- 判断传入的对象是否为 null，为 null 则抛出异样，不为 null 持续流程
- 将传入的对象转换为 RunnableFuture 对象
- 执行 execute 办法，传入 RunnableFuture 对象
- 返回 RunnableFuture 对象
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;
}execute 办法
public void execute(Runnable command) {
// 传进来的线程为 null，则抛出空指针异样
if (command == null)
throw new NullPointerException();
// 获取以后线程池的状态 + 线程个数变量
int c = ctl.get();
/**
* 3 个步骤
*/
//1. 判断以后线程池线程个数是否小于 corePoolSize, 小于则调用 addWorker 办法创立新线程运行,
// 且传进来的 Runnable 当做第一个工作执行。// 如果调用 addWorker 办法返回 false，则间接返回
if (workerCountOf(c) < corePoolSize) {// 增加一个 core 线程(外围线程)。此处参数的 true，示意增加的线程是 core 容量下的线程
if (addWorker(command, true))
return;
// 刷新数据，乐观锁就是没有锁
c = ctl.get();}
/* isRunning 办法的定义：private static boolean isRunning(int c)
{return c < SHUTDOWN;}
2.SHUTDOWN 值为 0，即如果 c 小于 0，示意在运行；offer 用来判断工作是否胜利入队 */
if (isRunning(c) && workQueue.offer(command)) {
// 二次查看
int recheck = ctl.get();
// 如果以后线程池状态不是 RUNNING 则从队列删除工作，并执行回绝策略
if (! isRunning(recheck) && remove(command))
// 执行回绝策略
reject(command);
// 否则如果以后线程池线程空，则增加一个线程
else if (workerCountOf(recheck) == 0)
// 增加一个空线程进线程池，应用非 core 容量线程
// 仅有一种状况，会走这步，core 线程数为 0，max 线程数 >0, 队列容量 >0
// 创立一个非 core 容量的线程，线程池会将队列的 command 执行
addWorker(null, false);
}
// 线程池进行了或者队列已满，增加 maximumPoolSize 容量工作线程，如果失败，执行回绝策略
else if (!addWorker(command, false))
reject(command);
}ThreadPoolExecutor.addWorker()
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {int c = ctl.get(); // 获取运行状态和工作数量
int rs = runStateOf(c); // 获取以后线程池运行的状态
// Check if queue empty only if necessary.
// 条件代表着以下几个场景，间接返回 false 阐明当前工作线程创立失败
//1.rs>SHUTDOWN 此时不再接管新工作，且所有的工作曾经执行结束
//2.rs=SHUTDOWN 此时不再接管新工作，然而会执行队列中的工作
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
for (;;) {int wc = workerCountOf(c);
// 先判断以后流动的线程数是否大于最大值，如果超过了就间接返回 false 阐明线程创立失败
// 如果没有超过再依据 core 的值再进行以下判断
//1. core 为 true，则判断以后流动的线程数是否大于 corePoolSize
//2. core 为 false，则判断以后流动线程数是否大于 maximumPoolSize
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
// 比拟以后值是否和 c 雷同，如果雷同，则改为 c +1，并且跳出大循环，间接执行 Worker 进行线程创立
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get(); // Re-read ctl
// 查看下以后线程池的状态是否曾经产生扭转
// 如果曾经扭转了，则进行外层 retry 大循环，否则只进行内层的循环
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 的也是 Runnable 的实现类
w = new Worker(firstTask);
// 因为不能够间接在 Worker 的构造方法中进行线程创立
// 所以要把它的援用赋给 t 不便前面进行线程创立
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);// 将创立的线程增加到 workers 容器中
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;
}Worker 办法
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;
Worker(Runnable firstTask) {setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
}Worker 在 ThreadPoolExecutor 为一个外部类实现了 Runnable 接口。只有一个构造方法，在下面的 addWorker()中 final Thread t = w.thread; 晓得其实是获取了线程的对象，因为在构造方法中，线程的援用即是它本人。
因而在调用 t.start()执行的是（Worker 类中的办法）：
/** Delegates main run loop to outer runWorker */
public void run() {
// 这里执行的是 ThreadPoolExecutor 中的 runWorker
runWorker(this);
}ThreadPoolExecutor.runWorker()
final void runWorker(Worker w) {Thread wt = Thread.currentThread();
Runnable task = w.firstTask;// 获取 Worker 中的工作
w.firstTask = null; // 将 Woeker 中的工作置空
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
// 如果当前任务为空 那么就从 getTask 中取得工作
/**
* 如果 task 不为空，执行完 task 后则将 task 置空
* 持续进入循环，则从 getTask 中获取工作
*/
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;
} finally {processWorkerExit(w, completedAbruptly);
}
}从下面能够简略了解，就是执行工作，只是执行工作须要进行解决，包含取得工作、工作开始前解决、工作执行、工作执行后处理。然而，要害代码还是外面所调用的一个办法 getTask()。
beforeExecute(Thread t, Runnable r)与afterExecute(Runnable r, Throwable t)并未在类中有解决业务的逻辑，即能够通过继承线程池的形式来重写这两个办法，这样就可能对工作的执行进行监控。
processWorkerExit
- 从 While 循环体中能够晓得，当线程运行时出现异常，那么都会退出循环，进入到 processWorkerExit()
- 从 getTask()取得后果为 null，则也会进到 processWorkerExit()
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(true)
// 或者以后运行的工作数大于设置的外围线程数
// timed = true
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
/** ------------------------ 以上的操作跟之前相似 ----------------------- */
/** ------------------------ 关键在于上面的代码 ------------------------- */
/** ------------------------ 从阻塞队列中获取工作 ----------------------- */
try {
Runnable r = timed ?
// 对于阻塞队列，poll(long timeout, TimeUnit unit) 将会在规定的工夫内去工作
// 如果没取到就返回 null
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
//take 会始终阻塞，期待工作的增加
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {timedOut = false;}
}
}线程池可能保障始终期待工作而不被销毁，其实就是进入了阻塞状态
ThreadPoolExecutor.processWorkerExit()
/**
* @param completedAbruptly
*/
private void processWorkerExit(Worker w, boolean completedAbruptly) {if (completedAbruptly) // 如果忽然被打断，工作线程数不会被缩小
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)) {if (!completedAbruptly) {// 失常退出，也就是 task == null
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
if (min == 0 && ! workQueue.isEmpty())
min = 1;
if (workerCountOf(c) >= min)
return; // replacement not needed
}
// 新增一个工作线程，代替原来的工作线程
addWorker(null, false);
}
}线程池敞开
能够通过调用线程池的 shutdown 或 shutdownNow 办法来敞开线程池。它们的原理是遍历线程池中的工作线程，而后一一调用线程的 interrupt 办法来中断线程，所以无奈响应中断的工作可能永远无奈终止。然而它们存在肯定的区别，shutdownNow 首先将线程池的状态设置成 STOP，而后尝试进行所有的正在执行或暂停工作的线程，并返回期待执行工作的列表，而 shutdown 只是将线程池的状态设置成 SHUTDOWN 状态，而后中断所有没有正在执行工作的线程。
只有调用了这两个敞开办法中的任意一个，isShutdown 办法就会返回 true。当所有的工作都已敞开后，才示意线程池敞开胜利，这时调用 isTerminaed 办法会返回 true。至于应该调用哪一种办法来敞开线程池，应该由提交到线程池的工作个性决定，通常调用 shutdown 办法来敞开线程池，如果工作不肯定要执行完，则能够调用 shutdownNow 办法。
shutdown
当调用 shutdown 办法时，线程池将不会再接管新的工作，而后将先前放在队列中的工作执行实现。
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 查看权限
checkShutdownAccess();
//CAS 更新线程池状态
advanceRunState(SHUTDOWN);
// 中断所有闲暇的线程
interruptIdleWorkers();
// 敞开，此处是 do nothing
onShutdown();} finally {mainLock.unlock();
}
// 尝试完结，下面代码已剖析
tryTerminate();}
shutdownNow
立刻进行所有的执行工作，并将队列中的工作返回
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;
}总结
- 线程池优先应用 corePoolSize 的数量执行工作工作
- 如果超过 corePoolSize，队列入队
- 超过队列，应用 maximumPoolSize-corePoolSize 的线程解决，这部分线程超时不干活就销毁掉。
- 每个线程执行完结的时候，会判断以后的工作线程和工作数，如果工作数多，就会创立空线程从队列拿工作。
- 线程池执行实现，不会主动销毁，须要手工 shutdown，批改线程池状态，中断所有线程。
调配线程池大小的根据
从以下几个角度思考
- 工作的性质：CPU 密集型工作、IO 密集型工作和混合型工作。
- 工作的优先级：高、中和低。
- 工作的执行工夫：长、中和短。
- 工作的依赖性：是否依赖其余系统资源，如数据库连贯。
性质不同的工作能够用不同规模的线程池离开解决。CPU 密集型工作应配置尽可能小的线程，如配置 cpu 个数 + 1 个线程的线程池。因为 IO 密集型工作线程并不是始终在执行工作，则应配置尽可能多的线程，如 2 *cpu 个数。混合型的工作，如果能够拆分，将其拆分成一个 CPU 密集型工作和一个 IO 密集型工作，只有这两个工作执行的工夫相差不是太大，那么合成后执行的吞吐量 将高于串行执行的吞吐量。如果这两个工作执行工夫相差太大，则没必要进行合成。能够通过 Runtime.getRuntime().availableProcessors()办法取得以后设施的 CPU 个数。优先级不同的工作能够应用优先级队列 PriorityBlockingQueue 来解决。它能够让优先级高的工作先执行。
执行工夫不同的工作能够交给不同规模的线程池来解决，或者能够应用优先级队列，让执行工夫短的工作先执行。
依赖数据库连接池的工作，因为线程提交 SQL 后须要期待数据库返回后果，期待的工夫越长，则 CPU 闲暇工夫就越长，那么线程数应该设置得越大，这样能力更好地利用 CPU。
应用有界队列
有界队列能减少零碎的稳定性和预警能力，能够依据须要设大一点儿，比方几千。有一次，咱们零碎里后台任务线程池的队列和线程池全满了，一直抛出摈弃工作的异样，通过排查发现是数据库呈现了问题，导致执行 SQL 变得十分迟缓，因为后台任务线程池里的工作全是须要向数据库查问和插入数据的，所以导致线程池里的工作线程全副阻塞，工作积压在线程池里。如果过后咱们设置成无界队列，那么线程池的队列就会越来越多，有可能会撑满内存，导致整个零碎不可用，而不只是后台任务呈现问题。当然，咱们的零碎所有的工作是用独自的服务器部署的，咱们应用不同规模的线程池实现不同类型的工作，然而呈现这样问题时也会影响到其余工作。
线程池监控
如果在零碎中大量应用线程池，则有必要对线程池进行监控，不便在呈现问题时，能够依据线程池的应用情况疾速定位问题。能够通过线程池提供的参数进行监控，在监控线程池的时候能够应用以下属性。
- taskCount：线程池须要执行的工作数量。
- completedTaskCount：线程池在运行过程中已实现的工作数量，小于或等于 taskCount。
- largestPoolSize：线程池里已经创立过的最大线程数量。通过这个数据能够晓得线程池是否已经满过。如该数值等于线程池的最大大小，则示意线程池已经满过。
- getPoolSize：线程池的线程数量。如果线程池不销毁的话，线程池里的线程不会主动销毁，所以这个大小只增不减。
- getActiveCount：获取流动的线程数。
- 通过扩大线程池进行监控。能够通过继承线程池来自定义线程池，重写线程池的 beforeExecute、afterExecute 和 terminated 办法，也能够在工作执行前、执行后和线程池敞开前执行一些代码来进行监控。例如，监控工作的均匀执行工夫、最大执行工夫和最小执行工夫等。这几个办法在线程池里是空办法。
关注微信公众号:【入门小站】, 解锁更多知识点
正文完
