java并发访问线程池_Java并发（三）线程池原理

Java中的线程池是运用场景最多的并发框架&＃xff0c;几乎所有需要异步或并发执行任务的程序都可以使用线程池。在开发过程中&＃xff0c;合理地使用线程池能够带来3个好处。

1. 降低资源消耗。通过重复利用已创建的线程降低线程创建和销毁造成的消耗&＃xff1b;

2. 提高响应速度。当任务到达时&＃xff0c;任务可以不需要等到线程创建就能立即执行&＃xff1b;

3. 提高线程的可管理性。线程是稀缺资源&＃xff0c;如果无限制地创建&＃xff0c;不仅会消耗系统资源&＃xff0c;还会降低系统的稳定性&＃xff0c;使用线程池可以进行统一分配、调优和监控。但是&＃xff0c;要做到合理利用线程池&＃xff0c;必须对其实现原理了如指掌。

线程池实现原理

当向线程池提交一个任务之后&＃xff0c;线程池是如何处理这个任务的呢&＃xff1f;本节来看一下线程池的主要处理流程&＃xff0c;处理流程图如下图所示&＃xff1a;

从图中可以看出&＃xff0c;当提交一个新任务到线程池时&＃xff0c;线程池的处理流程如下。

1. 线程池判断核心线程池里的线程是否都在执行任务。如果不是&＃xff0c;则创建一个新的工作线程来执行任务。如果核心线程池里的线程都在执行任务&＃xff0c;则进入下个流程。

2. 线程池判断工作队列是否已经满。如果工作队列没有满&＃xff0c;则将新提交的任务存储在这个工作队列里。如果工作队列满了&＃xff0c;则进入下个流程。

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

ThreadPoolExecutor执行execute()方法的示意图&＃xff0c;如下图所示

ThreadPoolExecutor执行execute方法分下面4种情况。

1)如果当前运行的线程少于corePoolSize&＃xff0c;则创建新线程来执行任务(注意&＃xff0c;执行这一步骤需要获取全局锁)。

2)如果运行的线程等于或多于corePoolSize&＃xff0c;则将任务加入BlockingQueue。

3)如果无法将任务加入BlockingQueue(队列已满)&＃xff0c;则创建新的线程来处理任务(注意&＃xff0c;执行这一步骤需要获取全局锁)。

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

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

源码分析&＃xff1a;上面的流程分析让我们很直观地了解了线程池的工作原理&＃xff0c;让我们再通过源代码来看看是如何实现的。

一、变量

