线程池原理（ThreadPoolExecutor）

一、创建一个线程池对象

使用Executors创建一个线程池常用的方法如下：

1、newFixedThreadPool()

说明：初始化一个指定线程数的线程池，其中 corePoolSize == maxiPoolSize，使用 LinkedBlockingQuene 作为阻塞队列

特点：即使当线程池没有可执行任务时，也不会释放线程。

2、newCachedThreadPool()

说明：初始化一个可以缓存线程的线程池，默认缓存60s，线程池的线程数可达到 Integer.MAX_VALUE，即 2147483647，内部使用 SynchronousQueue 作为阻塞队列；

特点：在没有任务执行时，当线程的空闲时间超过 keepAliveTime，会自动释放线程资源；当提交新任务时，如果没有空闲线程，则创建新线程执行任务，会导致一定的系统开销；因此，使用时要注意控制并发的任务数，防止因创建大量的线程导致而降低性能。

3、newSingleThreadExecutor()

说明：初始化只有一个线程的线程池，内部使用 LinkedBlockingQueue 作为阻塞队列。

特点：如果该线程异常结束，会重新创建一个新的线程继续执行任务，唯一的线程可以保证所提交任务的顺序执行

4、newScheduledThreadPool()

特点：初始化的线程池可以在指定的时间内周期性的执行所提交的任务，在实际的业务场景中可以使用该线程池定期的同步数据。

线程池的构造方法如下：

    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;}

默认的线程工厂,Executors类的DefaultThreadFactory方法：

    /*** Thread factory capturing access control context and class loader*/static class PrivilegedThreadFactory extends DefaultThreadFactory {private final AccessControlContext acc;private final ClassLoader ccl;PrivilegedThreadFactory() {super();SecurityManager sm = System.getSecurityManager();if (sm != null) {// Calls to getContextClassLoader from this class// never trigger a security check, but we check// whether our callers have this permission anyways.sm.checkPermission(SecurityConstants.GET_CLASSLOADER_PERMISSION);// Fail fastsm.checkPermission(new RuntimePermission("setContextClassLoader"));}this.acc = AccessController.getContext();this.ccl = Thread.currentThread().getContextClassLoader();}public Thread newThread(final Runnable r) {return super.newThread(new Runnable() {public void run() {AccessController.doPrivileged(new PrivilegedAction<Void>() {public Void run() {Thread.currentThread().setContextClassLoader(ccl);r.run();return null;}}, acc);}});}}

默认的拒绝策略

    /*** The default rejected execution handler*/private static final RejectedExecutionHandler defaultHandler =new AbortPolicy();/*** A handler for rejected tasks that throws a* {@code RejectedExecutionException}.*/public static class AbortPolicy implements RejectedExecutionHandler {/*** Creates an {@code AbortPolicy}.*/public AbortPolicy() { }/*** Always throws RejectedExecutionException.** @param r the runnable task requested to be executed* @param e the executor attempting to execute this task* @throws RejectedExecutionException always.*/public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {throw new RejectedExecutionException("Task " + r.toString() +" rejected from " +e.toString());}}

二、添加任务到线程池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.*/// 获取ctl对应的int值。该int值保存了"线程池中任务的数量"和"线程池状态"信息int c = ctl.get();// 当线程池中的任务数量 < "核心池大小"时，即线程池中少于corePoolSize个任务。// 则通过addWorker(command,true)新建一个线程，并将任务(command)添加到该线程中；然后，启动该线程从而执行任务。if (workerCountOf(c) < corePoolSize) {if (addWorker(command, true))return;c = ctl.get();}// 当线程池中的任务数量 >= "核心池大小"时，// 而且，"线程池处于允许状态"时，则尝试将任务添加到阻塞队列中。if (isRunning(c) && workQueue.offer(command)) {int recheck = ctl.get();// 再次确认“线程池状态”，若线程池异常终止了，则删除任务；然后通过reject()执行相应的拒绝策略的内容。if (! isRunning(recheck) && remove(command))reject(command);// 否则，如果"线程池中任务数量"为0，则通过addWorker(null,false)尝试新建一个线程，新建线程对应的任务为null。else if (workerCountOf(recheck) == 0)addWorker(null, false);}// 通过addWorker(command,false)新建一个线程，并将任务(command)添加到该线程中；然后，启动该线程从而执行任务。// 如果addWorker(command, false)执行失败，则通过reject()执行相应的拒绝策略的内else if (!addWorker(command, false))reject(command);}

可以看到execute方法流程如下：（可参照英文注释）

线程池中的任务数量小于核心池大小时，新建线程；
线程池中的任务数量大于等于核心池大小时，添加到任务队列中；
若不能将任务添加到队列中，则尝试新建一个线程，失败的话执行拒绝策略

下面来看addWorker方法：

    private boolean addWorker(Runnable firstTask, boolean core) {retry:// 这部分是对状态进行检查，更新ctl的值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 ctlif (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 {final ReentrantLock mainLock = this.mainLock;// 新建Worker，并且指定firstTask为Worker的第一个任务w = new Worker(firstTask);final Thread t = w.thread;if (t != null) {mainLock.lock();try {// Recheck while holding lock.// Back out on ThreadFactory failure or if// shut down before lock acquired.int c = ctl.get();int rs = runStateOf(c);if (rs < SHUTDOWN ||(rs == SHUTDOWN && firstTask == null)) {if (t.isAlive()) // precheck that t is startablethrow 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;}

三、关闭线程池

调用线程池的shutdown方法可以关闭线程池：

    /*** 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();}

四、线程池状态

线程池使用了原子类的Integer的对象来标识线程池的状态：

    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 bitsprivate 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;

可以看到线程池公有以下五种状态：

RUNNING：线程池处在RUNNING状态时，能够接收新任务，以及对已添加的任务进行处理。线程池的初始化状态是RUNNING。
SHUTDOWN：调用线程池的shutdown()接口时，线程池由RUNNING转到SHUTDOWN。线程池处在SHUTDOWN状态时，不接收新任务，但能处理已添加的任务。
STOP：调用线程池的shutdownNow()接口时，线程池由(RUNNING or SHUTDOWN ) -> STOP。线程池处在STOP状态时，不接收新任务，不处理已添加的任务，并且会中断正在处理的任务。
TIDYING：当线程池在SHUTDOWN状态下，阻塞队列为空并且线程池中执行的任务也为空时，就会由 SHUTDOWN -> TIDYING。当线程池在STOP状态下，线程池中执行的任务为空时，就会由STOP -> TIDYING。当所有的任务已终止，ctl记录的"任务数量"为0，线程池会变为TIDYING状态。当线程池变为TIDYING状态时，会执行钩子函数terminated()。terminated()在ThreadPoolExecutor类中是空的，若用户想在线程池变为TIDYING时，进行相应的处理；可以通过重载terminated()函数来实现。：
TERMINATED：线程池处在TIDYING状态时，执行完terminated()之后，就会由TIDYING->TERMINATED。线程池彻底终止，就变成TERMINATED状态。

五、线程池拒绝策略

线程池共包括4种拒绝策略：

AbortPolicy：当任务添加到线程池中被拒绝时，它将抛出 RejectedExecutionException 异常。
CallerRunsPolicy：当任务添加到线程池中被拒绝时，会在线程池当前正在运行的Thread线程池中处理被拒绝的任务。
DiscardOldestPolicy：当任务添加到线程池中被拒绝时，线程池会放弃等待队列中最旧的未处理任务，然后将被拒绝的任务添加到等待队列中。
DiscardPolicy：当任务添加到线程池中被拒绝时，线程池将丢弃被拒绝的任务。