为什么要使用线程池?
合理利用线程池能够带来三个好处。第一:降低资源消耗。通过重复利用已创建的线程降低线程创建和销毁造成的消耗。第二:提高响应速度。当任务到达时,任务可以不需要等到线程创建就能立即执行。第三:提高线程的可管理性。线程是稀缺资源,如果无限制的创建,不仅会消耗系统资源,还会降低系统的稳定性,使用线程池可以进行统一的分配,调优和监控。但是要做到合理的利用线程池,必须对其原理了如指掌。——摘自http://www.infoq.com/cn/articles/java-threadPool。
可以通过线程池的以下属性监控线程池的当前状态:
getTaskCount():线程池已经执行的和未执行的任务总数,因为统计的过程中可能会发生变化,该值是个近似值;
getCompletedTaskCount():已完成的任务数量,是个近似值,该值小于等于TaskCount;
getLargestPoolSize():线程池曾经的最大线程数量,可以通过该值判断线程池是否满过。如该数值等于线程池的最大大小,则表示线程池曾经满过;
getPoolSize():线程池当前的线程数量;
getActiveCount():线程池中活动的线程数(正在执行任务),是个近似值。
还可以通过重写线程池提供的hook方法(beforeExecute、afterExecute和terminated)进行监控,例如监控任务的平均执行时间、最大执行时间和最小执行时间等。
程序员可以通过重写钩子 hook 方法(如beforeExecute)实现ThreadPoolExecutor的扩展。
扩展示例:添加了简单的暂停/恢复功能的子类
1 class PausableThreadPoolExecutor extends ThreadPoolExecutor {
2 private boolean isPaused; //标志是否被暂停
3 private ReentrantLock pauseLock = new ReentrantLock(); //访问isPaused时需要加锁,保证线程安全
4 private Condition unpaused = pauseLock.newCondition();
5
6 public PausableThreadPoolExecutor(...) { super(...); }
7
8 //beforeExecute为ThreadPoolExecutor提供的hood方法
9 protected void beforeExecute(Thread t, Runnable r) {
10 super.beforeExecute(t, r);
11 pauseLock.lock();
12 try {
13 while (isPaused)
14 unpaused.await();
15 } catch(InterruptedException ie) {
16 t.interrupt();
17 } finally {
18 pauseLock.unlock();
19 }
20 }
21 //暂停
22 public void pause() {
23 pauseLock.lock();
24 try {
25 isPaused = true;
26 } finally {
27 pauseLock.unlock();
28 }
29 }
30 //取消暂停
31 public void resume() {
32 pauseLock.lock();
33 try {
34 isPaused = false;
35 unpaused.signalAll();
36 } finally {
37 pauseLock.unlock();
38 }
39 }
40 }
1 //ctl是控制线程池状态的一个变量,包含有效的线程数(workerCount)和线程池的运行状态(runState)两部分信息。高3位表示runState,低29位表示workerCount。
2 private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
3 private static final int COUNT_BITS = Integer.SIZE - 3; //表示workerCount的位数,29位。
4 private static final int CAPACITY = (1 << COUNT_BITS) - 1; //线程数的上限,(2^29)-1,大约5亿
5
6 // runState is stored in the high-order bits
7 private static final int RUNNING = -1 << COUNT_BITS; //能接收新任务和处理队列中的任务
8 private static final int SHUTDOWN = 0 << COUNT_BITS; //不能接收新任务,但可以处理队列中的任务
9 private static final int STOP = 1 << COUNT_BITS; //不能接收新任务,不能处理队列中的任务,中断正在执行的任务
10 private static final int TIDYING = 2 << COUNT_BITS; //所有的线程都被终止,workerCount为0时会进入该状态.
11 private static final int TERMINATED = 3 << COUNT_BITS; //terminated()方法完成后将进入该状态。
以上ThreadPoolExecutor的成员变量表示线程池的状态,状态信息存储在ctl变量中,ctl包含有效线程数(workerCount)和线程池运行状态(runState)两部分信息,ctl的高3位表示runState,低29位表示workerCount。ctl初始值为RUNNING状态且线程数为0。
线程池运行状态的转换如下:
1)线程池在RUNNING状态下调用shutdown()方法会进入到SHUTDOWN状态,(finalize()方法也会调用shutdownNow())。
2)在RUNNING和SHUTDOWN状态下调用 shutdownNow() 方法会进入到STOP状态。
3)在SHUTDOWN状态下,当阻塞队列为空且线程数为0时进入TIDYING状态;在STOP状态下,当线程数为0时进入TIDYING状态。
4)在TIDYING状态,调用terminated()方法完成后进入TERMINATED状态。
1 //阻塞队列
2 private final BlockingQueue<Runnable> workQueue;
3 //可重入锁。访问woker线程和相关记录信息时需要获取该锁
4 private final ReentrantLock mainLock = new ReentrantLock();
5 //包含全部worker线程集合,Accessed only under mainLock,HashSet是非线程安全的.
6 private final HashSet<Worker> workers = new HashSet<Worker>();
7 private final Condition termination = mainLock.newCondition();
8 //记录最大的线程数量,Accessed only under mainLock.
9 private int largestPoolSize;
10 //完成任务的数量,Accessed only under mainLock.
11 private long completedTaskCount;
12
13
14 //以下所有程序员可以控制的参数都被声明为volatile变量,保证可见性。
15
16 //创建线程的工厂
17 private volatile ThreadFactory threadFactory;
18 //线程池饱和或关闭时的处理策略(提供了四种饱和策略)
19 private volatile RejectedExecutionHandler handler;
20 //超出corePoolSize数量的空闲线程存活时间(allowCoreThreadTimeOut=true时有效)
21 private volatile long keepAliveTime;
22 //allowCoreThreadTimeOut=false,线程不会因为空闲时间超过keepAliveTime而被停止
23 private volatile boolean allowCoreThreadTimeOut;
24 //核心线程数
25 private volatile int corePoolSize;
26 //最大线程数,此变量的最大上限为CAPACITY
27 private volatile int maximumPoolSize;
一、线程池核心线程数和最大线程数
ThreadPoolExecutor 将根据 corePoolSize (核心线程数)和 maximumPoolSize(最大线程数)设置的边界自动调整线程池大小。当新任务在方法 execute(java.lang.Runnable) 中提交时,如果运行的线程少于 corePoolSize,则创建新线程来处理请求,即使其他辅助线程是空闲的。如果运行的线程多于 corePoolSize 而少于 maximumPoolSize,则仅当队列满时才创建新线程。如果设置的 corePoolSize 和 maximumPoolSize 相同,则创建了固定大小的线程池。如果将 maximumPoolSize 设置为基本的无界值(如 Integer.MAX_VALUE),则允许池适应任意数量的并发任务。在大多数情况下,核心和最大池大小仅基于构造函数来设置,不过也可以使用 setCorePoolSize(int) 和 setMaximumPoolSize(int) 进行动态更改。
二、任务队列
workQueue是一个阻塞队列,用来存储执行的任务。所有的BlockingQueue都可用于workQueue。
如果有效的线程数小于 corePoolSize,则线程池首选添加新线程,而不进行排队。
如果有效的线程数大于等于 corePoolSize,则线程池首选将任务加入队列,而不添加新的线程。
如果队列已满,则创建新的线程,当线程数超出 maximumPoolSize 时,任务将被拒绝。
常用的三种阻塞队列的实现:
1)直接提交。SynchronousQueue是一个不存储元素的阻塞队列。每个插入操作必须等到另一个线程调用移除操作,否则插入操作一直处于阻塞状态。它将任务直接提交给线程而不存储任务。直接提交通常要求不限制 maximumPoolSizes 以避免拒绝新提交的任务。Executors.newCachedThreadPool使用了这个队列。
2)无界队列。LinkedBlockingQueue是一个基于链表结构的阻塞队列,默认的大小是Integer.MAX_VALUE。创建的线程就不会超过 corePoolSize,会使maximumPoolSize 的值无效。
3)有界队列。ArrayBlockingQueue是一个基于数组结构的有界阻塞队列。有助于防止资源耗尽,但是可能较难调整和控制。
三、饱和策略
当 Executor 已经关闭,或者 Executor 将有限边界用于最大线程和工作队列容量且已经饱和时,在方法 execute(Runnable) 中提交的新任务将被拒绝。线程池提供了4种饱和策略:
1)AbortPolicy。默认的饱和策略,直接抛出RejectedExecutionException异常。
2)CallerRunsPolicy。用调用者所在的线程来执行任务,此策略提供简单的反馈控制机制,能够减缓新任务的提交速度。
3)DiscardPolicy。直接丢弃任务。
4)DiscardOldestPolicy。如果执行程序尚未关闭,则丢弃阻塞队列中最靠前的任务,然后重试执行新任务(如果再次失败,则重复此过程)。
也可以使用自定义的 RejectedExecutionHandler 类,但需要非常小心,尤其是当策略仅用于特定容量或排队策略时。
四、threadFactory
使用 ThreadFactory 创建新线程,默认情况下在同一个 ThreadGroup 中一律使用 Executors.defaultThreadFactory() 创建线程,这些线程具有相同的 NORM_PRIORITY 优先级和非守护进程状态。通过自定义的 ThreadFactory创建新线程,可以改变线程的名称、线程组、优先级、守护进程状态等。
五、workers用来存储工作线程,注意HashSet<Worker>是非线程安全的,访问时需要获取mainLock;
六、mainLock是一个独占式可重入锁,用来保证访问workers和其他监控变量(如largestPoolSize、completedTaskCount等)的线程安全。
七、keepAliveTime为线程池的工作线程空闲后,保持存活的时间。所以如果任务很多,并且每个任务执行的时间比较短,可以调大这个时间,提高线程的利用率。allowCoreThreadTimeout变量表示是否允许核心线程超时,如果allowCoreThreadTimeOut=false,那么当线程空闲时间达到keepAliveTime时,线程会退出,直到线程数量=corePoolSize;如果allowCoreThreadTimeOut=true,那么当线程空闲时间达到keepAliveTime时,线程会退出,直到线程数量=0。
1 public void execute(Runnable command) {
2 if (command == null)
3 throw new NullPointerException();
4 /*
5 * Proceed in 3 steps:
6 *
7 * 1. If fewer than corePoolSize threads are running, try to
8 * start a new thread with the given command as its first
9 * task. The call to addWorker atomically checks runState and
10 * workerCount, and so prevents false alarms that would add
11 * threads when it shouldn't, by returning false.
12 *
13 * 2. If a task can be successfully queued, then we still need
14 * to double-check whether we should have added a thread
15 * (because existing ones died since last checking) or that
16 * the pool shut down since entry into this method. So we
17 * recheck state and if necessary roll back the enqueuing if
18 * stopped, or start a new thread if there are none.
19 *
20 * 3. If we cannot queue task, then we try to add a new
21 * thread. If it fails, we know we are shut down or saturated
22 * and so reject the task.
23 */
24 int c = ctl.get(); //获取线程池的状态(runState和workerCount)
25 //如果线程数小于corePoolSize,新建一个线程执行该任务。
26 if (workerCountOf(c) < corePoolSize) {
27 if (addWorker(command, true))
28 return;
29 c = ctl.get();
30 }
31 //如果线程池是运行状态,并且添加任务到队列成功(队列未满)
32 if (isRunning(c) && workQueue.offer(command)) {
33 int recheck = ctl.get();
34 //再次判断线程池的运行状态,如果不是运行状态,需要从队列删除该任务。使用拒绝策略处理该任务。
35 if (! isRunning(recheck) && remove(command))
36 reject(command);
37 //如果线程数为0,执行addWorker方法。参数为null的原因是任务已经加入到队列,新建的线程从队列取任务执行即可。
38 else if (workerCountOf(recheck) == 0)
39 addWorker(null, false);
40 }
41 //线程池不是RUNNING状态或队列已满,尝试新建一个线程执行该任务。如果失败则拒绝该任务。
42 else if (!addWorker(command, false))
43 reject(command);
44 }
线程被封装在Worker类中。
1 //参数firstTask表示新建线程执行的第一个任务。如果firstTask为null,表示
2 //如果参数core=true,把corePoolSize作为线程数上限的判断条件;如果为false,把maximumPoolSize作为线程数上限的判断条件
3 private boolean addWorker(Runnable firstTask, boolean core) {
4 retry:
5 for (;;) {
6 int c = ctl.get();
7 int rs = runStateOf(c);
8 /*
9 * rs >= SHUTDOWN表示不再接受新任务。
10 * 1)线程池的运行状态为SHUTDOWN;2)firstTask == null;3)阻塞队列不为空,只有这三个条件同时满足才不返回false
11 */
12 // Check if queue empty only if necessary.
13 if (rs >= SHUTDOWN &&
14 ! (rs == SHUTDOWN &&
15 firstTask == null &&
16 ! workQueue.isEmpty()))
17 return false;
18
19 //自旋CAS递增workerCount
20 for (;;) {
21 int wc = workerCountOf(c);
22 //如果线程数超过上限,返回false。如果参数core=true,把corePoolSize作为线程数上限的判断条件;如果为false,把maximumPoolSize作为线程数上限的判断条件
23 if (wc >= CAPACITY ||
24 wc >= (core ? corePoolSize : maximumPoolSize))
25 return false;
26 //CAS递增线程数。如果成功,跳出最外层循环;如果失败,且运行状态没有改变,继续内层循环直到成功。
27 if (compareAndIncrementWorkerCount(c))
28 break retry;
29 //判断runState是否改变,如果改变则继续外层循环
30 c = ctl.get(); // Re-read ctl
31 if (runStateOf(c) != rs)
32 continue retry;
33 // else CAS failed due to workerCount change; retry inner loop
34 }
35 }
36
37 //走到这说明需要新建线程,且workerCount更新成功
38 //下面是新建Worker的过程。
39 boolean workerStarted = false; //新建的Worker是否启动标识
40 boolean workerAdded = false; //新建的Worker是否被添加到workers标识
41 Worker w = null;
42 try {
43 final ReentrantLock mainLock = this.mainLock;
44 w = new Worker(firstTask); //新建Worker
45 final Thread t = w.thread;
46 //什么情况下线程会为null呢?在ThreadFactory创建线程失败时可能会出现。
47 if (t != null) {
48 mainLock.lock(); //获取mainLock锁。对workers(HashSet非线程安全)和largestPoolSize更新必须加锁
49 try {
50 // Recheck while holding lock.
51 // Back out on ThreadFactory failure or if
52 // shut down before lock acquired.
53 int c = ctl.get();
54 int rs = runStateOf(c);
55 /*
56 * 如果运行状态是RUNNING,或者运行状态是SHUTDOWN且firstTask为null,才将新建的Worker添加到workers
57 */
58 if (rs < SHUTDOWN ||
59 (rs == SHUTDOWN && firstTask == null)) {
60 if (t.isAlive()) // precheck that t is startable
61 throw new IllegalThreadStateException();
62 workers.add(w);
63 //更新largestPoolSize,标识线程池曾经出现过的最大线程数
64 int s = workers.size();
65 if (s > largestPoolSize)
66 largestPoolSize = s;
67 workerAdded = true;
68 }
69 } finally {
70 mainLock.unlock(); //释放mainLock锁
71 }
72 if (workerAdded) {
73 //启动线程
74 t.start();
75 workerStarted = true;
76 }
77 }
78 } finally {
79 //新建的Worker未启动,进行失败处理
80 if (! workerStarted)
81 addWorkerFailed(w);
82 }
83 return workerStarted;
84 }
每个线程被封装为一个Worker类实例。Worker类继承了AbstractQueuedSynchronizer,并实现了一个互斥非重入锁。Worker类同时继承了Runnable,Worker类的实例也是一个线程。
1 private final class Worker
2 extends AbstractQueuedSynchronizer
3 implements Runnable
4 {
5 /**
6 * This class will never be serialized, but we provide a
7 * serialVersionUID to suppress a javac warning.
8 */
9 private static final long serialVersionUID = 6138294804551838833L;
10
11 /** Thread this worker is running in. Null if factory fails. */
12 final Thread thread; //处理任务的线程
13 /** Initial task to run. Possibly null. */
14 Runnable firstTask; //传入的任务
15 /** Per-thread task counter */
16 volatile long completedTasks; //完成的任务数
17
18 /**
19 * Creates with given first task and thread from ThreadFactory.
20 * @param firstTask the first task (null if none)
21 */
22 Worker(Runnable firstTask) {
23 //同步状态初始化为-1,在执行runWorker方法前禁止中断当前线程
24 setState(-1); // inhibit interrupts until runWorker
25 this.firstTask = firstTask;
26 this.thread = getThreadFactory().newThread(this); //通过ThreadFactory创建线程
27 }
28
29 /** Delegates main run loop to outer runWorker */
30 public void run() {
31 runWorker(this);
32 }
33
34 // Lock methods
35 //
36 // The value 0 represents the unlocked state.
37 // The value 1 represents the locked state.
38 //实现了一个非重入互斥锁,state=0表示解锁状态,state=1表示加锁状态
39 protected boolean isHeldExclusively() {
40 return getState() != 0;
41 }
42
43 protected boolean tryAcquire(int unused) {
44 if (compareAndSetState(0, 1)) {
45 setExclusiveOwnerThread(Thread.currentThread());
46 return true;
47 }
48 return false;
49 }
50
51 protected boolean tryRelease(int unused) {
52 setExclusiveOwnerThread(null);
53 setState(0);
54 return true;
55 }
56
57 public void lock() { acquire(1); }
58 public boolean tryLock() { return tryAcquire(1); }
59 public void unlock() { release(1); }
60 public boolean isLocked() { return isHeldExclusively(); }
61
62 void interruptIfStarted() {
63 Thread t;
64 if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
65 try {
66 t.interrupt();
67 } catch (SecurityException ignore) {
68 }
69 }
70 }
71 }
1 final void runWorker(Worker w) {
2 Thread wt = Thread.currentThread();
3 Runnable task = w.firstTask;
4 w.firstTask = null;
5 //Worker初始化时同步状态置为-1,此处进行解锁操作目的是将同步状态置为0,允许中断。
6 w.unlock(); // allow interrupts
7 boolean completedAbruptly = true; //是否因为异常跳出循环
8 try {
9 //如果firstTask为null则通过getTask()方法从队列中获取。
10 //正常情况下,会一直执行While循环,如果队列为空,getTask()方法中会阻塞当前线程,getTask()返回null时会跳出循环
11 while (task != null || (task = getTask()) != null) {
12 w.lock(); //加Worker锁
13 // If pool is stopping, ensure thread is interrupted;
14 // if not, ensure thread is not interrupted. This
15 // requires a recheck in second case to deal with
16 // shutdownNow race while clearing interrupt
17 /*
18 * 如果线程池正在停止,要保证当前线程是中断状态
19 * 如果不是,则要保证当前线程不是中断状态
20 * STOP状态要中断线程池中的所有线程,而这里使用Thread.interrupted()来判断是否中断是为了确保在RUNNING或者SHUTDOWN状态时线程是非中断状态的,因为Thread.interrupted()方法会复位中断的状态。
21 */
22 if ((runStateAtLeast(ctl.get(), STOP) ||
23 (Thread.interrupted() &&
24 runStateAtLeast(ctl.get(), STOP))) &&
25 !wt.isInterrupted())
26 wt.interrupt();
27 try {
28 beforeExecute(wt, task); //钩子方法
29 Throwable thrown = null;
30 try {
31 task.run(); //调用任务的run方法,而不是start()方法,因为Worker本身就是一个线程类
32 } catch (RuntimeException x) {
33 thrown = x; throw x;
34 } catch (Error x) {
35 thrown = x; throw x;
36 } catch (Throwable x) {
37 thrown = x; throw new Error(x);
38 } finally {
39 afterExecute(task, thrown); //钩子方法
40 }
41 } finally {
42 task = null;
43 w.completedTasks++;
44 w.unlock(); //释放Worker锁
45 }
46 }
47 completedAbruptly = false;
48 } finally {
49 //跳出循环,执行processWorkerExit()方法
50 processWorkerExit(w, completedAbruptly);
51 }
52 }
1 //如果返回null,在runWorker方法中会执行processWorkerExit,即关闭该线程。
2 private Runnable getTask() {
3 //表示上次从队列获取任务是否超时
4 boolean timedOut = false; // Did the last poll() time out?
5
6 retry:
7 for (;;) {
8 int c = ctl.get();
9 int rs = runStateOf(c);
10
11 // Check if queue empty only if necessary.
12 // 如果rs >= STOP,或者 rs=SHUTDOWN且队列为空,此时不再接收新任务,将WorkerCount递减并返回null。
13 if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
14 decrementWorkerCount(); //自旋CAS递减workerCount直到成功
15 return null;
16 }
17
18 //timed用于判断是否需要重试控制
19 boolean timed; // Are workers subject to culling?
20
21 for (;;) {
22 //allowCoreThreadTimeOut默认是false,核心线程不进行超时控制,当线程数量大于corePoolSize时需要进行超时控制
23 int wc = workerCountOf(c);
24 timed = allowCoreThreadTimeOut || wc > corePoolSize;
25
26 //如果wc <= maximumPoolSize ,且上次从队列获取任务超时或本次需要进行超时控制,则跳出内层循环。
27 //timedOut=true表示上次从队列获取元素超时,说明队列在上次获取的keepAliveTime时间内是空的。
28 //timed=true说明线程数量大于corePoolSize。
29 //所以timedOut=true和timed=true同时满足则说明当前线程已经空闲了keepAliveTime时间,并且线程池的数量大于corePoolSize。这时就需要关闭多余的空闲线程(即compareAndDecrementWorkerCount并返回null)。
30 if (wc <= maximumPoolSize && ! (timedOut && timed))
31 break;
32 //如果线程数量大于maximumPoolSize,或者上次从队列获取任务超时且本次需要进行超时控制。需要递减WorkerCount,如果递减成功则返回null
33 if (compareAndDecrementWorkerCount(c))
34 return null;
35 //检查线程池运行状态是否改变。如果改变,那么继续外层循环,如果未改变,那么继续内层循环。
36 c = ctl.get(); // Re-read ctl
37 if (runStateOf(c) != rs)
38 continue retry;
39 // else CAS failed due to workerCount change; retry inner loop
40 }
41
42 try {
43 Runnable r = timed ?
44 workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
45 //超时方式获取,注意keepAliveTime为超出corePoolSize大小的线程的空闲存活时间
46 workQueue.take(); //阻塞方式获取,如果队列为空阻塞当前线程
47 if (r != null)
48 return r;
49 timedOut = true; //如果超时,继续循环。
50 } catch (InterruptedException retry) {
51 //如果发生中断,则将timedOut置为false,继续循环
52 timedOut = false;
53 }
54 }
55 }
1 private void processWorkerExit(Worker w, boolean completedAbruptly) {
2 //如果completedAbruptly=false,说明是由getTask返回null导致的,WorkerCount递减的操作已经执行。
3 //如果completedAbruptly=true,说明是由执行任务的过程中发生异常导致,需要进行WorkerCount递减的操作。
4 if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
5 decrementWorkerCount();
6
7 final ReentrantLock mainLock = this.mainLock;
8 mainLock.lock();
9 try {
10 completedTaskCount += w.completedTasks;
11 workers.remove(w); //从workers中删除当前worker,对workers更新需要加mainLock锁。
12 } finally {
13 mainLock.unlock();
14 }
15
16 tryTerminate();
17
18 //如果是异常结束(completedAbruptly=true),需要重新调用addWorker()增加一个线程,保持线程数量。
19 //如果是由getTask()返回null导致的线程结束,需要进行以下判断:
20 // 1)如果allowCoreThreadTimeOut=true且队列不为空,那么需要至少保证有一个线程。
21 // 2)如果allowCoreThreadTimeOut=false,那么需要保证线程数大于等于corePoolSize。
22 //
23 int c = ctl.get();
24 if (runStateLessThan(c, STOP)) {
25 if (!completedAbruptly) {
26 int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
27 if (min == 0 && ! workQueue.isEmpty())
28 min = 1;
29 if (workerCountOf(c) >= min)
30 return; // replacement not needed
31 }
32 addWorker(null, false);
33 }
34 }
1 //根据线程池状态判断是否结束线程池
2 final void tryTerminate() {
3 for (;;) {
4 int c = ctl.get();
5 //如果线程池运行状态是RUNNING,或者大于等于TIDYING,或者运行状态为SHUTDOWN且队列为空,则直接return。
6 if (isRunning(c) ||
7 runStateAtLeast(c, TIDYING) ||
8 (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
9 return;
10 //如果线程数不为0,则中断一个空闲线程并return。为什么有这一步操作。
11 if (workerCountOf(c) != 0) { // Eligible to terminate
12 interruptIdleWorkers(ONLY_ONE);
13 return;
14 }
15
16 final ReentrantLock mainLock = this.mainLock;
17 mainLock.lock();
18 try {
19 //尝试将状态设置为TIDYING状态,
20 if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
21 try {
22 //如果CAS成功,执行terminated()方法
23 terminated();
24 } finally {
25 ctl.set(ctlOf(TERMINATED, 0));
26 termination.signalAll();
27 }
28 return;
29 }
30 } finally {
31 mainLock.unlock();
32 }
33 // else retry on failed CAS
34 }
35 }
线程池运行状态由RUNNING到SHUTDOWN的转换。
1 public void shutdown() {
2 final ReentrantLock mainLock = this.mainLock;
3 mainLock.lock();
4 try {
5 //安全管理,检查方法调用者是否有权限中断Worker线程
6 checkShutdownAccess();
7 //运行状态改为SHUTDOWN
8 advanceRunState(SHUTDOWN); //自旋CAS
9 //中断空闲线程
10 interruptIdleWorkers();
11 onShutdown(); // hook for ScheduledThreadPoolExecutor
12 } finally {
13 mainLock.unlock();
14 }
15 //尝试结束线程池
16 tryTerminate();
17 }
18
19 private void interruptIdleWorkers() {
20 interruptIdleWorkers(false);
21 }
22
23 private void interruptIdleWorkers(boolean onlyOne) {
24 final ReentrantLock mainLock = this.mainLock;
25 mainLock.lock(); //对workers的操作需要获取mainLock
26 try {
27 //遍历所有的线程,如果没有被中断且获取锁成功则中断线程。获取锁失败时很可能该线程正在执行任务(woker执行任务时需要对woker加锁)。
28 for (Worker w : workers) {
29 Thread t = w.thread;
30 if (!t.isInterrupted() && w.tryLock()) {
31 try {
32 t.interrupt();
33 } catch (SecurityException ignore) {
34 } finally {
35 w.unlock();
36 }
37 }
38 if (onlyOne)
39 break;
40 }
41 } finally {
42 mainLock.unlock();
43 }
44 }
1 public List<Runnable> shutdownNow() {
2 List<Runnable> tasks;
3 final ReentrantLock mainLock = this.mainLock;
4 mainLock.lock();
5 try {
6 checkShutdownAccess();
7 advanceRunState(STOP);
8 //中断所有线程,即使线程正在执行任务
9 interruptWorkers();
10 //取出队列中的任务
11 tasks = drainQueue();
12 } finally {
13 mainLock.unlock();
14 }
15 //尝试结束线程池
16 tryTerminate();
17 return tasks;
18 }
19
20 private void interruptWorkers() {
21 final ReentrantLock mainLock = this.mainLock;
22 mainLock.lock();
23 try {
24 for (Worker w : workers)
25 w.interruptIfStarted();
26 } finally {
27 mainLock.unlock();
28 }
29 }
30
31 private List<Runnable> drainQueue() {
32 BlockingQueue<Runnable> q = workQueue;
33 List<Runnable> taskList = new ArrayList<Runnable>();
34 q.drainTo(taskList);
35 if (!q.isEmpty()) {
36 for (Runnable r : q.toArray(new Runnable[0])) {
37 if (q.remove(r))
38 taskList.add(r);
39 }
40 }
41 return taskList;
42 }
利用FutureTask可以实现获取异步任务的返回值、取消异步任务等功能。看一下ThreadPoolExecutor的submit方法。submit方法根据任务构造一个FutureTask对象并返回,在主线程中可以根据FutureTask提供的方法进行任务取消和获取异步任务的返回值。
1 public <T> Future<T> submit(Callable<T> task) {
2 if (task == null) throw new NullPointerException();
3 RunnableFuture<T> ftask = newTaskFor(task);
4 execute(ftask); //实际执行的任务是ftask
5 return ftask;
6 }
private volatile int state; //状态,新创建时状态为NEW
private static final int NEW = 0; //新创建
private static final int COMPLETING = 1; //正在执行
private static final int NORMAL = 2; //正常完成
private static final int EXCEPTIONAL = 3; //执行过程中出现异常
private static final int CANCELLED = 4; //被取消
private static final int INTERRUPTING = 5; //
private static final int INTERRUPTED = 6;
/** The underlying callable; nulled out after running */
private Callable<V> callable; //要执行的任务
/** The result to return or exception to throw from get() */
private Object outcome; // non-volatile, protected by state reads/writes
/** The thread running the callable; CASed during run() */
private volatile Thread runner; //执行callable的线程
/** Treiber stack of waiting threads */
private volatile WaitNode waiters; //Treiber算法实现的栈,用于存储等待的线程
static final class WaitNode {
volatile Thread thread;
volatile WaitNode next;
WaitNode() { thread = Thread.currentThread(); }
}
状态的转换有以下几种情况:
1)NEW -> COMPLETING -> NORMAL 正常执行并返回;
2)NEW -> COMPLETING -> EXCEPTIONAL 执行过程中出现异常;
3)NEW -> CANCELLED 执行前被取消
4)NEW -> INTERRUPTING -> INTERRUPTED 取消时被中断。
1 public FutureTask(Callable<V> callable) {
2 if (callable == null)
3 throw new NullPointerException();
4 this.callable = callable;
5 this.state = NEW; // ensure visibility of callable
6 }
7
8 public FutureTask(Runnable runnable, V result) {
9 //由于Runnable没有返回值,通过Executors将Runnable转换为Callable。
10 this.callable = Executors.callable(runnable, result);
11 this.state = NEW; // ensure visibility of callable
12 }
1 public void run() {
2 //只执行state=NEW的任务。如果state!=NEW说明任务已经执行。
3 //如果state=NEW,则通过CAS将runner置为当前线程。如果失败说明其他线程已经执行。
4 if (state != NEW ||
5 !UNSAFE.compareAndSwapObject(this, runnerOffset,
6 null, Thread.currentThread()))
7 return;
8 try {
9 Callable<V> c = callable;
10 if (c != null && state == NEW) {
11 V result; //任务执行结果
12 boolean ran; //任务执行期间是否发生异常
13 try {
14 result = c.call(); //执行任务
15 ran = true;
16 } catch (Throwable ex) {
17 result = null;
18 ran = false;
19 //如果发生异常,执行setException(ex)
20 setException(ex);
21 }
22 //如果正常结束,执行set(result).
23 if (ran)
24 set(result);
25 }
26 } finally {
27 // runner must be non-null until state is settled to
28 // prevent concurrent calls to run()
29 //不管任务执行是否正常,都需要将runner置为null
30 runner = null;
31 // state must be re-read after nulling runner to prevent
32 // leaked interrupts
33 //防止中断泄露,需要结合cancel方法研究
34 //如果s>=INTERRUPTING,说明状态变换为NEW -> INTERRUPTING -> INTERRUPTED,即在取消时被中断。
35 int s = state;
36 if (s >= INTERRUPTING)
37 handlePossibleCancellationInterrupt(s);
38 }
39 }
任务执行正常结束:
1 //任务正常结束,通过CAS更新state为COMPLETING,如果成功,将state更新为NORMAL,唤醒等待线程。
2 protected void set(V v) {
3 if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
4 outcome = v; //将运行结果result赋给outcome
5 UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
6 //删除和唤醒所有的等待线程
7 finishCompletion();
8 }
9 }
任务执行时发生异常:
1 //任务执行时发生异常,通过CAS更新state为COMPLETING,如果成功,将state更新为EXCEPTIONAL,唤醒等待线程
2 protected void setException(Throwable t) {
3 if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
4 outcome = t; //将异常信息赋给outcome
5 UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
6 finishCompletion();
7 }
8 }
唤醒等待获取任务运行结果的线程:
1 private void finishCompletion() {
2 // assert state > COMPLETING;
3 //自旋CAS更新waiters为null直到成功
4 for (WaitNode q; (q = waiters) != null;) {
5 if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
6 for (;;) {
7 Thread t = q.thread;
8 if (t != null) {
9 q.thread = null;
10 LockSupport.unpark(t); //唤醒等待线程,WaitNode是在get方法中添加的
11 }
12 WaitNode next = q.next;
13 if (next == null)
14 break;
15 q.next = null; // unlink to help gc
16 q = next;
17 }
18 break;
19 }
20 }
21
22 done(); //hook方法,默认不执行任何操作,子类可以重写该方法完成指定的功能(例如:回调)
23
24 callable = null; // to reduce footprint
25 }
handlePossibleCancellationInterrupt方法要确保cancel(true)产生的中断发生在run或runAndReset方法执行的过程中。这里会循环的调用Thread.yield()来确保状态在cancel方法中被设置为INTERRUPTED。
1 private void handlePossibleCancellationInterrupt(int s) {
2 // It is possible for our interrupter to stall before getting a
3 // chance to interrupt us. Let's spin-wait patiently.
4 if (s == INTERRUPTING)
5 while (state == INTERRUPTING)
6 Thread.yield(); // wait out pending interrupt
7
8 // assert state == INTERRUPTED;
9
10 // We want to clear any interrupt we may have received from
11 // cancel(true). However, it is permissible to use interrupts
12 // as an independent mechanism for a task to communicate with
13 // its caller, and there is no way to clear only the
14 // cancellation interrupt.
15 //
16 // Thread.interrupted();
17 }
1 public V get() throws InterruptedException, ExecutionException {
2 int s = state;
3 //如果state为NEW或COMPLETING,调用awaitDone方法将当前线程添加到waiters中并阻塞
4 if (s <= COMPLETING)
5 s = awaitDone(false, 0L);
6 //如果已经完成(包括正常结束或异常结束),返回
7 return report(s);
8 }
9
10 //如果超时则抛出TimeoutException异常
11 public V get(long timeout, TimeUnit unit)
12 throws InterruptedException, ExecutionException, TimeoutException {
13 if (unit == null)
14 throw new NullPointerException();
15 int s = state;
16 if (s <= COMPLETING &&
17 (s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING)
18 throw new TimeoutException();
19 return report(s);
20 }
awaitDone方法,阻塞线程。
1 //timed参数表示是否使用超时机制
2 private int awaitDone(boolean timed, long nanos)
3 throws InterruptedException {
4 final long deadline = timed ? System.nanoTime() + nanos : 0L;
5 WaitNode q = null;
6 boolean queued = false; //是否已经入栈
7 for (;;) {
8 //若当前线程被中断,则删除q并抛出InterruptedException()
9 if (Thread.interrupted()) {
10 removeWaiter(q);
11 throw new InterruptedException();
12 }
13
14 int s = state;
15 //如果state大于COMPLETING,表明任务已经完成,则将节点q的线程置为null并返回状态值。
16 if (s > COMPLETING) {
17 if (q != null)
18 q.thread = null;
19 return s;
20 }
21 //s==COMPLETING,说明任务已经执行完成但还没有设置最终状态。
22 //Thread.yield();让当前正在运行的线程回到可运行状态,以允许其他线程(包括当前线程)获得运行的机会。注意目的是尝试让状态改变,继续下个循环。
23 else if (s == COMPLETING) // cannot time out yet
24 Thread.yield();
25 else if (q == null)
26 q = new WaitNode(); //新建WaitNode节点
27 //CAS添加到waiters栈,在阻塞之前先将节点q添加栈,入栈成功后queued更新为true。
28 else if (!queued)
29 queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
30 q.next = waiters, q);
31 else if (timed) {
32 nanos = deadline - System.nanoTime();
33 //如果已经过期,则删除节点q并返回
34 if (nanos <= 0L) {
35 removeWaiter(q);
36 return state;
37 }
38 LockSupport.parkNanos(this, nanos); //超时机制阻塞当前线程
39 }
40 else
41 LockSupport.park(this); //阻塞当前线程
42 }
43 }
44
45 //删除指定节点(Treiber算法实现的栈)
46 private void removeWaiter(WaitNode node) {
47 if (node != null) {
48 node.thread = null; //将线程置为null,因为下面要根据thread是否为null判断是否要把node移出
49 retry:
50 for (;;) { // restart on removeWaiter race
51 for (WaitNode pred = null, q = waiters, s; q != null; q = s) {
52 s = q.next;
53 if (q.thread != null)
54 pred = q;
55 else if (pred != null) {
56 pred.next = s;
57 if (pred.thread == null) // check for race
58 continue retry;
59 }
60 else if (!UNSAFE.compareAndSwapObject(this, waitersOffset, q, s))
61 continue retry;
62 }
63 break;
64 }
65 }
66 }
report方法,返回运行结果或抛出异常。
1 //任务完成返回执行结果或抛出异常
2 private V report(int s) throws ExecutionException {
3 Object x = outcome;
4 //如果任务正常完成,返回执行结果
5 if (s == NORMAL)
6 return (V)x;
7 //如果s >= CANCELLED,说明任务被取消,那么就抛出CancellationException
8 if (s >= CANCELLED)
9 throw new CancellationException();
10 //最后s==EXCEPTIONAL,任务执行时发生异常,抛出该异常
11 throw new ExecutionException((Throwable)x);
12 }
试图取消对此任务的执行。如果任务已完成、或已取消,或者由于某些其他原因而无法取消,则此尝试将失败。当调用 cancel 时,如果调用成功,而此任务尚未启动,则此任务将永不运行。如果任务已经启动,则 mayInterruptIfRunning 参数确定是否应该以试图停止任务的方式来中断执行此任务的线程。
1 public boolean cancel(boolean mayInterruptIfRunning) {
2 //若state != NEW,说明任务已经启动,则直接返回失败。
3 if (state != NEW)
4 return false;
5 //如果mayInterruptIfRunning为true,要中断当前执行任务的线程。
6 if (mayInterruptIfRunning) {
7 //CAS更新state为INTERRUPTING不成功,说明state已被改变(即state != NEW),则直接返回失败。如果成功则中断正在执行任务的线程,并唤醒等待获取结果的线程。
8 if (!UNSAFE.compareAndSwapInt(this, stateOffset, NEW, INTERRUPTING))
9 return false;
10 Thread t = runner;
11 if (t != null)
12 t.interrupt(); //中断当前线程
13 //更新state为INTERRUPTED
14 UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED); // final state
15 }
16 //mayInterruptIfRunning=flase,CAS更新state为CANCELLED,若成功则唤醒等待的线程(不中断正在执行任务的线程),若失败返回false。
17 else if (!UNSAFE.compareAndSwapInt(this, stateOffset, NEW, CANCELLED))
18 return false;
19 finishCompletion();
20 return true;
21 }
Executors是一个工具类,提供了公共的静态方法,例如创建默认线程工厂、创建线程池、把Runnable包装成Callable的方法等。
DefaultThreadFactory类
1 static class DefaultThreadFactory implements ThreadFactory {
2 private static final AtomicInteger poolNumber = new AtomicInteger(1); //线程池序号
3 private final ThreadGroup group; //线程组
4 private final AtomicInteger threadNumber = new AtomicInteger(1); //线程号
5 private final String namePrefix;
6
7 DefaultThreadFactory() {
8 SecurityManager s = System.getSecurityManager();
9 group = (s != null) ? s.getThreadGroup() :
10 Thread.currentThread().getThreadGroup();
11 namePrefix = "pool-" + poolNumber.getAndIncrement() + "-thread-";
12 }
13
14 public Thread newThread(Runnable r) {
15 Thread t = new Thread(group, r,
16 namePrefix + threadNumber.getAndIncrement(), //线程名
17 0);
18 //非守护线程
19 if (t.isDaemon())
20 t.setDaemon(false);
21 //相同的优先级
22 if (t.getPriority() != Thread.NORM_PRIORITY)
23 t.setPriority(Thread.NORM_PRIORITY);
24 return t;
25 }
26 }
创建默认工厂方法:
1 public static ThreadFactory defaultThreadFactory() {
2 return new DefaultThreadFactory();
3 }
1) newFixedThreadPool方法
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
固定线程数的线程池,corePoolSize和 maximumPoolSize 都被设置为nThreads,keepAliveTime=0,由于corePoolSize等于maximumPoolSize,所以keepAliveTime和maximumPoolSize参数是无效的。阻塞队列是LinkedBlockingQueue,是一个无界队列。正常情况下(未执行方法shutdown()或shutdownNow()),不会调用饱和策略。
2)newSingleThreadExecutor方法
1 public static ExecutorService newSingleThreadExecutor() {
2 return new FinalizableDelegatedExecutorService
3 (new ThreadPoolExecutor(1, 1,
4 0L, TimeUnit.MILLISECONDS,
5 new LinkedBlockingQueue<Runnable>()));
6 }
单个线程的线程池,corePoolSize和maximumPoolSize都为1,其他同FixedThreadPool。能保证任务按顺序执行。
3)newCachedThreadPool方法
1 public static ExecutorService newCachedThreadPool() {
2 return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
3 60L, TimeUnit.SECONDS,
4 new SynchronousQueue<Runnable>());
5 }
线程数可改变的线程池,corePoolSize=0,maximumPoolSize=Integer.MAX_VALUE,核心线程数为0,最大线程数为CAPACITY(因为CAPACITY<Integer.MAX_VALUE).keepAliveTime=60L,意味着CachedThreadPool中的空闲线程等待新任务的最长时间为60秒,空闲线程超过60秒后将会被终止。CachedThreadPool使用没有容量的SynchronousQueue作为线程池的工作队列.这意味着,如果主线程提交任务的速度高于maximumPool中线程处理任务的速度时,CachedThreadPool会不断创建新线程。极端情况下,CachedThreadPool会因为创建过多线程而耗尽CPU和内存资源。
1 public static <T> Callable<T> callable(Runnable task, T result) {
2 if (task == null)
3 throw new NullPointerException();
4 return new RunnableAdapter<T>(task, result);
5 }
6
7 public static Callable<Object> callable(Runnable task) {
8 if (task == null)
9 throw new NullPointerException();
10 return new RunnableAdapter<Object>(task, null);
11 }