public class ThreadPoolExecutor extendsAbstractExecutorService {/*** 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&＃39;t accept new tasks, but process queued tasks

* STOP: Don&＃39;t accept new tasks, don&＃39;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&＃39;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).*/

/*** ctl 为原子类型的变量, 有两个概念

* workerCount, 表示有效的线程数

* runState, 表示线程状态, 是否正在运行, 关闭等*/

private final AtomicInteger ctl &＃61; new AtomicInteger(ctlOf(RUNNING, 0));//29

private static final int COUNT_BITS &＃61; Integer.SIZE - 3;//容量 2²⁹-1

private static final int CAPACITY &＃61; (1 <

private static final int RUNNING &＃61; -1 <

private static final int SHUTDOWN &＃61; 0 <

private static final int STOP &＃61; 1 <

private static final int TIDYING &＃61; 2 <

private static final int TERMINATED &＃61; 3 <

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

... ...

}

ctl 是对线程池的运行状态和线程池中有效线程的数量进行控制的一个字段&＃xff0c; 它包含两部分的信息&＃xff1a;线程池的运行状态 (runState) 和线程池内有效线程的数量 (workerCount)&＃xff0c;这里可以看到&＃xff0c;使用了Integer类型来保存&＃xff0c;高3位保存runState&＃xff0c;低29位保存workerCount。COUNT_BITS 就是29&＃xff0c;CAPACITY就是1左移29位减1(29个1)&＃xff0c;这个常量表示workerCount的上限值&＃xff0c;大约是5亿。

下面再介绍下线程池的运行状态&＃xff0c;线程池一共有五种状态&＃xff0c;分别是&＃xff1a;

状态

描述

RUNNING

能接受新提交的任务&＃xff0c;并且也能处理阻塞队列中的任务

SHUTDOWN

关闭状态&＃xff0c;不再接受新提交的任务&＃xff0c;但却可以继续处理阻塞队列中已保存的任务。在线程池处于 RUNNING 状态时&＃xff0c;调用 shutdown()方法会使线程池进入到该状态。(finalize() 方法在执行过程中也会调用shutdown()方法进入该状态)

STOP

不能接受新任务&＃xff0c;也不处理队列中的任务&＃xff0c;会中断正在处理任务的线程。在线程池处于 RUNNING 或 SHUTDOWN 状态时&＃xff0c;调用 shutdownNow() 方法会使线程池进入到该状态

TIDYING

如果所有的任务都已终止了&＃xff0c;workerCount (有效线程数) 为0&＃xff0c;线程池进入该状态后会调用 terminated() 方法进入TERMINATED 状态

TERMINATED

在terminated() 方法执行完后进入该状态&＃xff0c;默认terminated()方法中什么也没有做

进入TERMINATED的条件如下&＃xff1a;

线程池不是RUNNING状态&＃xff1b;

线程池状态不是TIDYING状态或TERMINATED状态&＃xff1b;

如果线程池状态是SHUTDOWN并且workerQueue为空&＃xff1b;

workerCount为0&＃xff1b;

设置TIDYING状态成功。

下图为线程池的状态转换过程&＃xff1a;

计算线程的几个方法&＃xff1a;

方法

描述

runStateOf

获取运行状态

workerCountOf

获取活动线程数

ctlOf

获取运行状态和活动线程数的值

二、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 {&＃64;codeRejectedExecutionHandler}.

*

*&＃64;paramcommand the task to execute

*&＃64;throwsRejectedExecutionException at discretion of

* {&＃64;codeRejectedExecutionHandler}, if the task

* cannot be accepted for execution

*&＃64;throwsNullPointerException if {&＃64;codecommand} is null*/

public voidexecute(Runnable command) {//空则抛出异常

if (command &＃61;&＃61; null)throw newNullPointerException();/** 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&＃39;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.*/

/** 获取当前线程池的状态

* clt记录着runState和workerCount

**/

int c &＃61;ctl.get();/** 计算工作线程数 并判断是否小于核心线程数

* workerCountOf方法取出低29位的值&＃xff0c;表示当前活动的线程数;

* 如果当前活动线程数小于corePoolSize&＃xff0c;则新建一个线程放入线程池中;

* 并把任务添加到该线程中。

**/

if (workerCountOf(c)

/** addWorker中的第二个参数表示限制添加线程的数量是根据corePoolSize来判断还是maximumPoolSize来判断&＃xff1b;

* 如果为true&＃xff0c;根据corePoolSize来判断&＃xff1b;

* 如果为false&＃xff0c;则根据maximumPoolSize来判断*/

if (addWorker(command, true))return;//提交失败再次获取当前状态

c &＃61;ctl.get();

}//判断线程状态, 并插入队列, 失败则移除

/** 如果当前线程池是运行状态并且任务添加到队列成功*/

if (isRunning(c) &&workQueue.offer(command)) {//再次获取状态

int recheck &＃61;ctl.get();//如果状态不是RUNNING, 并移除失败

/** 再次判断线程池的运行状态&＃xff0c;如果不是运行状态&＃xff0c;由于之前已经把command添加到workQueue中了&＃xff0c;

* 这时需要移除该command

* 执行过后通过handler使用拒绝策略对该任务进行处理&＃xff0c;整个方法返回*/

if (! isRunning(recheck) &&remove(command))//调用拒绝策略

reject(command);//如果工作线程为0 则调用 addWorker

/** 获取线程池中的有效线程数&＃xff0c;如果数量是0&＃xff0c;则执行addWorker方法

* 这里传入的参数表示&＃xff1a;

* 1. 第一个参数为null&＃xff0c;表示在线程池中创建一个线程&＃xff0c;但不去启动&＃xff1b;

* 2. 第二个参数为false&＃xff0c;将线程池的有限线程数量的上限设置为maximumPoolSize&＃xff0c;添加线程时根据maximumPoolSize来判断&＃xff1b;

* 如果判断workerCount大于0&＃xff0c;则直接返回&＃xff0c;在workQueue中新增的command会在将来的某个时刻被执行。*/

else if (workerCountOf(recheck) &＃61;&＃61; 0)

addWorker(null, false);

}//提交任务失败 走拒绝策略

/** 如果执行到这里&＃xff0c;有两种情况&＃xff1a;

* 1. 线程池已经不是RUNNING状态&＃xff1b;

* 2. 线程池是RUNNING状态&＃xff0c;但workerCount >&＃61; corePoolSize并且workQueue已满。

* 这时&＃xff0c;再次调用addWorker方法&＃xff0c;但第二个参数传入为false&＃xff0c;将线程池的有限线程数量的上限设置为maximumPoolSize&＃xff1b;

* 如果失败则拒绝该任务*/

else if (!addWorker(command, false))

reject(command);

}

简单来说&＃xff0c;在执行 execute() 方法时如果状态一直是RUNNING时&＃xff0c;的执行过程如下&＃xff1a;

如果workerCount

如果workerCount >&＃61; corePoolSize&＃xff0c;且线程池内的阻塞队列未满&＃xff0c;则将任务添加到该阻塞队列中&＃xff1b;

如果workerCount >&＃61; corePoolSize && workerCount

如果workerCount >&＃61; maximumPoolSize&＃xff0c;并且线程池内的阻塞队列已满, 则根据拒绝策略来处理该任务, 默认的处理方式是直接抛异常。

这里要注意一下 addWorker(null, false) &＃xff0c;也就是创建一个线程&＃xff0c;但并没有传入任务&＃xff0c;因为任务已经被添加到workQueue中了&＃xff0c;所以worker在执行的时候&＃xff0c;会直接从workQueue中获取任务。所以&＃xff0c;在 workerCountOf(recheck) &＃61;&＃61; 0 时执行 addWorker(null, false) 也是为了保证线程池在RUNNING状态下必须要有一个线程来执行任务。

execute方法执行流程如下&＃xff1a;

三、addWorker方法

addWorker方法的主要工作是在线程池中创建一个新的线程并执行&＃xff0c;firstTask参数 用于指定新增的线程执行的第一个任务&＃xff0c;core参数为true表示在新增线程时会判断当前活动线程数是否少于corePoolSize&＃xff0c;false表示新增线程前需要判断当前活动线程数是否少于maximumPoolSize&＃xff0c;代码如下&＃xff1a;

/*** 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.

*

*&＃64;paramfirstTask 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.

*

*&＃64;paramcore 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).

*&＃64;returntrue if successful*/

/*** 检查任务是否可以提交*/

private boolean addWorker(Runnable firstTask, booleancore) {

retry://外层循环

for(;;) {//获取运行状态

int c &＃61;ctl.get();int rs &＃61;runStateOf(c);/** 这个if判断

* 如果rs >&＃61; SHUTDOWN&＃xff0c;则表示此时不再接收新任务&＃xff1b;

* 接着判断以下3个条件&＃xff0c;只要有1个不满足&＃xff0c;则返回false&＃xff1a;

* 1. rs &＃61;&＃61; SHUTDOWN&＃xff0c;这时表示关闭状态&＃xff0c;不再接受新提交的任务&＃xff0c;但却可以继续处理阻塞队列中已保存的任务

* 2. firsTask为空

* 3. 阻塞队列不为空

*

* 首先考虑rs &＃61;&＃61; SHUTDOWN的情况

* 这种情况下不会接受新提交的任务&＃xff0c;所以在firstTask不为空的时候会返回false&＃xff1b;

* 然后&＃xff0c;如果firstTask为空&＃xff0c;并且workQueue也为空&＃xff0c;则返回false&＃xff0c;

* 因为队列中已经没有任务了&＃xff0c;不需要再添加线程了*/

//Check if queue empty only if necessary. 检查线程池是否关闭

if (rs >&＃61; SHUTDOWN &&

! (rs &＃61;&＃61; SHUTDOWN &&firstTask&＃61;&＃61; null &&

!workQueue.isEmpty()))return false;//内层循环

for(;;) {//获取线程数

int wc &＃61;workerCountOf(c);//工作线程大于容量 或者大于 核心或最大线程数

/** 如果wc超过CAPACITY&＃xff0c;也就是ctl的低29位的最大值(二进制是29个1)&＃xff0c;返回false&＃xff1b;

* 这里的core是addWorker方法的第二个参数&＃xff0c;如果为true表示根据corePoolSize来比较&＃xff0c;

* 如果为false则根据maximumPoolSize来比较。*/

if (wc >&＃61; CAPACITY ||wc>&＃61; (core ?corePoolSize : maximumPoolSize))return false;//CAS 线程数增加, 成功则调到外层循环

/** 尝试增加workerCount&＃xff0c;如果成功&＃xff0c;则跳出第一个for循环*/

if(compareAndIncrementWorkerCount(c))breakretry;//如果增加workerCount失败&＃xff0c;则重新获取ctl的值

c &＃61; ctl.get(); //Re-read ctl//如果当前的运行状态不等于rs&＃xff0c;说明状态已被改变&＃xff0c;返回第一个for循环继续执行

if (runStateOf(c) !&＃61;rs)continueretry;//else CAS failed due to workerCount change; retry inner loop

}

}/*** 创建新worker 开始新线程*/

boolean workerStarted &＃61; false;boolean workerAdded &＃61; false;

Worker w&＃61; null;try{//根据firstTask来创建Worker对象

w &＃61; newWorker(firstTask);//每一个Worker对象都会创建一个线程

final Thread t &＃61;w.thread;if (t !&＃61; null) {final ReentrantLock mainLock &＃61; this.mainLock;//加锁

mainLock.lock();try{//Recheck while holding lock.//Back out on ThreadFactory failure or if//shut down before lock acquired.

int rs &＃61;runStateOf(ctl.get());/** rs

* 如果rs是RUNNING状态或者rs是SHUTDOWN状态并且firstTask为null&＃xff0c;向线程池中添加线程。

* 因为在SHUTDOWN时不会在添加新的任务&＃xff0c;但还是会执行workQueue中的任务*/

if (rs

if (t.isAlive()) //precheck that t is startable

throw newIllegalThreadStateException();//设置包含池中的所有工作线程。仅在持有mainLock时访问 workers是 HashSet 集合

workers.add(w);int s &＃61;workers.size();//设置池最大大小, 并将 workerAdded设置为 true//largestPoolSize记录着线程池中出现过的最大线程数量

if (s >largestPoolSize)

largestPoolSize&＃61;s;

workerAdded&＃61; true;

}

}finally{//解锁

mainLock.unlock();

}//添加成功 开始启动线程 并将 workerStarted 设置为 true

if(workerAdded) {//启动线程

t.start();

workerStarted&＃61; true;

}

}

}finally{//启动线程失败

if (!workerStarted)

addWorkerFailed(w);

}returnworkerStarted;

}

注意一下这里的 t.start() 这个语句&＃xff0c;启动时会调用Worker类中的run方法&＃xff0c;Worker本身实现了Runnable接口&＃xff0c;所以一个Worker类型的对象也是一个线程。

四、Worker类

工作线程&＃xff1a;线程池创建线程时&＃xff0c;会将线程封装成工作线程Worker&＃xff0c;接下来看看源码&＃xff1a;

private final classWorkerextendsAbstractQueuedSynchronizerimplementsRunnable

{/*** This class will never be serialized, but we provide a

* serialVersionUID to suppress a javac warning.*/

private static final long serialVersionUID &＃61; 6138294804551838833L;/**Thread this worker is running in. Null if factory fails.*/

finalThread thread;/**Initial task to run. Possibly null.*/Runnable firstTask;/**Per-thread task counter*/

volatile longcompletedTasks;/*** Creates with given first task and thread from ThreadFactory.

*&＃64;paramfirstTask the first task (null if none)*/Worker(Runnable firstTask) {

setState(-1); //inhibit interrupts until runWorker

this.firstTask &＃61;firstTask;this.thread &＃61; getThreadFactory().newThread(this);

}/**Delegates main run loop to outer runWorker*/

public voidrun() {

runWorker(this);

}//Lock methods//

//The value 0 represents the unlocked state.//The value 1 represents the locked state.

protected booleanisHeldExclusively() {return getState() !&＃61; 0;

}protected boolean tryAcquire(intunused) {if (compareAndSetState(0, 1)) {

setExclusiveOwnerThread(Thread.currentThread());return true;

}return false;

}protected boolean tryRelease(intunused) {

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() { returnisHeldExclusively(); }voidinterruptIfStarted() {

Thread t;if (getState() >&＃61; 0 && (t &＃61; thread) !&＃61; null && !t.isInterrupted()) {try{

t.interrupt();

}catch(SecurityException ignore) {

}

}

}

}

Worker类继承了AQS&＃xff0c;并实现了Runnable接口&＃xff0c;注意其中的 firstTask 和 thread 属性&＃xff1a; firstTask 用它来保存传入的任务&＃xff1b; thread 是在调用构造方法时通过 ThreadFactory 来创建的线程&＃xff0c;是用来处理任务的线程。

在调用构造方法时&＃xff0c;需要把任务传入&＃xff0c;这里通过 getThreadFactory().newThread(this)来新建一个线程&＃xff0c; newThread 方法传入的参数是this&＃xff0c;因为Worker本身继承了Runnable接口&＃xff0c;也就是一个线程&＃xff0c;所以一个Worker对象在启动的时候会调用Worker类中的run方法。

Worker继承了AQS&＃xff0c;使用AQS来实现独占锁的功能。为什么不使用ReentrantLock来实现呢&＃xff1f;可以看到tryAcquire方法&＃xff0c;它是不允许重入的&＃xff0c;而ReentrantLock是允许重入的&＃xff1a;

lock方法一旦获取了独占锁&＃xff0c;表示当前线程正在执行任务中&＃xff1b;

如果正在执行任务&＃xff0c;则不应该中断线程&＃xff1b;

如果该线程现在不是独占锁的状态&＃xff0c;也就是空闲的状态&＃xff0c;说明它没有在处理任务&＃xff0c;这时可以对该线程进行中断&＃xff1b;

线程池在执行shutdown方法或tryTerminate方法时会调用interruptIdleWorkers方法来中断空闲的线程&＃xff0c;interruptIdleWorkers方法会使用tryLock方法来判断线程池中的线程是否是空闲状态&＃xff1b;

之所以设置为不可重入&＃xff0c;是因为我们不希望任务在调用像setCorePoolSize这样的线程池控制方法时重新获取锁。如果使用ReentrantLock&＃xff0c;它是可重入的&＃xff0c;这样如果在任务中调用了如setCorePoolSize这类线程池控制的方法&＃xff0c;会中断正在运行的线程。

所以&＃xff0c;Worker继承自AQS&＃xff0c;用于判断线程是否空闲以及是否可以被中断。

此外&＃xff0c;在构造方法中执行了 setState(-1) &＃xff0c;把state变量设置为-1&＃xff0c;为什么这么做呢&＃xff1f;是因为AQS中默认的state是0&＃xff0c;如果刚创建了一个Worker对象&＃xff0c;还没有执行任务时&＃xff0c;这时就不应该被中断&＃xff0c;看一下tryAquire方法&＃xff1a;

protected boolean tryAcquire(intunused) {

if (compareAndSetState(0, 1)) {

setExclusiveOwnerThread(Thread.currentThread());

return true;

}

return false;

}

tryAcquire方法是根据state是否是0来判断的&＃xff0c;所以&＃xff0c;setState(-1);将state设置为-1是为了禁止在执行任务前对线程进行中断。

正因为如此&＃xff0c;在runWorker方法中会先调用Worker对象的unlock方法将state设置为0。

五、runWorker方法

在Worker类中的run方法调用了runWorker方法来执行任务&＃xff0c;runWorker方法的代码如下&＃xff1a;

/*** 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&＃39;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&＃39;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&＃39;s UncaughtExceptionHandler have as accurate

* information as we can provide about any problems encountered by

* user code.

*

*&＃64;paramw the worker*/

final voidrunWorker(Worker w) {

Thread wt&＃61;Thread.currentThread();//获取第一个任务

Runnable task &＃61;w.firstTask;

w.firstTask&＃61; null;//允许中断

w.unlock(); //allow interrupts//是否因为异常退出循环

boolean completedAbruptly &＃61; true;try{//如果task为空&＃xff0c;则通过getTask来获取任务//getTask()方法循环获取工作队列的任务

while (task !&＃61; null || (task &＃61; getTask()) !&＃61; 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&＃61; null;try{

task.run();

}catch(RuntimeException x) {

thrown&＃61; x; throwx;

}catch(Error x) {

thrown&＃61; x; throwx;

}catch(Throwable x) {

thrown&＃61; x; throw newError(x);

}finally{

afterExecute(task, thrown);

}

}finally{

task&＃61; null;

w.completedTasks&＃43;&＃43;;

w.unlock();

}

}

completedAbruptly&＃61; false;

}finally{

processWorkerExit(w, completedAbruptly);

}

}

这里说明一下第一个if判断&＃xff0c;目的是&＃xff1a;

如果线程池正在停止&＃xff0c;那么要保证当前线程是中断状态&＃xff1b;

如果不是的话&＃xff0c;则要保证当前线程不是中断状态&＃xff1b;

这里要考虑在执行该if语句期间可能也执行了shutdownNow方法&＃xff0c;shutdownNow方法会把状态设置为STOP&＃xff0c;回顾一下STOP状态&＃xff1a;

不能接受新任务&＃xff0c;也不处理队列中的任务&＃xff0c;会中断正在处理任务的线程。在线程池处于 RUNNING 或 SHUTDOWN 状态时&＃xff0c;调用 shutdownNow() 方法会使线程池进入到该状态。

STOP状态要中断线程池中的所有线程&＃xff0c;而这里使用 Thread.interrupted() 来判断是否中断是为了确保在RUNNING或者SHUTDOWN状态时线程是非中断状态的&＃xff0c;因为 Thread.interrupted() 方法会复位中断的状态。

总结一下runWorker方法的执行过程&＃xff1a;

while循环不断地通过getTask()方法获取任务&＃xff1b;

getTask()方法从阻塞队列中取任务&＃xff1b;

如果线程池正在停止&＃xff0c;那么要保证当前线程是中断状态&＃xff0c;否则要保证当前线程不是中断状态&＃xff1b;

调用task.run()执行任务&＃xff1b;

如果task为null则跳出循环&＃xff0c;执行processWorkerExit()方法&＃xff1b;

runWorker方法执行完毕&＃xff0c;也代表着Worker中的run方法执行完毕&＃xff0c;销毁线程。

这里的beforeExecute方法和afterExecute方法在ThreadPoolExecutor类中是空的&＃xff0c;留给子类来实现。

completedAbruptly变量来表示在执行任务过程中是否出现了异常&＃xff0c;在processWorkerExit方法中会对该变量的值进行判断。

六、getTask方法

getTask方法用来从阻塞队列中取任务&＃xff0c;代码如下&＃xff1a;

/*** 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,

* {&＃64;codeallowCoreThreadTimeOut || 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.

*

*&＃64;returntask, or null if the worker must exit, in which case

* workerCount is decremented*/

privateRunnable getTask() {//timeOut变量的值表示上次从阻塞队列中取任务时是否超时

boolean timedOut &＃61; false; //Did the last poll() time out?

for(;;) {int c &＃61;ctl.get();int rs &＃61;runStateOf(c);//Check if queue empty only if necessary.

/** 如果线程池状态rs >&＃61; SHUTDOWN&＃xff0c;也就是非RUNNING状态&＃xff0c;再进行以下判断&＃xff1a;

* 1. rs >&＃61; STOP&＃xff0c;线程池是否正在stop&＃xff1b;

* 2. 阻塞队列是否为空。

* 如果以上条件满足&＃xff0c;则将workerCount减1并返回null。

* 因为如果当前线程池状态的值是SHUTDOWN或以上时&＃xff0c;不允许再向阻塞队列中添加任务。*/

if (rs >&＃61; SHUTDOWN && (rs >&＃61; STOP ||workQueue.isEmpty())) {

decrementWorkerCount();return null;

}int wc &＃61;workerCountOf(c);//Are workers subject to culling?//允许核心线程超时 或者当前线程数大于核心线程数

/*timed变量用于判断是否需要进行超时控制。

* allowCoreThreadTimeOut默认是false&＃xff0c;也就是核心线程不允许进行超时&＃xff1b;

* wc > corePoolSize&＃xff0c;表示当前线程池中的线程数量大于核心线程数量&＃xff1b;

* 对于超过核心线程数量的这些线程&＃xff0c;需要进行超时控制*/

boolean timed &＃61; allowCoreThreadTimeOut || wc >corePoolSize;/** wc > maximumPoolSize的情况是因为可能在此方法执行阶段同时执行了setMaximumPoolSize方法&＃xff1b;

* timed && timedOut 如果为true&＃xff0c;表示当前操作需要进行超时控制&＃xff0c;并且上次从阻塞队列中获取任务发生了超时

* 接下来判断&＃xff0c;如果有效线程数量大于1&＃xff0c;或者阻塞队列是空的&＃xff0c;那么尝试将workerCount减1&＃xff1b;

* 如果减1失败&＃xff0c;则返回重试。

* 如果wc &＃61;&＃61; 1时&＃xff0c;也就说明当前线程是线程池中唯一的一个线程了。*/

if ((wc > maximumPoolSize || (timed &&timedOut))&& (wc > 1 ||workQueue.isEmpty())) {if(compareAndDecrementWorkerCount(c))return null;continue;

}try{/** 根据timed来判断&＃xff0c;如果为true&＃xff0c;则通过阻塞队列的poll方法进行超时控制&＃xff0c;如果在keepAliveTime时间内没有获取到任务&＃xff0c;则返回null&＃xff1b;

* 否则通过take方法&＃xff0c;如果这时队列为空&＃xff0c;则take方法会阻塞直到队列不为空。

**/Runnable r&＃61; timed ?

//从工作队列poll任务&＃xff0c;不阻塞

workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) ://阻塞等待任务

workQueue.take();if (r !&＃61; null)returnr;//如果 r &＃61;&＃61; null&＃xff0c;说明已经超时&＃xff0c;timedOut设置为true

timedOut &＃61; true;

}catch(InterruptedException retry) {//如果获取任务时当前线程发生了中断&＃xff0c;则设置timedOut为false并返回循环重试

timedOut &＃61; false;

}

}

}

