Java 线程池(ThreadPoolExecutor)原理分析与使用

深碍√TFBOYSˉ_ 2022-06-10 12:57 277阅读 0赞

ThreadPoolExecutor原理概述

在我们的开发中“池”的概念并不罕见，有数据库连接池、线程池、对象池、常量池等等。下面我们主要针对线程池来一步一步揭开线程池的面纱。

## 使用线程池的好处 ##

1、降低资源消耗

可以重复利用已创建的线程降低线程创建和销毁造成的消耗。

2、提高响应速度

当任务到达时，任务可以不需要等到线程创建就能立即执行。

3、提高线程的可管理性

线程是稀缺资源，如果无限制地创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一分配、调优和监控

## 线程池的工作原理 ##

首先我们看下当一个新的任务提交到线程池之后，线程池是如何处理的

1、线程池判断核心线程池里的线程是否都在执行任务。如果不是，则创建一个新的工作线程来执行任务。如果核心线程池里的线程都在执行任务，则执行第二步。

2、线程池判断工作队列是否已经满。如果工作队列没有满，则将新提交的任务存储在这个工作队列里进行等待。如果工作队列满了，则执行第三步

3、线程池判断线程池的线程是否都处于工作状态。如果没有，则创建一个新的工作线程来执行任务。如果已经满了，则交给饱和策略来处理这个任务

## 线程池饱和策略 ##

这里提到了线程池的饱和策略，那我们就简单介绍下有哪些饱和策略：

**AbortPolicy**

为Java线程池默认的阻塞策略，不执行此任务，而且直接抛出一个运行时异常，切记ThreadPool[Executor][].execute需要try catch，否则程序会直接退出。

**DiscardPolicy**

直接抛弃，任务不执行，空方法

**DiscardOldestPolicy**

从队列里面抛弃head的一个任务，并再次execute 此task。

**CallerRunsPolicy**

在调用execute的线程里面执行此command，会阻塞入口

**用户自定义拒绝策略（最常用）**

实现RejectedExecutionHandler，并自己定义策略模式

**下我们以ThreadPoolExecutor为例展示下线程池的工作流程图**

![20170819200608739][]

![20170819200619605][]

1、如果当前运行的线程少于corePoolSize，则创建新线程来执行任务（注意，执行这一步骤需要获取全局锁）。

2、如果运行的线程等于或多于corePoolSize，则将任务加入BlockingQueue。

3、如果无法将任务加入BlockingQueue（队列已满），则在非corePool中创建新的线程来处理任务（注意，执行这一步骤需要获取全局锁）。

4、如果创建新线程将使当前运行的线程超出maximumPoolSize，任务将被拒绝，并调用RejectedExecutionHandler.rejectedExecution()方法。

ThreadPoolExecutor采取上述步骤的总体设计思路，是为了在执行execute()方法时，尽可能地避免获取全局锁（那将会是一个严重的可伸缩瓶颈）。在ThreadPoolExecutor完成预热之后（当前运行的线程数大于等于corePoolSize），几乎所有的execute()方法调用都是执行步骤2，而步骤2不需要获取全局锁。

**线程池状态以及含义**

RUNNING 运行态

SHUTDOWN 关闭，此时不接受新的任务，但继续处理队列中的任务。

STOP 停止，此时不接受新的任务，不处理队列中的任务，并中断正在执行的任务

TIDYING 所有的工作线程全部停止，并工作线程数量为0，将调用terminated方法，进入到TERMINATED

TERMINATED 终止状态

\-------------------------------------------------------------------

**ThreadPoolExecutor原理详细解析**

1、ThreadPoolExecutor概述

由于本人英语水平不高，为了不误导大家，我将源码中的注释复制下来，我不翻译原文，我从入学6个视角试图窥探一下ThreadPoolExecutor全貌。

1）创建线程池的方式

2）核心线程数、最大线程数

3）线程的创建

4）线程的Keep-alive（保持存活的空闲时间）

5）队列

6）任务的丢弃策略

