Java线程池中的核心线程是如何被重复利用的
Java线程池中的核心线程是如何被重复利用的？
引言
在Java开发中，经常需要创建线程去执行一些任务，实现起来也非常方便，但如果并发的线程数量很多，并且每个线程都是执行一个时间很短的任务就结束了，这样频繁创建线程就会大大降低系统的效率，因为频繁创建线程和销毁线程需要时间。此时，我们很自然会想到使用线程池来解决这个问题。
使用线程池的好处：
降低资源消耗。java中所有的池化技术都有一个好处，就是通过复用池中的对象，降低系统资源消耗。设想一下如果我们有n多个子任务需要执行，如果我们为每个子任务都创建一个执行线程，而创建线程的过程是需要一定的系统消耗的，最后肯定会拖慢整个系统的处理速度。而通过线程池我们可以做到复用线程，任务有多个，但执行任务的线程可以通过线程池来复用，这样减少了创建线程的开销，系统资源利用率得到了提升。
降低管理线程的难度。多线程环境下对线程的管理是最容易出现问题的，而线程池通过框架为我们降低了管理线程的难度。我们不用再去担心何时该销毁线程，如何最大限度的避免多线程的资源竞争。这些事情线程池都帮我们代劳了。
提升任务处理速度。线程池中长期驻留了一定数量的活线程，当任务需要执行时，我们不必先去创建线程，线程池会自己选择利用现有的活线程来处理任务。
- // 获取运行状态
- private static int runStateOf(int c) { return c & ~CAPACITY; }
- // 获取活动线程数
- private static int workerCountOf(int c) { return c & CAPACITY; }123456
- public void execute(Runnable command) {
- if (command == null)
- throw new NullPointerException();
- 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);
- }123456789101112131415161718192021
- // 为分析而简化后的代码
- public void execute(Runnable command) {
- int c = ctl.get();
- if (workerCountOf(c) < corePoolSize) {
- // 如果当前活动线程数小于corePoolSize，则新建一个线程放入线程池中，并把任务添加到该线程中
- if (addWorker(command, true))
- return;
- c = ctl.get();
- }
- // 如果当前活动线程数大于等于corePoolSize，则尝试将任务放入缓存队列
- if (workQueue.offer(command)) {
- int recheck = ctl.get();
- if (workerCountOf(recheck) == 0)
- addWorker(null, false);
- }else {
- // 缓存已满，新建一个线程放入线程池，并把任务添加到该线程中（此时新建的线程相当于非核心线程）
- addWorker(command, false)
- }
- }12345678910111213141516171819202122
如果 当前活动线程数 < 指定的核心线程数，则创建并启动一个线程来执行新提交的任务（此时新建的线程相当于核心线程）；
如果 当前活动线程数 >= 指定的核心线程数，且缓存队列未满，则将任务添加到缓存队列中；
如果 当前活动线程数 >= 指定的核心线程数，且缓存队列已满，则创建并启动一个线程来执行新提交的任务（此时新建的线程相当于非核心线程）；
- private boolean addWorker(Runnable firstTask, boolean core) {
- 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);
- 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;
- }12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667
- // 为分析而简化后的代码
- private boolean addWorker(Runnable firstTask, boolean core) {
- int wc = workerCountOf(c);
- if (wc >= (core ? corePoolSize : maximumPoolSize))
- // 如果当前活动线程数 >= 指定的核心线程数，不创建核心线程
- // 如果当前活动线程数 >= 指定的最大线程数，不创建非核心线程
- return false;
- boolean workerStarted = false;
- boolean workerAdded = false;
- Worker w = null;
- try {
- // 新建一个Worker，将要执行的任务作为参数传进去
- w = new Worker(firstTask);
- final Thread t = w.thread;
- if (t != null) {
- workers.add(w);
- workerAdded = true;
- if (workerAdded) {
- // 启动刚刚新建的那个worker持有的线程，等下要看看这个线程做了啥
- t.start();
- workerStarted = true;
- }
- }
- } finally {
- if (! workerStarted)
- addWorkerFailed(w);
- }
- return workerStarted;
- }1234567891011121314151617181920212223242526272829303132
- private final class Worker
- extends AbstractQueuedSynchronizer
- implements Runnable{
- // ....
- }
- Worker(Runnable firstTask) {
- setState(-1); // inhibit interrupts until runWorker
- this.firstTask = firstTask;
- this.thread = getThreadFactory().newThread(this);
- }123456789101112
- 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
- boolean completedAbruptly = true;
- try {
- 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);
- }
- }123456789101112131415161718192021222324252627282930313233343536373839404142434445464748
- // 为分析而简化后的代码
- final void runWorker(Worker w) {
- Runnable task = w.firstTask;
- w.firstTask = null;
- while (task != null || (task = getTask()) != null) {
- try {
- task.run();
- } finally {
- task = null;
- }
- }
- }12345678910111213
- 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?
- 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;
- }
- }
- }1234567891011121314151617181920212223242526272829303132333435363738
- // 为分析而简化后的代码
- private Runnable getTask() {
- boolean timedOut = false;
- for (;;) {
- int c = ctl.get();
- int wc = workerCountOf(c);
- // timed变量用于判断是否需要进行超时控制。
- // allowCoreThreadTimeOut默认是false，也就是核心线程不允许进行超时；
- // wc > corePoolSize，表示当前线程池中的线程数量大于核心线程数量；
- // 对于超过核心线程数量的这些线程，需要进行超时控制
- boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
- if (timed && timedOut) {
- // 如果需要进行超时控制，且上次从缓存队列中获取任务时发生了超时，那么尝试将workerCount减1,即当前活动线程数减1，
- // 如果减1成功，则返回null，这就意味着runWorker()方法中的while循环会被退出，其对应的线程就要销毁了，也就是线程池中少了一个线程了
- if (compareAndDecrementWorkerCount(c))
- return null;
- continue;
- }
- try {
- Runnable r = timed ?
- workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
- workQueue.take();
- // 注意workQueue中的poll()方法与take()方法的区别
- //poll方式取任务的特点是从缓存队列中取任务,最长等待keepAliveTime的时长，取不到返回null
- //take方式取任务的特点是从缓存队列中取任务，若队列为空,则进入阻塞状态，直到能取出对象为止
- if (r != null)
- return r;
- timedOut = true;
- } catch (InterruptedException retry) {
- timedOut = false;
- }
- }
- }123456789101112131415161718192021222324252627282930313233343536373839
如果当前活动线程数大于核心线程数，当去缓存队列中取任务的时候，如果缓存队列中没任务了，则等待keepAliveTime的时长，此时还没任务就返回null，这就意味着runWorker()方法中的while循环会被退出，其对应的线程就要销毁了，也就是线程池中少了一个线程了。因此只要线程池中的线程数大于核心线程数就会这样一个一个地销毁这些多余的线程。
如果当前活动线程数小于等于核心线程数，同样也是去缓存队列中取任务，但当缓存队列中没任务了，就会进入阻塞状态，直到能取出任务为止，因此这个线程是处于阻塞状态的，并不会因为缓存队列中没有任务了而被销毁。这样就保证了线程池有N个线程是活的，可以随时处理任务，从而达到重复利用的目的。
当有新任务来的时候，先看看当前的线程数有没有超过核心线程数，如果没超过就直接新建一个线程来执行新的任务，如果超过了就看看缓存队列有没有满，没满就将新任务放进缓存队列中，满了就新建一个线程来执行新的任务，如果线程池中的线程数已经达到了指定的最大线程数了，那就根据相应的策略拒绝任务。
当缓存队列中的任务都执行完了的时候，线程池中的线程数如果大于核心线程数，就销毁多出来的线程，直到线程池中的线程数等于核心线程数。此时这些线程就不会被销毁了，它们一直处于阻塞状态，等待新的任务到来。
很显然，线程池一个很显著的特征就是“长期驻留了一定数量的活线程”，避免了频繁创建线程和销毁线程的开销，那么它是如何做到的呢？我们知道一个线程只要执行完了run()方法内的代码，这个线程的使命就完成了，等待它的就是销毁。既然这是个“活线程”，自然是不能很快就销毁的。为了搞清楚这个“活线程”是如何工作的，下面通过追踪源码来看看能不能解开这个疑问。
分析方法
在分析源码之前先来思考一下要怎么去分析，源码往往是比较复杂的，如果知识储备不够丰厚，很有可能会读不下去，或者读岔了。一般来讲要时刻紧跟着自己的目标来看代码，跟目标关系不大的代码可以不理会它，一些异常的处理也可以暂不理会，先看正常的流程。就我们现在要分析的源码而言，目标就是看看线程是如何被复用的。那么对于线程池的状态的管理以及非正常状态下的处理代码就可以不理会，具体来讲，在ThreadPollExcutor类中，有一个字段 private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); 是对线程池的运行状态和线程池中有效线程的数量进行控制的， 它包含两部分信息: 线程池的运行状态 (runState) 和线程池内有效线程的数量 (workerCount)，还有几个对ctl进行计算的方法:
以上两个方法在源码中经常用到，结合我们的目标，对运行状态的一些判断及处理可以不用去管，而对当前活动线程数要加以关注等等。
下面将遵循这些原则来分析源码。
解惑
当我们要向线程池添加一个任务时是调用ThreadPollExcutor对象的execute(Runnable command)方法来完成的，所以先来看看ThreadPollExcutor类中的execute(Runnable command)方法的源码：
按照我们在分析方法中提到的一些原则，去掉一些相关性不强的代码，看看核心代码是怎样的。
这样一看，逻辑应该清晰很多了。
接下来看 addWorker(Runnable firstTask, boolean core)方法
同样，我们也来简化一下：
看到这里，我们大概能猜测到，addWorker方法的功能就是新建一个线程并启动这个线程，要执行的任务应该就是在这个线程中执行。为了证实我们的这种猜测需要再来看看Worker这个类。
从上面的Worker类的声明可以看到，它实现了Runnable接口，以及从它的构造方法中可以知道待执行的任务赋值给了它的变量firstTask，并以它自己为参数新建了一个线程赋值给它的变量thread，那么运行这个线程的时候其实就是执行Worker的run()方法，来看一下这个方法：
在run()方法中只调了一下 runWorker(this) 方法，再来简化一下这个 runWorker() 方法
很明显，runWorker()方法里面执行了我们新建Worker对象时传进去的待执行的任务，到这里为止貌似这个worker的run()方法就执行完了，既然执行完了那么这个线程也就没用了，只有等待虚拟机销毁了。那么回顾一下我们的目标：Java线程池中的核心线程是如何被重复利用的？好像并没有重复利用啊，新建一个线程，执行一个任务，然后就结束了，销毁了。没什么特别的啊，难道有什么地方漏掉了，被忽略了？再仔细看一下runWorker()方法的代码，有一个while循环，当执行完firstTask后task==null了，那么就会执行判断条件 (task = getTask()) != null，我们假设这个条件成立的话，那么这个线程就不止只执行一个任务了，可以执行多个任务了，也就实现了重复利用了。答案呼之欲出了，接着看getTask()方法
老规矩，简化一下代码来看：
从以上代码可以看出，getTask()的作用是
小结
通过以上的分析，应该算是比较清楚地解答了“线程池中的核心线程是如何被重复利用的”这个问题，同时也对线程池的实现机制有了更进一步的理解：
注意：
本文所说的“核心线程”、“非核心线程”是一个虚拟的概念，是为了方便描述而虚拟出来的概念，在代码中并没有哪个线程被标记为“核心线程”或“非核心线程”，所有线程都是一样的，只是当线程池中的线程多于指定的核心线程数量时，会将多出来的线程销毁掉，池中只保留指定个数的线程。那些被销毁的线程是随机的，可能是第一个创建的线程，也可能是最后一个创建的线程，或其它时候创建的线程。一开始我以为会有一些线程被标记为“核心线程”，而其它的则是“非核心线程”，在销毁多余线程的时候只销毁那些“非核心线程”，而“核心线程”不被销毁。这种理解是错误的。
另外还有一个重要的接口 BlockingQueue 值得去了解，它定义了一些入队出队同步操作的方法，还可以阻塞，作用很大。