这里重要的地方是第二个if判断&＃xff0c;目的是控制线程池的有效线程数量。由上文中的分析可以知道&＃xff0c;在执行execute方法时&＃xff0c;如果当前线程池的线程数量超过了corePoolSize且小于maximumPoolSize&＃xff0c;并且workQueue已满时&＃xff0c;则可以增加工作线程&＃xff0c;但这时如果超时没有获取到任务&＃xff0c;也就是timedOut为true的情况&＃xff0c;说明workQueue已经为空了&＃xff0c;也就说明了当前线程池中不需要那么多线程来执行任务了&＃xff0c;可以把多于corePoolSize数量的线程销毁掉&＃xff0c;保持线程数量在corePoolSize即可。

什么时候会销毁&＃xff1f;当然是runWorker方法执行完之后&＃xff0c;也就是Worker中的run方法执行完&＃xff0c;由JVM自动回收。

getTask方法返回null时&＃xff0c;在runWorker方法中会跳出while循环&＃xff0c;然后会执行processWorkerExit方法。

ThreadPoolExecutor中线程执行任务的示意图如下图所示。

线程池中的线程执行任务分两种情况&＃xff0c;如下。

1)在execute()方法中创建一个线程时&＃xff0c;会让这个线程执行当前任务。

2)这个线程执行完上图中1的任务后&＃xff0c;会反复从BlockingQueue获取任务来执行。