/** 
 * An {@link ExecutorService} that executes each submitted task using 
 * one of possibly several pooled threads, normally configured 
 * using {@link Executors} factory methods. 
 * 
 * Thread pools address two different problems: they usually 
 * provide improved performance when executing large numbers of 
 * asynchronous tasks, due to reduced per-task invocation overhead, 
 * and they provide a means of bounding and managing the resources, 
 * including threads, consumed when executing a collection of tasks. 
 * Each {@code ThreadPoolExecutor} also maintains some basic 
 * statistics, such as the number of completed tasks. 
 * 
 * To be useful across a wide range of contexts, this class 
 * provides many adjustable parameters and extensibility 
 * hooks. However, programmers are urged to use the more convenient 
 * {@link Executors} factory methods {@link 
 * Executors#newCachedThreadPool} (unbounded thread pool, with 
 * automatic thread reclamation), {@link Executors#newFixedThreadPool} 
 * (fixed size thread pool) and {@link 
 * Executors#newSingleThreadExecutor} (single background thread), that 
 * preconfigure settings for the most common usage 
 * scenarios. Otherwise, use the following guide when manually 
 * configuring and tuning this class: 
 * 
 1、提供如下工厂方法创建线程池对象（ExecutorService实现类） 
 1）Executors.newCachedThreadPool,创建一个线程容量为Integer.MAX_VALUE的线程池，空闲时间为60s。 
 2）Executors.newFixedThreadPool,创建一个固定容量的线程池 
 3）Executors.newSingleThreadExecutor，创建一个线程的线程池 
 
 * <dl> 
 *核心线程与最大线程篇 
 * <dt>Core and maximum pool sizes</dt> 
 corePoolSize 核心线程数 
 maximumPoolSize 核心线程数 
 当一个任务提交到线程池 
 1） 如果当前线程池中的线程数小于corePoolSize时，直接创建一个先的线程。 
 2） 如果当前线程池中的线程数大于等于corePoolSize时，如果队列未满，直接将线程放入队列中，不新建线程。 
 3） 如果队列已满，但线程没有超过maximumPoolSize,则新建一个线程。 
 在运行过程中，可以通过调用setCorePoolSize,setMaximumPoolSize改变这两个参数 
 * 
 * <dd>A {@code ThreadPoolExecutor} will automatically adjust the 
 * pool size (see {@link #getPoolSize}) 
 * according to the bounds set by 
 * corePoolSize (see {@link #getCorePoolSize}) and 
 * maximumPoolSize (see {@link #getMaximumPoolSize}). 
 * 
 * When a new task is submitted in method {@link #execute}, and fewer 
 * than corePoolSize threads are running, a new thread is created to 
 * handle the request, even if other worker threads are idle. If 
 * there are more than corePoolSize but less than maximumPoolSize 
 * threads running, a new thread will be created only if the queue is 
 * full. By setting corePoolSize and maximumPoolSize the same, you 
 * create a fixed-size thread pool. By setting maximumPoolSize to an 
 * essentially unbounded value such as {@code Integer.MAX_VALUE}, you 
 * allow the pool to accommodate an arbitrary number of concurrent 
 * tasks. Most typically, core and maximum pool sizes are set only 
 * upon construction, but they may also be changed dynamically using 
 * {@link #setCorePoolSize} and {@link #setMaximumPoolSize}. </dd> 
 * 
 * 
 * 
 * <dt>On-demand construction</dt 
 * 核心线程的创建通常是有任务提交时新建的，当然，我们可以通过调用prestartCoreThread，或 
 prestartAllCoreThreads方法，预先创建核心线程数。 
 * <dd> By default, even core threads are initially created and 
 * started only when new tasks arrive, but this can be overridden 
 * dynamically using method {@link #prestartCoreThread} or {@link 
 * #prestartAllCoreThreads}. You probably want to prestart threads if 
 * you construct the pool with a non-empty queue. </dd> 
 * 
 
 
 
 * <dt>Creating new threads</dt> 
 *线程创建篇，新线程的创建，默认使用Executors.defautThreadFactory来创建线程，同一个线程创建工厂创建的线程具有相同的线程组，优先级，是否是后台线程(daemon),我们可以提供资金的线程创建工厂来改变这些属性，一般我们使用自己定义的线程工厂，主要的目的还是修改线程的名称，方便理解与跟踪。 
 * <dd>New threads are created using a {@link ThreadFactory}. If not 
 * otherwise specified, a {@link Executors#defaultThreadFactory} is 
 * used, that creates threads to all be in the same {@link 
 * ThreadGroup} and with the same {@code NORM_PRIORITY} priority and 
 * non-daemon status. By supplying a different ThreadFactory, you can 
 * alter the thread's name, thread group, priority, daemon status, 
 * etc. If a {@code ThreadFactory} fails to create a thread when asked 
 * by returning null from {@code newThread}, the executor will 
 * continue, but might not be able to execute any tasks. Threads 
 * should possess the "modifyThread" {@code RuntimePermission}. If 
 * worker threads or other threads using the pool do not possess this 
 * permission, service may be degraded: configuration changes may not 
 * take effect in a timely manner, and a shutdown pool may remain in a 
 * state in which termination is possible but not completed.</dd> 
 * 
 * <dt>Keep-alive times</dt> 
 * 如果线程池中线程数量超过了核心线程数，超过的线程如果空闲时间超过了keepAliveTime的线程会被终止； 
 先提出一个疑问：如果核心线程数设置为10，目前有12个线程，其中有3个超过了keepALiveTime,那有3个线程会被终止，还是只有两个，按照上述描述，应该是2个会被终止，，因为有个管家子 excess threads,从源码中去找答案吧。 
 * <dd>If the pool currently has more than corePoolSize threads, 
 * excess threads will be terminated if they have been idle for more 
 * than the keepAliveTime (see {@link #getKeepAliveTime}). This 
 * provides a means of reducing resource consumption when the pool is 
 * not being actively used. If the pool becomes more active later, new 
 * threads will be constructed. This parameter can also be changed 
 * dynamically using method {@link #setKeepAliveTime}. Using a value 
 * of {@code Long.MAX_VALUE} {@link TimeUnit#NANOSECONDS} effectively 
 * disables idle threads from ever terminating prior to shut down. By 
 * default, the keep-alive policy applies only when there are more 
 * than corePoolSizeThreads. But method {@link 
 * #allowCoreThreadTimeOut(boolean)} can be used to apply this 
 * time-out policy to core threads as well, so long as the 
 * keepAliveTime value is non-zero. </dd> 
 * 
 * <dt>Queuing</dt> 
 * 
 * <dd>Any {@link BlockingQueue} may be used to transfer and hold 
 * submitted tasks. The use of this queue interacts with pool sizing: 
 * 
 * <ul> 
 * 
 * <li> If fewer than corePoolSize threads are running, the Executor 
 * always prefers adding a new thread 
 * rather than queuing.</li> 
 * 
 * <li> If corePoolSize or more threads are running, the Executor 
 * always prefers queuing a request rather than adding a new 
 * thread.</li> 
 * 
 * <li> If a request cannot be queued, a new thread is created unless 
 * this would exceed maximumPoolSize, in which case, the task will be 
 * rejected.</li> 
 * 
 * </ul> 
 * 
 * There are three general strategies for queuing: 
 三种队列方案 
 1）直接传递，所有提交任务任务不入队列，直接传递给线程池。 
 2）有界队列 
 3）无界队列 
 采取何种队列，会对线程池中 核心线程数产生影响 
 再重复一下 核心线程的产生过程 
 1）如果当前线程池中线程数小于核心线程数，新任务到达，不管有没有队列，都是直接新建一个核心线程。 
 2）如果线程池中允许的线程达到核心线程数量时，根据不同的队列机制，有如下的处理方法： 
 a、如果是直接传递，则直接新增线程运行（没有达到最大线程数量） 
 b、如果是有界队列，先将任务入队列，如果任务队列已满，在线程数没有超过最大线程数限制的情况下，新 
 建一个线程来运行任务。 
 c、无界队列，则线程池中最大的线程数量等于核心线程数量，最大线程数量不会有产生任何影响。 
 * <ol> 
 * 
 * <li> Direct handoffs. A good default choice for a work 
 * queue is a {@link SynchronousQueue} that hands off tasks to threads 
 * without otherwise holding them. Here, an attempt to queue a task 
 * will fail if no threads are immediately available to run it, so a 
 * new thread will be constructed. This policy avoids lockups when 
 * handling sets of requests that might have internal dependencies. 
 * Direct handoffs generally require unbounded maximumPoolSizes to 
 * avoid rejection of new submitted tasks. This in turn admits the 
 * possibility of unbounded thread growth when commands continue to 
 * arrive on average faster than they can be processed. </li> 
 * 
 * <li> Unbounded queues. Using an unbounded queue (for 
 * example a {@link LinkedBlockingQueue} without a predefined 
 * capacity) will cause new tasks to wait in the queue when all 
 * corePoolSize threads are busy. Thus, no more than corePoolSize 
 * threads will ever be created. (And the value of the maximumPoolSize 
 * therefore doesn't have any effect.) This may be appropriate when 
 * each task is completely independent of others, so tasks cannot 
 * affect each others execution; for example, in a web page server. 
 * While this style of queuing can be useful in smoothing out 
 * transient bursts of requests, it admits the possibility of 
 * unbounded work queue growth when commands continue to arrive on 
 * average faster than they can be processed. </li> 
 * 
 * <li>Bounded queues. A bounded queue (for example, an 
 * {@link ArrayBlockingQueue}) helps prevent resource exhaustion when 
 * used with finite maximumPoolSizes, but can be more difficult to 
 * tune and control. Queue sizes and maximum pool sizes may be traded 
 * off for each other: Using large queues and small pools minimizes 
 * CPU usage, OS resources, and context-switching overhead, but can 
 * lead to artificially low throughput. If tasks frequently block (for 
 * example if they are I/O bound), a system may be able to schedule 
 * time for more threads than you otherwise allow. Use of small queues 
 * generally requires larger pool sizes, which keeps CPUs busier but 
 * may encounter unacceptable scheduling overhead, which also 
 * decreases throughput. </li> 
 * 
 * </ol> 
 * 
 * </dd> 
 * 
 * <dt>Rejected tasks</dt> 
 * 任务拒绝策略 
 1）AbortPolicy，抛出运行时异常 
 2）CallerRunsPolicy 调用者直接运行，不在线程中运行。 
 3）DiscardPolicy 直接将任务丢弃 
 4）DiscardOldestPolicy 丢弃队列中头部的任务。 
 * <dd> New tasks submitted in method {@link #execute} will be 
 * rejected when the Executor has been shut down, and also 
 * when the Executor uses finite bounds for both maximum threads and 
 * work queue capacity, and is saturated. In either case, the {@code 
 * execute} method invokes the {@link 
 * RejectedExecutionHandler#rejectedExecution} method of its {@link 
 * RejectedExecutionHandler}. Four predefined handler policies are 
 * provided: 
 * 
 * <ol> 
 * 
 * <li> In the default {@link ThreadPoolExecutor.AbortPolicy}, the 
 * handler throws a runtime {@link RejectedExecutionException} upon 
 * rejection. </li> 
 * 
 * <li> In {@link ThreadPoolExecutor.CallerRunsPolicy}, the thread 
 * that invokes {@code execute} itself runs the task. This provides a 
 * simple feedback control mechanism that will slow down the rate that 
 * new tasks are submitted. </li> 
 * 
 * <li> In {@link ThreadPoolExecutor.DiscardPolicy}, a task that 
 * cannot be executed is simply dropped. </li> 
 * 
 * <li>In {@link ThreadPoolExecutor.DiscardOldestPolicy}, if the 
 * executor is not shut down, the task at the head of the work queue 
 * is dropped, and then execution is retried (which can fail again, 
 * causing this to be repeated.) </li> 
 * 
 * </ol> 
 * 
 * It is possible to define and use other kinds of {@link 
 * RejectedExecutionHandler} classes. Doing so requires some care 
 * especially when policies are designed to work only under particular 
 * capacity or queuing policies. </dd> 
 * 
 * <dt>Hook methods</dt> 
 * 
 * <dd>This class provides {@code protected} overridable {@link 
 * #beforeExecute} and {@link #afterExecute} methods that are called 
 * before and after execution of each task. These can be used to 
 * manipulate the execution environment; for example, reinitializing 
 * ThreadLocals, gathering statistics, or adding log 
 * entries. Additionally, method {@link #terminated} can be overridden 
 * to perform any special processing that needs to be done once the 
 * Executor has fully terminated. 
 * 
 * If hook or callback methods throw exceptions, internal worker 
 * threads may in turn fail and abruptly terminate.</dd> 
 * 
 * <dt>Queue maintenance</dt> 
 * 
 * <dd> Method {@link #getQueue} allows access to the work queue for 
 * purposes of monitoring and debugging. Use of this method for any 
 * other purpose is strongly discouraged. Two supplied methods, 
 * {@link #remove} and {@link #purge} are available to assist in 
 * storage reclamation when large numbers of queued tasks become 
 * cancelled.</dd> 
 * 
 * <dt>Finalization</dt> 
 * 
 * <dd> A pool that is no longer referenced in a program AND 
 * has no remaining threads will be {@code shutdown} automatically. If 
 * you would like to ensure that unreferenced pools are reclaimed even 
 * if users forget to call {@link #shutdown}, then you must arrange 
 * that unused threads eventually die, by setting appropriate 
 * keep-alive times, using a lower bound of zero core threads and/or 
 * setting {@link #allowCoreThreadTimeOut(boolean)}. </dd> 
 * 
 * </dl> 
 * 
 * Extension example. Most extensions of this class 
 * override one or more of the protected hook methods. For example, 
 * here is a subclass that adds a simple pause/resume feature: 
 * 
 * <pre> {@code 
 * class PausableThreadPoolExecutor extends ThreadPoolExecutor { 
 * private boolean isPaused; 
 * private ReentrantLock pauseLock = new ReentrantLock(); 
 * private Condition unpaused = pauseLock.newCondition(); 
 * 
 * public PausableThreadPoolExecutor(...) { super(...); } 
 * 
 * protected void beforeExecute(Thread t, Runnable r) { 
 * super.beforeExecute(t, r); 
 * pauseLock.lock(); 
 * try { 
 * while (isPaused) unpaused.await(); 
 * } catch (InterruptedException ie) { 
 * t.interrupt(); 
 * } finally { 
 * pauseLock.unlock(); 
 * } 
 * } 
 * 
 * public void pause() { 
 * pauseLock.lock(); 
 * try { 
 * isPaused = true; 
 * } finally { 
 * pauseLock.unlock(); 
 * } 
 * } 
 * 
 * public void resume() { 
 * pauseLock.lock(); 
 * try { 
 * isPaused = false; 
 * unpaused.signalAll(); 
 * } finally { 
 * pauseLock.unlock(); 
 * } 
 * } 
 * }}</pre> 
 * 
 * @since 1.5 
 * @author Doug Lea 
 */

