共计 29848 个字符，预计需要花费 75 分钟才能阅读完成。
接上节内容
ThreadPool
在多线程开发中，如果并发的申请数量十分多，但每个线程执行的工夫很短，这样就会频繁的创立和销毁线程，如此一来就会大大降低零碎的效率。可能呈现服务器在为每个申请创立新线程和销毁线程上破费的工夫和耗费的系统资源要比解决理论的用户申请的工夫和资源更多。为了复用线程缩小资源节约，咱们就须要应用到线程池。
线程池提供了一种限度和治理线程资源的计划。通过线程池咱们能够达到如下目标：
- 升高资源耗费。通过反复利用已创立的线程升高线程创立和销毁造成的耗费。
- 进步响应速度。当工作达到时，工作能够不须要的等到线程创立就能立刻执行。
- 进步线程的可管理性。线程是稀缺资源，如果无限度的创立，不仅会耗费系统资源，还会升高零碎的稳定性，应用线程池能够进行对立的调配，调优和监控。
JUC 中有三个 Executor 接口：
Executor：一个运行新工作的简略接口
ExecutorService：扩大了 Executor 接口。增加了 Future 性能和一些用来治理执行器生命周期和工作生命周期的办法。
ScheduledExecutorService：扩大了 ExecutorService。反对定期执行工作。
JUC 中的 ThreadPoolExecutor 实现了 ExecutorService，也是大家最罕用到的线程池实现。咱们接下来就以它为例介绍线程池。
应用形式
在创立一个线程池时，须要几个外围参数，一个是 corePoolSize 它形容该线程池中常驻的线程池数量，其次是 maximumPoolSize 它形容该线程池中最多能有多少个线程。当线程池的线程数大于 corePoolSize 时，keepAliveTime 规定了那些多余的闲暇线程最多能存活多久。workQueue 是一个寄存工作的队列，用于缓存将要执行的工作。threadFactory 用于创立新的线程。
/**
* Creates a new {@code ThreadPoolExecutor} with the given initial * parameters. * * @param corePoolSize the number of threads to keep in the pool, even * if they are idle, unless {@code allowCoreThreadTimeOut} is set * @param maximumPoolSize the maximum number of threads to allow in the * pool * @param keepAliveTime when the number of threads is greater than * the core, this is the maximum time that excess idle threads * will wait for new tasks before terminating. * @param unit the time unit for the {@code keepAliveTime} argument * @param workQueue the queue to use for holding tasks before they are * executed. This queue will hold only the {@code Runnable} * tasks submitted by the {@code execute} method. * @param threadFactory the factory to use when the executor * creates a new thread * @param handler the handler to use when execution is blocked * because the thread bounds and queue capacities are reached * @throws IllegalArgumentException if one of the following holds:<br> * {@code corePoolSize < 0}<br> * {@code keepAliveTime < 0}<br> * {@code maximumPoolSize <= 0}<br> * {@code maximumPoolSize < corePoolSize} * @throws NullPointerException if {@code workQueue} * or {@code threadFactory} or {@code handler} is null */
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.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}RejectedExecutionHandler 用来形容当工作队列满了并且以后线程池内的线程数达到 maximumPoolSize 时，应该如何回应新退出的工作，ThreadPoolTaskExecutor 定义一些罕用的策略：
- ThreadPoolExecutor. AbortPolicy：抛出 RejectedExecutionException 来回绝新工作的解决。
- ThreadPoolExecutor. CallerRunsPolicy：调用执行本人的线程运行工作。您不会工作申请。然而这种策略会升高对于新工作提交速度，影响程序的整体性能。如果您的应用程序能够接受此提早并且你不能抛弃任何一个工作申请的话，你能够抉择这个策略。
- ThreadPoolExecutor. DiscardPolicy：不解决新工作，间接抛弃掉。
- ThreadPoolExecutor. DiscardOldestPolicy：此策略将抛弃最早的未解决的工作申请。
如果想让线程池执行工作的话须要实现的 Runnable 接口或 Callable 接口。Runnable 接口或者 Callable 接口实现类都能够被 ThreadPoolExecutor 或 ScheduledThreadPoolExecutor 执行。两者的区别在于 Runnable 接口不会返回后果然而 Callable 接口能够返回后果。
接下来咱们以 ThreadPoolExecutor 为例介绍一下线程池的罕用接口。
/**
* Executes the given task sometime in the future. The task * may execute in a new thread or in an existing pooled thread. * * If the task cannot be submitted for execution, either because this * executor has been shutdown or because its capacity has been reached, * the task is handled by the current {@code RejectedExecutionHandler}. * * @param command the task to execute * @throws RejectedExecutionException at discretion of * {@code RejectedExecutionHandler}, if the task * cannot be accepted for execution * @throws NullPointerException if {@code command} is null */
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) {// 线程数小于外围线程数时工作不入队，间接通过参数传递到 worker 线程中 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);
}
// 入队失败时，当前任务也是参数传递到 worker 线程中 else if (!addWorker(command, false))
reject(command);
}Runnable 接口次要是配合线程池的 execute 应用，用于提交不须要返回值的工作，所以无奈判断工作是否被线程池执行胜利与否。它的执行流程次要分 3 步：
- 如果当期线程数小于 corePoolSize，创立新的线程去执行该工作
- 如果线程数达到了 corePoolSize，则试图将工作增加到工作队列中
- 如果队列曾经满了，则增加新的线程，直到达到 maximumPoolSize，这时候就采纳后面所说的解决策略 RejectedExecutionHandler 来解决这些新工作。
提交工作除了应用 execute 办法之外，还能够应用 submit 办法，该办法用于提交须要返回值的工作线程池会返回一个 Future 类型的对象，通过这个 Future 对象能够判断工作是否执行胜利，并且能够通过 future 的 get() 办法来获取返回值，get() 办法会阻塞以后线程直到工作实现，而应用 get(long timeout,TimeUnit unit) 办法则会阻塞以后线程一段时间后立刻返回，这时候有可能工作没有执行完。
/**
* @throws RejectedExecutionException {@inheritDoc} * @throws NullPointerException {@inheritDoc} */
public <T> Future<T> submit(Callable<T> task) {if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task);
execute(ftask);
return ftask;
}在应用线程池时，举荐应用 ThreadPoolExecutor，并且须要依据本人的应用场景，正当的调整 corePoolSize，maximumPoolSize，workQueue，RejectedExecutionhandler，躲避资源耗尽的危险。
工作性质不同的工作能够用不同规模的线程池离开解决。CPU 密集型工作配置尽可能少的线程数量，如配置 Ncpu+ 1 个线程的线程池。IO 密集型工作则须要期待 IO 操作，线程并不是始终在执行工作，则配置尽可能多的线程，如 2xNcpu。混合型的工作，如果能够拆分，则将其拆分成一个 CPU 密集型工作和一个 IO 密集型工作，只有这两个工作执行的工夫相差不是太大，那么合成后执行的吞吐率要高于串行执行的吞吐率，如果这两个工作执行工夫相差太大，则没有必要进行合成。咱们能够通过 Runtime.getRuntime().avaliableProcessors() 办法取得以后设施的 CPU 个数。
优先级不同的工作能够应用优先级队列 PriorityBlockingQueue 来解决。它能够让优先级高的工作先失去执行，须要留神的是如果始终有优先级高的工作提交到队列里，那么优先级低的工作可能永远不能执行。
执行工夫不同的工作能够交给不同规模的线程池来解决，或者也能够应用优先级队列，让执行工夫短的工作先执行。
依赖数据库连接池的工作，因为线程提交 SQL 后须要期待数据库返回后果，如果期待的工夫越长 CPU 闲暇工夫就越长，那么线程数应该设置越大，这样能力更好的利用 CPU。并且，阻塞队列最好是应用有界队列，如果采纳无界队列的话，一旦工作积压在阻塞队列中的话就会占用过多的内存资源，甚至会使得零碎解体。
当咱们要敞开线程池时，能够通过 shutdown 和 shutdownNow 这两个办法。它们的原理都是遍历线程池中所有的线程，而后顺次中断线程。shutdown 和 shutdownNow 还是有不一样的中央：
- shutdownNow 首先将线程池的状态设置为 STOP, 而后尝试进行所有的正在执行和未执行工作的线程，并返回期待执行工作的列表
- shutdown 只是将线程池的状态设置为 SHUTDOWN 状态（这意味着不再承受新的工作），而后中断所有闲暇的线程，等所有现存工作执行完之后才会销毁线程池
能够看出 shutdown 办法会将正在执行的工作继续执行完，而 shutdownNow 会间接中断正在执行的工作。调用了这两个办法的任意一个，isShutdown 办法都会返回 true，当所有的线程都敞开胜利，才示意线程池胜利敞开，这时调用 isTerminated 办法才会返回 true。
实现形式
在介绍 ThreadPoolExecutor 的实现时，咱们着重介绍它的 execute 函数和 shutdown，shutdownNow，在介绍之前，让咱们来看一看 ThreadPoolExecutor 是如何保护外部数据的。
/**
* The main pool control state, ctl, is an atomic integer packing * two conceptual fields * workerCount, indicating the effective number of threads * runState, indicating whether running, shutting down etc * * In order to pack them into one int, we limit workerCount to * (2^29)-1 (about 500 million) threads rather than (2^31)-1 (2 * billion) otherwise representable. If this is ever an issue in * the future, the variable can be changed to be an AtomicLong, * and the shift/mask constants below adjusted. But until the need * arises, this code is a bit faster and simpler using an int. * * The workerCount is the number of workers that have been * permitted to start and not permitted to stop. The value may be * transiently different from the actual number of live threads, * for example when a ThreadFactory fails to create a thread when * asked, and when exiting threads are still performing * bookkeeping before terminating. The user-visible pool size is * reported as the current size of the workers set. * * The runState provides the main lifecycle control, taking on values: * * RUNNING: Accept new tasks and process queued tasks * SHUTDOWN: Don't accept new tasks, but process queued tasks * STOP: Don't accept new tasks, don't process queued tasks, * and interrupt in-progress tasks * TIDYING: All tasks have terminated, workerCount is zero, * the thread transitioning to state TIDYING * will run the terminated() hook method * TERMINATED: terminated() has completed * * The numerical order among these values matters, to allow * ordered comparisons. The runState monotonically increases over * time, but need not hit each state. The transitions are: * * RUNNING -> SHUTDOWN * On invocation of shutdown(), perhaps implicitly in finalize() * (RUNNING or SHUTDOWN) -> STOP * On invocation of shutdownNow() * SHUTDOWN -> TIDYING * When both queue and pool are empty * STOP -> TIDYING * When pool is empty * TIDYING -> TERMINATED * When the terminated() hook method has completed * * Threads waiting in awaitTermination() will return when the * state reaches TERMINATED. * * Detecting the transition from SHUTDOWN to TIDYING is less * straightforward than you'd like because the queue may become * empty after non-empty and vice versa during SHUTDOWN state, but * we can only terminate if, after seeing that it is empty, we see * that workerCount is 0 (which sometimes entails a recheck -- see * below). */
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 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 & ~CAPACITY;}
private static int workerCountOf(int c) {return c & CAPACITY;}
private static int ctlOf(int rs, int wc) {return rs | wc;}通过正文，咱们会发现，它应用了一个 int 来保留状态信息和以后的工作线程数，其中 32 位 int 的前 3 位用来保留状态，之后的 29 位保留了工作线程的数量。那么 ThreadPoolExecutor 都有哪些状态呢？
- RUNNING: 承受新工作，并且正在解决工作，这是失常工作状态
- SHUTDOWN: 不再承受新工作，然而会解决队列中的残余工作，这是执行完 shutdown 接口之后的状态
- STOP: 不再承受新工作，同时，也不再解决队列中的残余工作，并且会打断所有进行中的工作，这是执行完 shutdownNow 接口之后的状态
- TIDYING: 所有工作曾经解决完，并且工作线程数为 0，然而执行 terminated 回调函数之前
- TERMINATED: 执行 terminated 回调函数之后, terminated 是 ThreadPoolExecutor 的一个函数，能够通过继承 ThreadPoolExecutor 来覆写该函数
明确了外部数据的组织形式之后，再来看 execute 的实现逻辑就清晰多了。
/**
* Executes the given task sometime in the future. The task * may execute in a new thread or in an existing pooled thread. * * If the task cannot be submitted for execution, either because this * executor has been shutdown or because its capacity has been reached, * the task is handled by the current {@code RejectedExecutionHandler}. * * @param command the task to execute * @throws RejectedExecutionException at discretion of * {@code RejectedExecutionHandler}, if the task * cannot be accepted for execution * @throws NullPointerException if {@code command} is null */
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();
// 首先它会获取以后外部数据 ctl，而后从中提取工作线程数（后 29 位），如果小于 corePoolSize，则创立新的线程 if (workerCountOf(c) < corePoolSize) {// 线程数小于外围线程数时工作不入队，间接通过参数传递到 worker 线程中 if (addWorker(command, true))
// 创立胜利的话，间接返回 return;
// 拉取最新外部数据 ctl c = ctl.get();}
if (isRunning(c) && workQueue.offer(command)) {// 如果以后还处于运行状态，并且工作队列没满 int recheck = ctl.get();
// 因为查看状态时并没有用到锁，所以这里在查看一次如果以后状态不是运行中，就把工作从队列中删除，而后回绝该工作 if (! isRunning(recheck) && remove(command))
reject(command);
// 否则，查看工作线程数是否为 0，是的话就增加一个工作线程，之所以会呈现这种状况是因为在 workQueue.offer 执行之前可能最初一个线程被销毁了（思考 keepAliveTime）else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
// 如果以后状态不是运行中，或者工作线程满了，入队失败时，当前任务也是参数传递到 worker 线程中 else if (!addWorker(command, false))
// 如果 addWorker 失败，既有可能是线程池曾经进行，也有可能是线程数达到 maxThreadSize，无论是哪一种，都须要回绝该工作 reject(command);
// 如果 addWorker 胜利了，阐明以后状态还是运行中，是工作队列满了，然而线程数没有达到 maxThreadSize }-
首先它会获取以后外部数据 ctl，而后从中提取工作线程数（后 29 位），如果小于 corePoolSize，则创立新的线程，线程数小于外围线程数时工作不入队，间接通过参数传递到 worker 线程中
* 创立胜利的话，间接返回 * 否则，拉取最新外部数据 ctl -
走到第二步有两个可能，一个是线程池可能曾经进行工作了，也有可能线程数曾经达到了 corePoolSize，所以在这里咱们要分状况解决
1. 如果以后还处于运行状态，并且工作队列没满, 又分两种状况 1. 因为查看状态时并没有用到锁，所以这里在查看一次如果以后状态不是运行中，就把工作从队列中删除，而后回绝该工作 2. 否则，查看工作线程数是否为 0，是的话就增加一个工作线程，之所以会呈现这种状况是因为在 workQueue.offer 执行之前可能最初一个线程被销毁了 -
如果以后状态不是运行中，或者工作线程满了，通过 addWorker 的后果来决定到底该怎么做
1. 如果 addWorker 胜利了，阐明以后状态还是运行中，是工作队列满了，然而线程数没有达到 maxThreadSize 2. 如果 addWorker 失败，即有可能是线程池曾经进行，也有可能是线程数达到 maxThreadSize，无论是哪一种，都须要回绝该工作
execute 的整体实现就是对线程池执行规定的复现，其中用到了比拟多的 CAS 操作，而不是通过加一个大锁，显然这样效率更好，而这里所说的 CAS 操作次要是指 addWorker 函数，那么它外部是怎么实现的呢？
/**
* Checks if a new worker can be added with respect to current * pool state and the given bound (either core or maximum). If so, * the worker count is adjusted accordingly, and, if possible, a * new worker is created and started, running firstTask as its * first task. This method returns false if the pool is stopped or * eligible to shut down. It also returns false if the thread * factory fails to create a thread when asked. If the thread * creation fails, either due to the thread factory returning * null, or due to an exception (typically OutOfMemoryError in * Thread.start()), we roll back cleanly. * * @param firstTask the task the new thread should run first (or * null if none). Workers are created with an initial first task * (in method execute()) to bypass queuing when there are fewer * than corePoolSize threads (in which case we always start one), * or when the queue is full (in which case we must bypass queue). * Initially idle threads are usually created via * prestartCoreThread or to replace other dying workers. * * @param core if true use corePoolSize as bound, else * maximumPoolSize. (A boolean indicator is used here rather than a * value to ensure reads of fresh values after checking other pool * state). * @return true if successful */
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {// 循环 CAS int c = ctl.get();
int rs = runStateOf(c);
// Check if queue empty only if necessary. // 如果调用过 shutdown 或者 shutdownNow 个别就不必创立工作线程了，然而这里要排除一种状况：// 思考到以后状态是 SHUTDOWN，并且 firstTask == null，就是 execute 中 addWorker(null, false) 的状况，这阐明方才有一个工作曾经入队了，然而最初一个工作线程可能在 workQueue.offer 执行之前被销毁了 // 而且 !workQueue.isEmpty() 工作队列不等于空，这时候就须要创立一个线程来吧 workQueue 中剩下的工作解决完 if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
// 通过了状态查看，这里咱们通过 CAS 批改工作线程数 for (;;) {int wc = workerCountOf(c);
// 查看线程数是否过多 if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
// cas 操作胜利则跳到下一步 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 }
}
// 走到这一步阐明工作线程数曾经胜利 +1，状态目前来看没问题 boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {// 创立工作线程 w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
// 这里应用到了一个锁 mainLock，它次要是用来爱护所有工作线程的汇合 workers，而且在执行 shutdown 时也会持有该锁，所以这里在锁的爱护下进行最终的状态确认 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());
// 最初的状态查看，并退出到 workers 汇合中，小于 SHUTDOWN 阐明以后状态是 RUNING，或者 rs == SHUTDOWN && firstTask == null，这就是 execute 中 addWorker(null, false) 的状况 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 {// 否则，通过 CAS 进行工作线程数 -1，查看终止状态，并帮助最终的状态转换 // 前两个操作比拟好了解，这里所谓的状态转换是指从 SHUTDOWN -> TIDYING->TERMINATED 的转换 // 在 SHUTDOWN 时，如果发现线程数为 0 了，就开始状态转换 if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
/**
* Rolls back the worker thread creation. * - removes worker from workers, if present * - decrements worker count * - rechecks for termination, in case the existence of this * worker was holding up termination */
private void addWorkerFailed(Worker w) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {if (w != null)
workers.remove(w);
decrementWorkerCount();
// 查看以后的线程池状态和工作线程数，如果工作队列空了，并且线程数为 0，就开始执行 terminated 回调 tryTerminate();} finally {mainLock.unlock();
}
}addWorker 的工作流程如下：
- 通过一个循环进行状态查看并减少工作线程数
-
线程数减少胜利，则开始创立理论的线程，创立好之后通过一把锁来进行最初的状态确认
1. 如果状态 OK 则将其退出到工作线程集中，并启动线程 2. 否则进行必要的清理工作
到此为止，线程池工作的协调工作局部就介绍完了，然而最外围工作线程 Worker 还没有讲，咱们来看一看它是怎么实现的。
/**
* Class Worker mainly maintains interrupt control state for * threads running tasks, along with other minor bookkeeping. * This class opportunistically extends AbstractQueuedSynchronizer * to simplify acquiring and releasing a lock surrounding each * task execution. This protects against interrupts that are * intended to wake up a worker thread waiting for a task from * instead interrupting a task being run. We implement a simple * non-reentrant mutual exclusion lock rather than use * ReentrantLock because we do not want worker tasks to be able to * reacquire the lock when they invoke pool control methods like * setCorePoolSize. Additionally, to suppress interrupts until * the thread actually starts running tasks, we initialize lock * state to a negative value, and clear it upon start (in * runWorker). */
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;
/**
* Creates with given first task and thread from ThreadFactory. * @param firstTask the first task (null if none) */
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);
}
// Lock methods // // The value 0 represents the unlocked state. // The value 1 represents the locked state.
protected boolean isHeldExclusively() {return getState() != 0;
}
protected boolean tryAcquire(int unused) {if (compareAndSetState(0, 1)) {setExclusiveOwnerThread(Thread.currentThread());
return true;
}
return false;
}
protected boolean tryRelease(int unused) {setExclusiveOwnerThread(null);
setState(0);
return true;
}
public void lock() { acquire(1); }
public boolean tryLock() { return tryAcquire(1); }
public void unlock() { release(1); }
public boolean isLocked() { return isHeldExclusively(); }
void interruptIfStarted() {
Thread t;
if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
try {t.interrupt();
} catch (SecurityException ignore) {}}
}
}能够看到 Worker 的外部实现了一个锁，为什么须要锁呢？咱们须要通过它来爱护运行中的工作，在执行 showdown 的时候，咱们是不能打断正在工作的线程的，所以在 showdown 过程中，打断工作线程前都须要尝试获取该锁，而且工作线程在执行工作时，也会始终持有该锁。
此外，还有一点就是 Worker 中的锁，初始是已加锁状态的，这是为什么呢？因为工作线程再退出到 Wrokers 工作线程汇合时，线程还没有被 start，这时候如果执行 showdown(前面具体介绍这个该过程)，可能会产生 interrupt 产生在 start 之前的状况。而如果 interrupt 产生在 start 之前，该线程的中断标记位并不会被置位，也就是说会失落中断。对于 Interrupt 的实现在本文的最初有介绍，看到那大家就能了解了。
Worker 线程的理论工作代码在 runWorker 中。
/**
* Main worker run loop. Repeatedly gets tasks from queue and * executes them, while coping with a number of 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. * * 2. Before running any task, the lock is acquired to prevent * other pool interrupts while the task is executing, and then we * ensure that unless pool is stopping, this thread does not have * its interrupt set. * * 3. Each task run is preceded by a call to beforeExecute, which * might throw an exception, in which case we cause thread to die * (breaking loop with completedAbruptly true) without processing * the task. * * 4. Assuming beforeExecute completes normally, we run the task, * gathering any of its thrown exceptions to send to afterExecute. * We separately handle RuntimeException, Error (both of which the * specs guarantee that we trap) and arbitrary Throwables. * Because we cannot rethrow Throwables within Runnable.run, we * wrap them within Errors on the way out (to the thread's * UncaughtExceptionHandler). Any thrown exception also * conservatively causes thread to die. * * 5. After task.run completes, we call afterExecute, which may * also throw an exception, which will also cause thread to * die. According to JLS Sec 14.20, this exception is the one that * will be in effect even if task.run throws. * * The net effect of the exception mechanics is that afterExecute * and the thread's UncaughtExceptionHandler have as accurate * information as we can provide about any problems encountered by * user code. * * @param w the worker */
final void runWorker(Worker w) {Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
// 解锁，尔后 shutdown 函数能力中断该线程 w.unlock(); // allow interrupts boolean completedAbruptly = true;
try {// 如果 firstTask 不等于 null，则不查看状态就开始试图执行 while (task != null || (task = getTask()) != null) {// 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 interrupt // 这个中央设计的比较复杂，就像后面说的如果执行 firstTask 时被 SHUTDOWN，那么这个没有进入工作队列的 firstTask 是须要失常执行完的，// 然而 firstTask 执行前可能线程被中断了（调用了 shutdown 函数），这时候咱们须要革除中断标记位才行，这样能力算失常执行，也就是上面的第一次 Thread.interrupted() 调用 // 而当咱们执行 shutdownNow 时，线程池的状态是 STOP，又或者在咱们方才进行革除中断标记位之后线程池的状态变成了 STOP，并且以后中断标记位没有被无效的设置的话 !wt.isInterrupted()，// 咱们就要补上方才误清的中断标记位，留神这里咱们并没有间接完结线程，而是设置标记位并执行指标工作，让指标工作去决定遇到中断标记位时须要作出什么解决，而不是线程池间接主持生杀大权 if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
// 如果状态曾经是 STOP，就打断本人, 只会设置中断标记位，工作还是会继续执行的 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);
}
}
/**
* Performs blocking or timed wait for a task, depending on * current configuration settings, or returns null if this worker * must exit because of any of: * 1. There are more than maximumPoolSize workers (due to * a call to setMaximumPoolSize). * 2. The pool is stopped. * 3. The pool is shutdown and the queue is empty. * 4. This worker timed out waiting for a task, and timed-out * workers are subject to termination (that is, * {@code allowCoreThreadTimeOut || workerCount > corePoolSize}) * both before and after the timed wait, and if the queue is * non-empty, this worker is not the last thread in the pool. * * @return task, or null if the worker must exit, in which case * workerCount is decremented */
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. // 如果线程池曾经 STOP 或者 SHUTDOWN 状态时工作队列为空，就销毁线程 if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
// Are workers subject to culling? boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
// 如果线程数过多或者超时，并且以后线程数 >1 或者工作队列为空，就销毁线程 if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
// 如果带超时性能，allowCoreThreadTimeOut || wc > corePoolSize 则获取工作的最大超时工夫是 keepAliveTime // 否则，无限期期待 Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {timedOut = false;}
}
}
/**
* Performs cleanup and bookkeeping for a dying worker. Called * only from worker threads. Unless completedAbruptly is set, * assumes that workerCount has already been adjusted to account * for exit. This method removes thread from worker set, and * possibly terminates the pool or replaces the worker if either * it exited due to user task exception or if fewer than * corePoolSize workers are running or queue is non-empty but * there are no workers. * * @param w the worker * @param completedAbruptly if the worker died due to user exception */
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 tryTerminate();
int c = ctl.get();
if (runStateLessThan(c, STOP)) {// 状态还没到 STOP 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);
}
}RunWorker 的工作流程是：
- 先解锁，容许线程被中断
- 如果创立线程时有 firstTask 则优先执行首要任务，执行实现后，再查看以后线程池状态并从工作队列中拉取工作执行
- 如果工作队列中没有工作，那么以后线程会依据是否存在 keepAliveTime 来决定是通过 poll 进行计时期待还是，通过 take 进行继续期待
- 在获取到工作时，先会进行 Worker 的加锁，而后再开始执行工作，在执行工作前后还别离有 beforeExecute 和 afterExecute 回调，执行期间如果抛出任何异样，都会导致线程的销毁
- 当线程退出时，会进行必要的清理工作，比方保护工作线程数量等，最初适时地进行状态转换 tryTerminate
线程池的工作逻辑大抵就是这样，最初咱们来简略介绍一下敞开一个线程池的逻辑。
/**
* Initiates an orderly shutdown in which previously submitted * tasks are executed, but no new tasks will be accepted. * Invocation has no additional effect if already shut down. * * <p>This method does not wait for previously submitted tasks to * complete execution. Use {@link #awaitTermination awaitTermination} * to do that. * * @throws SecurityException {@inheritDoc} */
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {checkShutdownAccess();
advanceRunState(SHUTDOWN);
interruptIdleWorkers();
onShutdown(); // hook for ScheduledThreadPoolExecutor} finally {mainLock.unlock();
}
tryTerminate();}
/**
* Interrupts threads that might be waiting for tasks (as * indicated by not being locked) so they can check for * termination or configuration changes. Ignores * SecurityExceptions (in which case some threads may remain * uninterrupted). * * @param onlyOne If true, interrupt at most one worker. This is * called only from tryTerminate when termination is otherwise * enabled but there are still other workers. In this case, at * most one waiting worker is interrupted to propagate shutdown * signals in case all threads are currently waiting. * Interrupting any arbitrary thread ensures that newly arriving * workers since shutdown began will also eventually exit. * To guarantee eventual termination, it suffices to always * interrupt only one idle worker, but shutdown() interrupts all * idle workers so that redundant workers exit promptly, not * waiting for a straggler task to finish. */
private void interruptIdleWorkers(boolean onlyOne) {
// mainLock 爱护 workers 汇合 final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {for (Worker w : workers) {
Thread t = w.thread;
// 先获取线程锁，而后再中断 if (!t.isInterrupted() && w.tryLock()) {
try {t.interrupt();
} catch (SecurityException ignore) { } finally {w.unlock();
}
}
if (onlyOne)
break;
}
} finally {mainLock.unlock();
}
}如果应用 shutdown 接口敞开线程池，解决的就比拟温和：
- 先通过 CAS 批改状态为 SHOTDOWN
-
对所有工作线程先试着获取其 Worker 锁
1. 如果胜利拿到锁，阐明它没有执行工作或者刚要执行 firstTask，如果没有执行工作间接 interrupt 并没有问题。而如果刚要执行 firstTask：1. 当状态是 SHUTDOWN 时显然咱们须要让该线程解决完 firstTask 再销毁才符合规范，毕竟 firstTask 没有进入工作队列，这部分的逻辑实际上是在 runWorker 中管制的。2. 当状态至多为 STOP 时，runWorker 也没有间接进行工作的执行，只是设置了中断标记位，具体这个工作被中断后是须要进行还是硬着头皮执行是须要该工作外部决定的，线程池只管发中断，不应该管制 "生杀大权" - 如果没有拿到锁，阐明工作正在执行中，让它继续执行
而 shutdownNow 的处理过程相较于 shutdown 就粗犷很多了。它不须要获取 Worker 锁而是间接执行线程的 interrupt 函数。
/**
* Attempts to stop all actively executing tasks, halts the * processing of waiting tasks, and returns a list of the tasks * that were awaiting execution. These tasks are drained (removed) * from the task queue upon return from this method. * * <p>This method does not wait for actively executing tasks to * terminate. Use {@link #awaitTermination awaitTermination} to * do that. * * <p>There are no guarantees beyond best-effort attempts to stop * processing actively executing tasks. This implementation * cancels tasks via {@link Thread#interrupt}, so any task that * fails to respond to interrupts may never terminate. * * @throws SecurityException {@inheritDoc} */
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;
}
/**
* Interrupts all threads, even if active. Ignores SecurityExceptions * (in which case some threads may remain uninterrupted). */
private void interruptWorkers() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {for (Worker w : workers)
w.interruptIfStarted();} finally {mainLock.unlock();
}
}最初咱们简略说一说 ScheduledThreadPoolExecutor，它是在 ThreadPoolExecutor 的根底上额定实现了定时工作的性能，你能够简略地认为它的核心内容就是实现了一个 ThreadPoolExecutor 中应用到的阻塞队列，ScheduledThreadPoolExecutor 的外部实现了一个延时工作队列, 队列中的工作依照执行工夫点排序 (应用到二分查找)，当第一个工作线程发现以后没有可执行的工作（下一个工作可能要在 N 秒之后执行）时，它会成为领头人线程，而其余线程都会在一个 Condition 上永恒期待，而领头人线程会在该 Condition 上期待 N 秒，当期待超时或者有须要立即执行的工作被增加时，领头人线程会苏醒过来并唤醒该 Condition 上的下一个线程。通过这个延时工作队列，ScheduledThreadPoolExecutor 达到了没有工作到期时，所有工作线程都在期待。有工作到期时，线程会立马唤醒并开始工作的成果。
Lock
JUC 中锁的实现次要有 3 个，别离是 ReentrantLock，ReentrantReadWriteLock，StampedLock。本节咱们次要介绍各种锁的应用，前面锁的分类局部，咱们会比照着介绍各个锁的实现计划。
ReentrantLock 是最根底的一个锁，它是一个可重入锁，也就是说持有锁的线程，能够屡次加锁，此外，通过参数咱们还能管制它是否是一个偏心锁，偏心锁就是说大家取得锁的程序是和尝试加锁的程序统一的。
public static class LockTest {private ReentrantLock lock = new ReentrantLock(true);
private void doThing() {lock.lock();
try {doSomeThings();
} finally {lock.unlock();
}
}
private void doSomeThings() {lock.lock();
try {// do something} finally {lock.unlock();
}
}
}ReentrantReadWriteLock 次要是在数据既有读又有写的场景中应用，它能保障读操作之间不互斥，然而读写和写写之间互斥。它外面有两个锁，在须要读数据时，对读锁加锁，在须要写数据时对写锁加锁。同样，咱们也能够在结构读写锁的时候通过参数管制其是否是偏心锁。
public static class LockTest {private ReentrantReadWriteLock lock = new ReentrantReadWriteLock(true);
private int data;
private void read() {lock.readLock().lock();
try {// read System.out.println(data);
} finally {lock.readLock().unlock();}
}
private void write() {lock.writeLock().lock();
try {// write data = 10;} finally {lock.writeLock().unlock();}
}
private void lockDowngrade() {lock.writeLock().lock();
try {
// write data = 10;
read();} finally {lock.writeLock().unlock();}
}
private void deadLock() {lock.readLock().lock();
try {// read System.out.println(data);
write();} finally {lock.readLock().unlock();}
}
}对于读写锁，有一个小知识点是，当持有写锁时，是能够取得到读锁的，因为有些写操作可能外部调用到了读操作，就像上例一样。而先持有读锁，再去取得写锁时，就会产生死锁，大家在应用的时候肯定要留神这个问题。
最初要介绍的是 StampedLock，它和 ReentrantReadWriteLock 很像。ReadWriteLock 中如果有线程正在读，写线程须要期待读线程开释锁后能力获取写锁，即读的过程中不容许写，这是一种乐观的读锁。要进一步晋升并发执行效率，Java 8 引入了新的读写锁：StampedLock。
在应用 StampedLock 时，咱们能够先应用乐观读锁，在这个过程中其余线程是能够取得写锁的，也就是说咱们读的数据就可能不统一，所以，须要一点额定的代码来判断读的过程中是否有写入。乐观锁的意思就是乐观地预计读的过程中大概率不会有写入，因而被称为乐观锁。反过来，乐观锁则是读的过程中回绝有写入，也就是写入必须期待。显然乐观锁的并发效率更高，但一旦有小概率的写入导致读取的数据不统一，须要能检测进去，再读一遍就行。
public class LockTest {private final StampedLock lock = new StampedLock();
private int num1;
private int num2;
public void change(int num1, int num2) {long stamp = lock.writeLock();
try {
this.num1 = num1;
this.num2 = num2;
} finally {lock.unlockWrite(stamp);
}
}
public double readAndCompute() {long stamp = lock.tryOptimisticRead(); // 获取以后数据版本号 int currentNum1 = num1;
int currentNum2 = num2;
if (!lock.validate(stamp)) {// 确认之前的版本号和最新版本号是否统一，如果统一，则阐明期间没产生数据更改，否则，可能数据被更改了 stamp = lock.readLock(); // 数据版本号不统一，通过乐观读锁来获取正确数据 try {
currentNum1 = num1;
currentNum2 = num2;
} finally {lock.unlockRead(stamp); // 开释乐观读锁 }
}
return currentNum1 + currentNum2;
}
}StampedLock 写锁的应用和读写锁齐全一样，区别在与多了一个 tryOptimisticRead 接口，它可能取得以后数据版本号，咱们记录下读数据之前的版本号，而后再读取所有数据，最初拿之前记录的版本号和最新版本号做比照，如果统一，则阐明期间没产生数据更改，能够失常应用，否则，可能数据被更改了，这时候就得改用乐观读锁加锁，在读取数据，这个和 ReentrantReadWriteLock 的应用流程就一样了。
可见，StampedLock 把读锁细分为乐观读和乐观读，能进一步晋升并发效率。但这也是有代价的：一是代码更加简单，二是 StampedLock 是不可重入锁，不能在一个线程中重复获取同一个锁。
StampedLock 还提供了更简单的将乐观读锁降级为写锁的性能，它次要应用在 if-then-update 的场景：即先读，如果读的数据满足条件，就返回，如果读的数据不满足条件，再尝试写。
正文完