七、processWorkerExit方法

private void processWorkerExit(Worker w, booleancompletedAbruptly) {//如果completedAbruptly值为true&＃xff0c;则说明线程执行时出现了异常&＃xff0c;需要将workerCount减1&＃xff1b;//如果线程执行时没有出现异常&＃xff0c;说明在getTask()方法中已经已经对workerCount进行了减1操作&＃xff0c;这里就不必再减了。

if (completedAbruptly) //If abrupt, then workerCount wasn&＃39;t adjusted

decrementWorkerCount();final ReentrantLock mainLock &＃61; this.mainLock;

mainLock.lock();try{//统计完成的任务数

completedTaskCount &＃43;&＃61;w.completedTasks;//从workers中移除&＃xff0c;也就表示着从线程池中移除了一个工作线程

workers.remove(w);

}finally{

mainLock.unlock();

}//根据线程池状态进行判断是否结束线程池

tryTerminate();int c &＃61;ctl.get();/** 当线程池是RUNNING或SHUTDOWN状态时&＃xff0c;如果worker是异常结束&＃xff0c;那么会直接addWorker&＃xff1b;

* 如果allowCoreThreadTimeOut&＃61;true&＃xff0c;并且等待队列有任务&＃xff0c;至少保留一个worker&＃xff1b;

* 如果allowCoreThreadTimeOut&＃61;false&＃xff0c;workerCount不少于corePoolSize。*/

if(runStateLessThan(c, STOP)) {if (!completedAbruptly) {int min &＃61; allowCoreThreadTimeOut ? 0: corePoolSize;if (min &＃61;&＃61; 0 && !workQueue.isEmpty())

min&＃61; 1;if (workerCountOf(c) >&＃61;min)return; //replacement not needed

}

addWorker(null, false);

}

}