2、ThreadPoolExecutors 内部数据结构与构造方法详解

ThreadPoolExecutors的完整构造函数如下，从构造函数中能得出线程池最核心的属性

**\[java\]** [view plain][] [copy][view plain]

1. /\*\*
2. \* Creates a new \{@code ThreadPoolExecutor\} with the given initial
3. \* parameters.
4. \*
5. \* @param corePoolSize the number of threads to keep in the pool, even
6. \* if they are idle, unless \{@code allowCoreThreadTimeOut\} is set
7. \* @param maximumPoolSize the maximum number of threads to allow in the
8. \* pool
9. \* @param keepAliveTime when the number of threads is greater than
10. \* the core, this is the maximum time that excess idle threads
11. \* will wait for new tasks before terminating.
12. \* @param unit the time unit for the \{@code keepAliveTime\} argument
13. \* @param workQueue the queue to use for holding tasks before they are
14. \* executed. This queue will hold only the \{@code Runnable\}
15. \* tasks submitted by the \{@code execute\} method.
16. \* @param threadFactory the factory to use when the executor
17. \* creates a new thread
18. \* @param handler the handler to use when execution is blocked
19. \* because the thread bounds and queue capacities are reached
20. \* @throws IllegalArgumentException if one of the following holds: 
21. \* \{@code corePoolSize < 0\} 
22. \* \{@code keepAliveTime < 0\} 
23. \* \{@code maximumPoolSize <= 0\} 
24. \* \{@code maximumPoolSize < corePoolSize\}
25. \* @throws NullPointerException if \{@code workQueue\}
26. \* or \{@code threadFactory\} or \{@code handler\} is null
27. \*/
28. **public** ThreadPoolExecutor(**int** corePoolSize, 
29. **int** maximumPoolSize, 
30. **long** keepAliveTime, 
31. TimeUnit unit, 
32. BlockingQueue<Runnable> workQueue, 
33. ThreadFactory threadFactory, 
34. RejectedExecutionHandler handler) \{ 
35. **if** (corePoolSize < 0 || 
36. maximumPoolSize <= 0 || 
37. maximumPoolSize < corePoolSize || 
38. keepAliveTime < 0) 
39. **throw****new** IllegalArgumentException(); 
40. **if** (workQueue == **null** || threadFactory == **null** || handler == **null**) 
41. **throw****new** NullPointerException(); 
42. **this**.corePoolSize = corePoolSize; 
43. **this**.maximumPoolSize = maximumPoolSize; 
44. **this**.workQueue = workQueue; 
45. **this**.keepAliveTime = unit.toNanos(keepAliveTime); 
46. **this**.threadFactory = threadFactory; 
47. **this**.handler = handler; 
48. \}

2、ThreadPoolExecutors 内部数据结构与构造方法详解

1) corePoolSize 核心线程数

2）maximumPoolSize 最大线程数

3）keepAliveTime 线程保持激活状态的时间，如果为0，永远处于激活状态

4）unit ，keepAliveTime的单位

5）workQueue，线程池使用的队列

6）threadFactory 创建线程的工厂

7）handler 当队列已满，无更大线程处理任务时的拒绝任务的策略。

除了这些核心参数外，我觉得有必要再关注如下

8） HashSet<Worker> workers

9） completedTaskCount 完成的任务数

10） allowCoreThreadTimeOut，该值默认为false,也就是默认keepAliveTime不会生效。

3、核心源码分析

3.1 线程状态与几个基础方法设计原理

**\[java\]** [view plain][] [copy][view plain]

1. /\*\*
2. \* The main pool control state, ctl, is an atomic integer packing
3. \* two conceptual fields
4. \* workerCount, indicating the effective number of threads
5. \* runState, indicating whether running, shutting down etc
6. \*
7. \* In order to pack them into one int, we limit workerCount to
8. \* (2^29)-1 (about 500 million) threads rather than (2^31)-1 (2
9. \* billion) otherwise representable. If this is ever an issue in
10. \* the future, the variable can be changed to be an AtomicLong,
11. \* and the shift/mask constants below adjusted. But until the need
12. \* arises, this code is a bit faster and simpler using an int.
13. \*
14. \* The workerCount is the number of workers that have been
15. \* permitted to start and not permitted to stop. The value may be
16. \* transiently different from the actual number of live threads,
17. \* for example when a ThreadFactory fails to create a thread when
18. \* asked, and when exiting threads are still performing
19. \* bookkeeping before terminating. The user-visible pool size is
20. \* reported as the current size of the workers set.
21. \*
22. \* The runState provides the main lifecyle control, taking on values:
23. \*
24. \* RUNNING: Accept new tasks and process queued tasks
25. \* SHUTDOWN: Don't accept new tasks, but process queued tasks
26. \* STOP: Don't accept new tasks, don't process queued tasks,
27. \* and interrupt in-progress tasks
28. \* TIDYING: All tasks have terminated, workerCount is zero,
29. \* the thread transitioning to state TIDYING
30. \* will run the terminated() hook method
31. \* TERMINATED: terminated() has completed
32. \*
33. \* The numerical order among these values matters, to allow
34. \* ordered comparisons. The runState monotonically increases over
35. \* time, but need not hit each state. The transitions are:
36. \*
37. \* RUNNING -> SHUTDOWN
38. \* On invocation of shutdown(), perhaps implicitly in finalize()
39. \* (RUNNING or SHUTDOWN) -> STOP
40. \* On invocation of shutdownNow()
41. \* SHUTDOWN -> TIDYING
42. \* When both queue and pool are empty
43. \* STOP -> TIDYING
44. \* When pool is empty
45. \* TIDYING -> TERMINATED
46. \* When the terminated() hook method has completed
47. \*
48. \* Threads waiting in awaitTermination() will return when the
49. \* state reaches TERMINATED.
50. \*
51. \* Detecting the transition from SHUTDOWN to TIDYING is less
52. \* straightforward than you'd like because the queue may become
53. \* empty after non-empty and vice versa during SHUTDOWN state, but
54. \* we can only terminate if, after seeing that it is empty, we see
55. \* that workerCount is 0 (which sometimes entails a recheck -- see
56. \* below).
57. \*/
58. **private****final** AtomicInteger ctl = **new** AtomicInteger(ctlOf(RUNNING, 0)); 
59. **private****static****final****int** COUNT\_BITS = Integer.SIZE - 3; 
60. **private****static****final****int** CAPACITY = (1 << COUNT\_BITS) - 1; 
61. 
62. // runState is stored in the high-order bits
63. **private****static****final****int** RUNNING = -1 << COUNT\_BITS; 
64. **private****static****final****int** SHUTDOWN = 0 << COUNT\_BITS; 
65. **private****static****final****int** STOP = 1 << COUNT\_BITS; 
66. **private****static****final****int** TIDYING = 2 << COUNT\_BITS; 
67. **private****static****final****int** TERMINATED = 3 << COUNT\_BITS; 
68. 
69. // Packing and unpacking ctl
70. **private****static****int** runStateOf(**int** c) \{ **return** c & ~CAPACITY; \} 
71. **private****static****int** workerCountOf(**int** c) \{ **return** c & CAPACITY; \} 
72. **private****static****int** ctlOf(**int** rs, **int** wc) \{ **return** rs | wc; \} 
73. 
74. **private****static****boolean** isRunning(**int** c) \{ 
75. **return** c < SHUTDOWN; 
76. \}