至此&＃xff0c;processWorkerExit执行完之后&＃xff0c;工作线程被销毁&＃xff0c;以上就是整个工作线程的生命周期&＃xff0c;从execute方法开始&＃xff0c;Worker使用ThreadFactory创建新的工作线程&＃xff0c;runWorker通过getTask获取任务&＃xff0c;然后执行任务&＃xff0c;如果getTask返回null&＃xff0c;进入processWorkerExit方法&＃xff0c;整个线程结束&＃xff0c;如图所示&＃xff1a;

八、tryTerminate方法

tryTerminate方法根据线程池状态进行判断是否结束线程池&＃xff0c;代码如下&＃xff1a;

final voidtryTerminate() {for(;;) {int c &＃61;ctl.get();/** 当前线程池的状态为以下几种情况时&＃xff0c;直接返回&＃xff1a;

* 1. RUNNING&＃xff0c;因为还在运行中&＃xff0c;不能停止&＃xff1b;

* 2. TIDYING或TERMINATED&＃xff0c;因为线程池中已经没有正在运行的线程了&＃xff1b;

* 3. SHUTDOWN并且等待队列非空&＃xff0c;这时要执行完workQueue中的task&＃xff1b;*/

if (isRunning(c) ||runStateAtLeast(c, TIDYING)||(runStateOf(c)&＃61;&＃61; SHUTDOWN && !workQueue.isEmpty()))return;//如果线程数量不为0&＃xff0c;则中断一个空闲的工作线程&＃xff0c;并返回

if (workerCountOf(c) !&＃61; 0) { //Eligible to terminate

interruptIdleWorkers(ONLY_ONE);return;

}final ReentrantLock mainLock &＃61; this.mainLock;

mainLock.lock();try{//这里尝试设置状态为TIDYING&＃xff0c;如果设置成功&＃xff0c;则调用terminated方法

if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {try{//terminated方法默认什么都不做&＃xff0c;留给子类实现

terminated();

}finally{//设置状态为TERMINATED

ctl.set(ctlOf(TERMINATED, 0));

termination.signalAll();

}return;

}

}finally{

mainLock.unlock();

}//else retry on failed CAS

}

}