2、ThreadPoolExecutors 内部数据结构与构造方法详解

相关源码解读：

不知大家有没有，为什么线程池的状态简单的定义为 -1,0,1,2,3不就得了，为什么还要用移位操作呢？

原来这样的，ThreadPool

ctl的这个变量的设计哲学是用int的高3位 + 29个0代表状态，，用高位000+低29位来表示线程池中工作线程的数量，太佩服了。

首先CAPACITY的值为workCount的最大容量，该值为 000 11111 11111111 11111111 11111111,29个1，

我们来看一下

private static int runStateOf(int c) \{ return c & ~CAPACITY; \}

用ctl里面的值与容量取反的方式获取状态值。由于CAPACITY的值为000 11111 11111111 11111111 11111111,那取反后为111 00000 00000000 00000000 00000000, 用 c 与 该值进行与运算，这样就直接保留了c的高三位，然后将c的低29位设置为0,这不就是线程池状态的存放规则吗，绝。

根据此方法，不难得出计算workCount的方法。

private static int ctlOf(int rs, int wc) \{ return rs | wc; \}

该方法，主要是用来更新运行状态的。确保工作线程数量不丢失。

线程池状态以及含义

RUNNING 运行态

SHUTDOWN 关闭，此时不接受新的任务，但继续处理队列中的任务。

STOP 停止，此时不接受新的任务，不处理队列中的任务，并中断正在执行的任务

TIDYING 所有的工作线程全部停止，并工作线程数量为0，将调用terminated方法，进入到TERMINATED

TERMINATED 终止状态

线程池默认状态 RUNNING

如果调用shutdwon() 方法，状态从 RUNNING ---> SHUTDOWN

如果调用shutdwonNow()方法，状态从RUUNING|SHUTDOWN--->STOP

SHUTDOWN ---> TIDYING

队列为空并且线程池空

STOP --> TIDYING

线程池为空

线程池设计原理：

1）线程池的工作线程为ThreadPoolExecutors的Worker线程，无论是submit还是executor方法中传入的Callable task,Runable参数，只是实现了Runnable接口，在线程池的调用过程，不会调用其start方法，只会调用Worker线程的start方法，然后在Worker线程的run方法中会调用入参的run方法。

2）众所周知，线程的生命周期在run方法运行结束后（包括异常退出）就结束。要想重复利用线程，我们要确保工作线程Worker的run方法运行在一个无限循环中，然后从任务队列中一个一个获取对象，如果任务队列为空，则阻塞，当然需要提供一些控制，结束无限循环，来销毁线程。在源码 runWorker方法与getTask来实现。

大概的实现思路是 如果getTask返回null,则该worker线程将被销毁。

那getTask在什么情况下会返回false呢？

1、如果线程池的状态为SHUTDOWN并且队列不为空

2、如果线程池的状态大于STOP

3、如果当前运行的线程数大于核心线程数，会返回null,已销毁该worker线程

对 keepAliveTime的理解，如果 allowCoreThreadTimeOut为真，那么keepAliveTime其实就是从任务队列获取任务等待的超时时间，也就是workerQueue.poll(keepALiveTime, TimeUnit. NANOSECONDS )

3.2 <T> FUture<T> submit(Callable<T> task) 方法详解

在看的代码的过程中，只要明白了上述基础方法，后面的代码看起来清晰可见，故，我只列出关键方法，大家可以浏览，应该不难。

**\[java\]** [view plain][] [copy][view plain]