interruptIdleWorkers(ONLY_ONE); 的作用是因为在getTask方法中执行 workQueue.take() 时&＃xff0c;如果不执行中断会一直阻塞。在下面介绍的shutdown方法中&＃xff0c;会中断所有空闲的工作线程&＃xff0c;如果在执行shutdown时工作线程没有空闲&＃xff0c;然后又去调用了getTask方法&＃xff0c;这时如果workQueue中没有任务了&＃xff0c;调用 workQueue.take() 时就会一直阻塞。所以每次在工作线程结束时调用tryTerminate方法来尝试中断一个空闲工作线程&＃xff0c;避免在队列为空时取任务一直阻塞的情况。

九、shutdown方法

shutdown方法要将线程池切换到SHUTDOWN状态&＃xff0c;并调用interruptIdleWorkers方法请求中断所有空闲的worker&＃xff0c;最后调用tryTerminate尝试结束线程池。

public voidshutdown() {final ReentrantLock mainLock &＃61; this.mainLock;

mainLock.lock();try{//安全策略判断

checkShutdownAccess();//切换状态为SHUTDOWN

advanceRunState(SHUTDOWN);//中断空闲线程

interruptIdleWorkers();

onShutdown();//hook for ScheduledThreadPoolExecutor

} finally{

mainLock.unlock();

}//尝试结束线程池

tryTerminate();

}

这里思考一个问题&＃xff1a;在runWorker方法中&＃xff0c;执行任务时对Worker对象w进行了lock操作&＃xff0c;为什么要在执行任务的时候对每个工作线程都加锁呢&＃xff1f;

下面仔细分析一下&＃xff1a;

在getTask方法中&＃xff0c;如果这时线程池的状态是SHUTDOWN并且workQueue为空&＃xff0c;那么就应该返回null来结束这个工作线程&＃xff0c;而使线程池进入SHUTDOWN状态需要调用shutdown方法&＃xff1b;

shutdown方法会调用interruptIdleWorkers来中断空闲的线程&＃xff0c;interruptIdleWorkers持有mainLock&＃xff0c;会遍历workers来逐个判断工作线程是否空闲。但getTask方法中没有mainLock&＃xff1b;

在getTask中&＃xff0c;如果判断当前线程池状态是RUNNING&＃xff0c;并且阻塞队列为空&＃xff0c;那么会调用 workQueue.take() 进行阻塞&＃xff1b;

如果在判断当前线程池状态是RUNNING后&＃xff0c;这时调用了shutdown方法把状态改为了SHUTDOWN&＃xff0c;这时如果不进行中断&＃xff0c;那么当前的工作线程在调用了 workQueue.take() 后会一直阻塞而不会被销毁&＃xff0c;因为在SHUTDOWN状态下不允许再有新的任务添加到workQueue中&＃xff0c;这样一来线程池永远都关闭不了了&＃xff1b;