1. /\*\*
2. \* Submits a value-returning task for execution and returns a
3. \* Future representing the pending results of the task. The
4. \* Future's <tt>get</tt> method will return the task's result upon
5. \* successful completion.
6. \*
7. \* 
8. \* If you would like to immediately block waiting
9. \* for a task, you can use constructions of the form
10. \* <tt>result = exec.submit(aCallable).get();</tt>
11. \*
12. \* Note: The \{@link Executors\} class includes a set of methods
13. \* that can convert some other common closure-like objects,
14. \* for example, \{@link java.security.PrivilegedAction\} to
15. \* \{@link Callable\} form so they can be submitted.
16. \*
17. \* @param task the task to submit
18. \* @return a Future representing pending completion of the task
19. \* @throws RejectedExecutionException if the task cannot be
20. \* scheduled for execution
21. \* @throws NullPointerException if the task is null
22. \*/
23. <T> Future<T> submit(Callable<T> task); 
24. 提交一个任务，并返回结构到Future,Future就是典型的Future设计模式，就是提交任务到线程池后，返回一个凭证，并直接返回，主线程继续执行，然后当线程池将任务运行完毕后，再将结果填充到凭证中，当主线程调用凭证future的get方法时，如果结果还未填充，则阻塞等待。 
25. 现将Callable与Future接口的源代码贴出来，然后重点分析submit方法的实现。 
26. **public****interface** Callable<V> \{ 
27. /\*\*
28. \* Computes a result, or throws an exception if unable to do so.
29. \*
30. \* @return computed result
31. \* @throws Exception if unable to compute a result
32. \*/
33. V call() **throws** Exception; 
34. \} 
35. 
36. **public****interface** Future<V> \{ 
37. 
38. /\*\*
39. \* Attempts to cancel execution of this task. This attempt will
40. \* fail if the task has already completed, has already been cancelled,
41. \* or could not be cancelled for some other reason. If successful,
42. \* and this task has not started when <tt>cancel</tt> is called,
43. \* this task should never run. If the task has already started,
44. \* then the <tt>mayInterruptIfRunning</tt> parameter determines
45. \* whether the thread executing this task should be interrupted in
46. \* an attempt to stop the task.
47. \*
48. \* After this method returns, subsequent calls to \{@link \#isDone\} will
49. \* always return <tt>true</tt>. Subsequent calls to \{@link \#isCancelled\}
50. \* will always return <tt>true</tt> if this method returned <tt>true</tt>.
51. \*
52. \* @param mayInterruptIfRunning <tt>true</tt> if the thread executing this
53. \* task should be interrupted; otherwise, in-progress tasks are allowed
54. \* to complete
55. \* @return <tt>false</tt> if the task could not be cancelled,
56. \* typically because it has already completed normally;
57. \* <tt>true</tt> otherwise
58. \*/
59. **boolean** cancel(**boolean** mayInterruptIfRunning); 
60. 
61. /\*\*
62. \* Returns <tt>true</tt> if this task was cancelled before it completed
63. \* normally.
64. \*
65. \* @return <tt>true</tt> if this task was cancelled before it completed
66. \*/
67. **boolean** isCancelled(); 
68. 
69. /\*\*
70. \* Returns <tt>true</tt> if this task completed.
71. \*
72. \* Completion may be due to normal termination, an exception, or
73. \* cancellation -- in all of these cases, this method will return
74. \* <tt>true</tt>.
75. \*
76. \* @return <tt>true</tt> if this task completed
77. \*/
78. **boolean** isDone(); 
79. 
80. /\*\*
81. \* Waits if necessary for the computation to complete, and then
82. \* retrieves its result.
83. \*
84. \* @return the computed result
85. \* @throws CancellationException if the computation was cancelled
86. \* @throws ExecutionException if the computation threw an
87. \* exception
88. \* @throws InterruptedException if the current thread was interrupted
89. \* while waiting
90. \*/
91. V get() **throws** InterruptedException, ExecutionException; 
92. 
93. /\*\*
94. \* Waits if necessary for at most the given time for the computation
95. \* to complete, and then retrieves its result, if available.
96. \*
97. \* @param timeout the maximum time to wait
98. \* @param unit the time unit of the timeout argument
99. \* @return the computed result
100. \* @throws CancellationException if the computation was cancelled
101. \* @throws ExecutionException if the computation threw an
102. \* exception
103. \* @throws InterruptedException if the current thread was interrupted
104. \* while waiting
105. \* @throws TimeoutException if the wait timed out
106. \*/
107. V get(**long** timeout, TimeUnit unit) 
108. **throws** InterruptedException, ExecutionException, TimeoutException; 
109. \} 
110. 
111. 现在开始探究submit的实现原理，该代码出自AbstractExecutorService中 
112. **public** Future<?> submit(Runnable task) \{ 
113. **if** (task == **null**) **throw****new** NullPointerException(); 
114. RunnableFuture<Void> ftask = newTaskFor(task, **null**); 
115. execute(ftask); 
116. **return** ftask; 
117. \} 
118. **protected** <T> RunnableFuture<T> newTaskFor(Callable<T> callable) \{ 
119. **return****new** FutureTask<T>(callable); 
120. \} 
121. 核心实现在ThreadPoolExecutor的execute方法 
122. /\*\*
123. \* Executes the given task sometime in the future. The task
124. \* may execute in a new thread or in an existing pooled thread.
125. \*
126. \* If the task cannot be submitted for execution, either because this
127. \* executor has been shutdown or because its capacity has been reached,
128. \* the task is handled by the current \{@code RejectedExecutionHandler\}.
129. \*
130. \* @param command the task to execute
131. \* @throws RejectedExecutionException at discretion of
132. \* \{@code RejectedExecutionHandler\}, if the task
133. \* cannot be accepted for execution
134. \* @throws NullPointerException if \{@code command\} is null
135. \*/
136. **public****void** execute(Runnable command) \{ 
137. **if** (command == **null**) 
138. **throw****new** NullPointerException(); 
139. /\*
140. \* Proceed in 3 steps:
141. \*
142. \* 1. If fewer than corePoolSize threads are running, try to
143. \* start a new thread with the given command as its first
144. \* task. The call to addWorker atomically checks runState and
145. \* workerCount, and so prevents false alarms that would add
146. \* threads when it shouldn't, by returning false.
147. \*
148. \* 2. If a task can be successfully queued, then we still need
149. \* to double-check whether we should have added a thread
150. \* (because existing ones died since last checking) or that
151. \* the pool shut down since entry into this method. So we
152. \* recheck state and if necessary roll back the enqueuing if
153. \* stopped, or start a new thread if there are none.
154. \*
155. \* 3. If we cannot queue task, then we try to add a new
156. \* thread. If it fails, we know we are shut down or saturated
157. \* and so reject the task.
158. \*/
159. **int** c = ctl.get(); 
160. **if** (workerCountOf(c) < corePoolSize) \{ // @1
161. **if** (addWorker(command, **true**)) // @2
162. **return**; 
163. c = ctl.get(); //@3
164. \} 
165. **if** (isRunning(c) && workQueue.offer(command)) \{ 
166. **int** recheck = ctl.get(); 
167. **if** (! isRunning(recheck) && remove(command)) 
168. reject(command); 
169. **else****if** (workerCountOf(recheck) == 0) 
170. addWorker(**null**, **false**); 
171. \} 
172. **else****if** (!addWorker(command, **false**)) 
173. reject(command); 
174. \} 
175. 代码@1,如果当前线程池中的线程数量小于核心线程数的话，尝试新增一个新的线程。所以我们把目光投入到addWorker方法中。 
176. 
177. addWorker源码详解： 
178. /\*\*
179. \* Checks if a new worker can be added with respect to current
180. \* pool state and the given bound (either core or maximum). If so,
181. \* the worker count is adjusted accordingly, and, if possible, a
182. \* new worker is created and started, running firstTask as its
183. \* first task. This method returns false if the pool is stopped or
184. \* eligible to shut down. It also returns false if the thread
185. \* factory fails to create a thread when asked. If the thread
186. \* creation fails, either due to the thread factory returning
187. \* null, or due to an exception (typically OutOfMemoryError in
188. \* Thread\#start), we roll back cleanly.
189. \*
190. \* @param firstTask the task the new thread should run first (or
191. \* null if none). Workers are created with an initial first task
192. \* (in method execute()) to bypass queuing when there are fewer
193. \* than corePoolSize threads (in which case we always start one),
194. \* or when the queue is full (in which case we must bypass queue).
195. \* Initially idle threads are usually created via
196. \* prestartCoreThread or to replace other dying workers.
197. \*
198. \* @param core if true use corePoolSize as bound, else
199. \* maximumPoolSize. (A boolean indicator is used here rather than a
200. \* value to ensure reads of fresh values after checking other pool
201. \* state).
202. \* @return true if successful
203. \*/
204. **private****boolean** addWorker(Runnable firstTask, **boolean** core) \{ 
205. retry: 
206. **for** (;;) \{ // @1 
207. **int** c = ctl.get(); 
208. **int** rs = runStateOf(c); // @2
209. 
210. // Check if queue empty only if necessary. 
211. **if** (rs >= SHUTDOWN && //@3 
212. ! (rs == SHUTDOWN && 
213. firstTask == **null** && 
214. ! workQueue.isEmpty())) 
215. **return****false**; 
216. 
217. **for** (;;) \{ //@4 
218. **int** wc = workerCountOf(c); 
219. **if** (wc >= CAPACITY || 
220. wc >= (core ? corePoolSize : maximumPoolSize)) //@5
221. **return****false**; 
222. **if** (compareAndIncrementWorkerCount(c)) 
223. **break** retry; // @6
224. c = ctl.get(); // Re-read ctl
225. **if** (runStateOf(c) != rs) 
226. **continue** retry; //@7
227. // else CAS failed due to workerCount change; retry inner loop
228. \} 
229. \} 
230. 
231. **boolean** workerStarted = **false**; 
232. **boolean** workerAdded = **false**; 
233. Worker w = **null**; 
234. **try** \{ 
235. **final** ReentrantLock mainLock = **this**.mainLock; 
236. w = **new** Worker(firstTask); 
237. **final** Thread t = w.thread; 
238. **if** (t != **null**) \{ 
239. mainLock.lock(); // @8 
240. **try** \{ 
241. // Recheck while holding lock.
242. // Back out on ThreadFactory failure or if
243. // shut down before lock acquired.
244. **int** c = ctl.get(); 
245. **int** rs = runStateOf(c); 
246. 
247. **if** (rs < SHUTDOWN || 
248. (rs == SHUTDOWN && firstTask == **null**)) \{ 
249. **if** (t.isAlive()) // precheck that t is startable
250. **throw****new** IllegalThreadStateException(); 
251. workers.add(w); 
252. **int** s = workers.size(); 
253. **if** (s > largestPoolSize) 
254. largestPoolSize = s; 
255. workerAdded = **true**; 
256. \} 
257. \} **finally** \{ 
258. mainLock.unlock(); 
259. \} 
260. **if** (workerAdded) \{ 
261. t.start(); // 运行线程 // @9
262. workerStarted = **true**; 
263. \} //@8 end
264. \} 
265. \} **finally** \{ 
266. **if** (! workerStarted) 
267. addWorkerFailed(w); // 增加工作线程失败
268. \} 
269. **return** workerStarted; 
270. \} 
271. 代码@1，外层循环（自旋模式） 
272. 代码@2，获取线程池的运行状态 
273. 代码@3，这里的判断条件，为什么不直接写 **if**(rs >= SHUTDOWN) **return****false**;而要加第二个条件，目前不明白，等在了解到参数firstTask在什么情况下为空。在这里，我们目前只要知道，只有线程池的状态为 RUNNING时，线程池才接收新的任务，去增加工作线程。 
274. 代码@4，内层循环，主要的目的就是利用CAS增加一个线程数量。 
275. 代码@5，判断当前线程池的数量，如果数量达到规定的数量，则直接返回**false**,添加工作线程失败。 
276. 代码@6，如果修改线程数量成功，则跳出循环，开始创建工作线程。 
277. 代码@7，如果修改线程数量不成功(CAS)有两种情况：1、线程数量变化，重试则好，2，如果线程的运行状态变化，则重新开始外层循环，重新判断addWork流程。 
278. 代码@8，在锁mainLock的保护下，完成 workers （HashSet）的维护。 
279. 接着分析一下代码@9，启动线程，执行关键的方法 runWorker方法： 
280. /\*\*
281. \* Main worker run loop. Repeatedly gets tasks from queue and
282. \* executes them, while coping with a number of issues:
283. \*
284. \* 1. We may start out with an initial task, in which case we
285. \* don't need to get the first one. Otherwise, as long as pool is
286. \* running, we get tasks from getTask. If it returns null then the
287. \* worker exits due to changed pool state or configuration
288. \* parameters. Other exits result from exception throws in
289. \* external code, in which case completedAbruptly holds, which
290. \* usually leads processWorkerExit to replace this thread.
291. \*
292. \* 2. Before running any task, the lock is acquired to prevent
293. \* other pool interrupts while the task is executing, and
294. \* clearInterruptsForTaskRun called to ensure that unless pool is
295. \* stopping, this thread does not have its interrupt set.
296. \*
297. \* 3. Each task run is preceded by a call to beforeExecute, which
298. \* might throw an exception, in which case we cause thread to die
299. \* (breaking loop with completedAbruptly true) without processing
300. \* the task.
301. \*
302. \* 4. Assuming beforeExecute completes normally, we run the task,
303. \* gathering any of its thrown exceptions to send to
304. \* afterExecute. We separately handle RuntimeException, Error
305. \* (both of which the specs guarantee that we trap) and arbitrary
306. \* Throwables. Because we cannot rethrow Throwables within
307. \* Runnable.run, we wrap them within Errors on the way out (to the
308. \* thread's UncaughtExceptionHandler). Any thrown exception also
309. \* conservatively causes thread to die.
310. \*
311. \* 5. After task.run completes, we call afterExecute, which may
312. \* also throw an exception, which will also cause thread to
313. \* die. According to JLS Sec 14.20, this exception is the one that
314. \* will be in effect even if task.run throws.
315. \*
316. \* The net effect of the exception mechanics is that afterExecute
317. \* and the thread's UncaughtExceptionHandler have as accurate
318. \* information as we can provide about any problems encountered by
319. \* user code.
320. \*
321. \* @param w the worker
322. \*/
323. **final****void** runWorker(Worker w) \{ 
324. Thread wt = Thread.currentThread(); 
325. Runnable task = w.firstTask; 
326. w.firstTask = **null**; 
327. w.unlock(); // allow interrupts
328. **boolean** completedAbruptly = **true**; 
329. **try** \{ 
330. **while** (task != **null** || (task = getTask()) != **null**) \{ 
331. w.lock(); 
332. // If pool is stopping, ensure thread is interrupted;
333. // if not, ensure thread is not interrupted. This
334. // requires a recheck in second case to deal with
335. // shutdownNow race while clearing interrupt
336. **if** ((runStateAtLeast(ctl.get(), STOP) || 
337. (Thread.interrupted() && 
338. runStateAtLeast(ctl.get(), STOP))) && 
339. !wt.isInterrupted()) 
340. wt.interrupt(); 
341. **try** \{ 
342. beforeExecute(wt, task); 
343. Throwable thrown = **null**; 
344. **try** \{ 
345. task.run(); 
346. \} **catch** (RuntimeException x) \{ 
347. thrown = x; **throw** x; 
348. \} **catch** (Error x) \{ 
349. thrown = x; **throw** x; 
350. \} **catch** (Throwable x) \{ 
351. thrown = x; **throw****new** Error(x); 
352. \} **finally** \{ 
353. afterExecute(task, thrown); 
354. \} 
355. \} **finally** \{ 
356. task = **null**; 
357. w.completedTasks++; 
358. w.unlock(); 
359. \} 
360. \} 
361. completedAbruptly = **false**; 
362. \} **finally** \{ 
363. processWorkerExit(w, completedAbruptly); 
364. \} 
365. \} 
366. 
367. 
368. /\*\*
369. \* Performs blocking or timed wait for a task, depending on
370. \* current configuration settings, or returns null if this worker
371. \* must exit because of any of:
372. \* 1. There are more than maximumPoolSize workers (due to
373. \* a call to setMaximumPoolSize).
374. \* 2. The pool is stopped.
375. \* 3. The pool is shutdown and the queue is empty.
376. \* 4. This worker timed out waiting for a task, and timed-out
377. \* workers are subject to termination (that is,
378. \* \{@code allowCoreThreadTimeOut || workerCount > corePoolSize\})
379. \* both before and after the timed wait.
380. \*
381. \* @return task, or null if the worker must exit, in which case
382. \* workerCount is decremented
383. \*/
384. **private** Runnable getTask() \{ 
385. **boolean** timedOut = **false**; // Did the last poll() time out?
386. 
387. retry: 
388. **for** (;;) \{ 
389. **int** c = ctl.get(); 
390. **int** rs = runStateOf(c); 
391. 
392. // Check if queue empty only if necessary.
393. **if** (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) \{ 
394. decrementWorkerCount(); 
395. **return****null**; 
396. \} 
397. 
398. **boolean** timed; // Are workers subject to culling?
399. 
400. **for** (;;) \{ 
401. **int** wc = workerCountOf(c); 
402. timed = allowCoreThreadTimeOut || wc > corePoolSize; 
403. 
404. **if** (wc <= maximumPoolSize && ! (timedOut && timed)) 
405. **break**; 
406. **if** (compareAndDecrementWorkerCount(c)) 
407. **return****null**; 
408. c = ctl.get(); // Re-read ctl
409. **if** (runStateOf(c) != rs) 
410. **continue** retry; 
411. // else CAS failed due to workerCount change; retry inner loop
412. \} 
413. 
414. **try** \{ 
415. Runnable r = timed ? 
416. workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : 
417. workQueue.take(); 
418. **if** (r != **null**) 
419. **return** r; 
420. timedOut = **true**; 
421. \} **catch** (InterruptedException retry) \{ 
422. timedOut = **false**; 
423. \} 
424. \} 
425. \} 
426.

[Executor]: http://www.codeceo.com/article/java-executor-learning.html
[20170819200608739]: /images/20220610/8884247122994f068198922aee47333d.png
[20170819200619605]: /images/20220610/e62057ca1afb4cd6a01f0c28ee562d36.png
[view plain]: http://blog.csdn.net/prestigeding/article/details/53929713#