由上可知&＃xff0c;shutdown方法与getTask方法(从队列中获取任务时)存在竞态条件&＃xff1b;

解决这一问题就需要用到线程的中断&＃xff0c;也就是为什么要用interruptIdleWorkers方法。在调用 workQueue.take() 时&＃xff0c;如果发现当前线程在执行之前或者执行期间是中断状态&＃xff0c;则会抛出InterruptedException&＃xff0c;解除阻塞的状态&＃xff1b;

但是要中断工作线程&＃xff0c;还要判断工作线程是否是空闲的&＃xff0c;如果工作线程正在处理任务&＃xff0c;就不应该发生中断&＃xff1b;

所以Worker继承自AQS&＃xff0c;在工作线程处理任务时会进行lock&＃xff0c;interruptIdleWorkers在进行中断时会使用tryLock来判断该工作线程是否正在处理任务&＃xff0c;如果tryLock返回true&＃xff0c;说明该工作线程当前未执行任务&＃xff0c;这时才可以被中断。

下面就来分析一下interruptIdleWorkers方法。

十、interruptIdleWorkers方法

private voidinterruptIdleWorkers() {

interruptIdleWorkers(false);

}private void interruptIdleWorkers(booleanonlyOne) {final ReentrantLock mainLock &＃61; this.mainLock;

mainLock.lock();try{for(Worker w : workers) {

Thread t&＃61;w.thread;if (!t.isInterrupted() &&w.tryLock()) {try{

t.interrupt();

}catch(SecurityException ignore) {

}finally{

w.unlock();

}

}if(onlyOne)break;

}

}finally{

mainLock.unlock();

}

}

interruptIdleWorkers遍历workers中所有的工作线程&＃xff0c;若线程没有被中断tryLock成功&＃xff0c;就中断该线程。

为什么需要持有mainLock&＃xff1f;因为workers是HashSet类型的&＃xff0c;不能保证线程安全。

十一、shutdownNow方法

public ListshutdownNow() {

Listtasks;final ReentrantLock mainLock &＃61; this.mainLock;

mainLock.lock();try{

checkShutdownAccess();

advanceRunState(STOP);//中断所有工作线程&＃xff0c;无论是否空闲

interruptWorkers();//取出队列中没有被执行的任务

tasks &＃61;drainQueue();

}finally{

mainLock.unlock();

}

tryTerminate();returntasks;

}

shutdownNow方法与shutdown方法类似&＃xff0c;不同的地方在于&＃xff1a;

设置状态为STOP&＃xff1b;

中断所有工作线程&＃xff0c;无论是否是空闲的&＃xff1b;

取出阻塞队列中没有被执行的任务并返回。

shutdownNow方法执行完之后调用tryTerminate方法&＃xff0c;该方法在上文已经分析过了&＃xff0c;目的就是使线程池的状态设置为TERMINATED。