简介
Executor两级调用模型
Executor的结构和成员
Executor框架的使用
ThreadPoolExecutor
ThreadPoolExecutor执行execute：
- 如果当前运行的线程少于corePoolSize，则创建新线程来执行任务（需要获取全局锁）。
- 如果运行的线程等于或多余corePoolSize，则将任务加入BlockingQueue
- 如果无法将任务加入BlockingQueue（队列已满），则创建新的线程来处理任务（需要获取全局锁）。
- 如果创建新线程将使当前运行的线程超过maximumPoolSize，任务将被拒绝，并调用RejectedExecutionHandler.rejectedExecution()方法。
FixedThreadPool
FixedThreadPool适用于为了满足管理资源的需求，而需要限制当前线程数量的应用场景，它适用于负载比较重的服务器。
1 | public static ExecutorService newFixedThreadPool(int nThreads) { |
其corePoolSize和maximumPoolSize都被设为nThreads的值。当线程池中的线程数大于corePoolSize时，keepAliveTime为多余的空闲线程等待新任务的最长时间，超过这个时间后多余的线程将被终止。具体在FixedThreadPool的执行过程如下：
- 如果当前运行的线程数少于corePoolSize，就创建新的线程执行任务
- 在线程池如果当前运行的线程数等于corePoolSize时，将任务加入到LinkedBlockingQueue等待执行
- 线程执行完1中的任务后，会在循环中反复从LinkedBlockingQueue获取任务来执行
- 由于LinkedBlockingQueue使用的无界队列，所以线程池中线程数不会超过corePoolSize，因此不断加入线程池中的任务将被执行，因为不会马上被执行的任务都加入到LinkedBlockingQueue等待了。
SingleThreadExecutor
SingleThreadExecutor适用于需要保证顺序地执行各个任务，并且在任意时间点不会有多个线程在活动的场景。
1 | public static ExecutorService newSingleThreadExecutor() { |
corePoolSize和maximumPoolSize都为1，且工作队列为无界队列，所以，当启动了一个线程后，以后所有的工作直接加入工作队列中。
CachedThreadPool
CachedThreadPool是大小无界的线程池，适用于执行很多的短期异步任务的小程序，或者负载比较轻的服务器;是一个根据需要创建线程的线程池
1 | public static ExecutorService newCachedThreadPool() { |
CachedThreadPool的corePoolSize为0，maximumPoolSize为Integer.MAX_VALUE，keepAliveTime为60L，意味着多余的空闲线程等待新任务的执行时间为60秒。
CachedThreadPool使用没有容量的SynchronousQueue作为线程池的工作队列（SynchronousQueue是一个没有容量的阻塞队列，
每个插入操作必须等待另一个线程的对应移除操作），但是CachedThreadPool的maximumPool是无界的。
这就意味着如果线程的提交速度高于线程的处理速度，CachedThreadPool会不断创建线程，极端情况是因为创建线程过多耗尽CPU和内存资源。
ScheduledThreadPoolExecutor
使用的DelayedWorkQueue是一个无界队列，所以maximumPoolSize参数无效。
创建一个支持定时及周期性的任务执行的线程池，多数情况下可用来替代Timer类。ScheduledThreadPoolExecutor适用于需要在多个后台线程执行周期任务，同时为了满足资源管理需求需要限制后台线程数量的应用场景。
ScheduledThreadPoolExecutor
1 | public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) { |
SingleThreadScheduledExecutor
1 | public static ScheduledExecutorService newSingleThreadScheduledExecutor() { |
DelayedWorkQueue
DelayedWorkQueue是一个基于堆的数据结构，类似于DelayQueue和PriorityQueue。在执行定时任务的时候，每个任务的执行时间都不同，所以DelayedWorkQueue的工作就是按照执行时间的升序来排列，执行时间距离当前时间越近的任务在队列的前面（注意：这里的顺序并不是绝对的，堆中的排序只保证了子节点的下次执行时间要比父节点的下次执行时间要大，而叶子节点之间并不一定是顺序的，下文中会说明）。
可见，DelayedWorkQueue是一个基于最小堆结构的队列。堆结构可以使用数组表示，可以转换成如下的数组：
在这种结构中，可以发现有如下特性：
假设，索引值从0开始，子节点的索引值为k，父节点的索引值为p，则：
- 一个节点的左子节点的索引为：k = p * 2 + 1；
- 一个节点的右子节点的索引为：k = (p + 1) * 2；
- 一个节点的父节点的索引为：p = (k - 1) / 2。
offer
既然是阻塞队列，入队的操作如add和put方法都调用了offer方法，下面查看一下offer方法：1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31public boolean offer(Runnable x) {
if (x == null)
throw new NullPointerException();
RunnableScheduledFuture<?> e = (RunnableScheduledFuture<?>)x;
final ReentrantLock lock = this.lock;
lock.lock();
try {
int i = size;
// queue是一个RunnableScheduledFuture类型的数组，如果容量不够需要扩容
if (i >= queue.length)
grow();
size = i + 1;
// i == 0 说明堆中还没有数据
if (i == 0) {
queue[0] = e;
setIndex(e, 0);
} else {
// i != 0 时，需要对堆进行重新排序
siftUp(i, e);
}
// 如果传入的任务已经是队列的第一个节点了，这时available需要发出信号
if (queue[0] == e) {
// leader设置为null为了使在take方法中的线程在通过available.signal();后会执行available.awaitNanos(delay);
leader = null;
available.signal();
}
} finally {
lock.unlock();
}
return true;
}
siftUp
1 | // 基于二叉树的实现 |
代码很好理解，就是循环的根据key节点与它的父节点来判断，如果key节点的执行时间小于父节点，则将两个节点交换，使执行时间靠前的节点排列在队列的前面。
假设新入队的节点的延迟时间（调用getDelay()方法获得）是5，执行过程如下：
- 先将新的节点添加到数组的尾部，这时新节点的索引k为7：
- 计算新父节点的索引：parent = (k - 1) >>> 1，parent = 3，那么queue[3]的时间间隔值为8，因为 5 < 8 ，将执行queue[7] = queue[3]：
- 这时将k设置为3，继续循环，再次计算parent为1，queue[1]的时间间隔为3，因为 5 > 3 ，这时退出循环，最终k为3：
可见，每次新增节点时，只是根据父节点来判断，而不会影响兄弟节点。
另外，setIndex方法只是设置了ScheduledFutureTask中的heapIndex属性：1
2
3
4private void setIndex(RunnableScheduledFuture<?> f, int idx) {
if (f instanceof ScheduledFutureTask)
((ScheduledFutureTask)f).heapIndex = idx;
}
take方法
1 | public RunnableScheduledFuture<?> take() throws InterruptedException { |
take方法是什么时候调用的呢？在深入理解Java线程池：ThreadPoolExecutor中，介绍了getTask方法，工作线程会循环地从workQueue中取任务。但定时任务却不同，因为如果一旦getTask方法取出了任务就开始执行了，而这时可能还没有到执行的时间，所以在take方法中，要保证只有在到指定的执行时间的时候任务才可以被取走。
再来说一下leader的作用，这里的leader是为了减少不必要的定时等待，当一个线程成为leader时，它只等待下一个节点的时间间隔，但其它线程无限期等待。 leader线程必须在从take（）或poll（）返回之前signal其它线程，除非其他线程成为了leader。
举例来说，如果没有leader，那么在执行take时，都要执行available.awaitNanos(delay)，假设当前线程执行了该段代码，这时还没有signal，第二个线程也执行了该段代码，则第二个线程也要被阻塞。多个这时执行该段代码是没有作用的，因为只能有一个线程会从take中返回queue[0]（因为有lock），其他线程这时再返回for循环执行时取的queue[0]，已经不是之前的queue[0]了，然后又要继续阻塞。
所以，为了不让多个线程频繁的做无用的定时等待，这里增加了leader，如果leader不为空，则说明队列中第一个节点已经在等待出队，这时其它的线程会一直阻塞，减少了无用的阻塞（注意，在finally中调用了signal()来唤醒一个线程，而不是signalAll()）。
finishPoll
1 | private RunnableScheduledFuture<?> finishPoll(RunnableScheduledFuture<?> f) { |
remove
类似finishPoll：1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26// 替换指定位置的元素，即重新平衡二叉树进而移除元素
public boolean remove(Object x) {
final ReentrantLock lock = this.lock;
lock.lock();
try {
int i = indexOf(x);
if (i < 0)
return false;
setIndex(queue[i], -1);
int s = --size;
RunnableScheduledFuture<?> replacement = queue[s];// 取最后一个元素进行替换
queue[s] = null;
if (s != i) {
siftDown(i, replacement);// 对i位置进行替换
if (queue[i] == replacement)
// 如果queue[i] == replacement，说明i是叶子节点
// 如果是这种情况，不能保证子节点的下次执行时间比父节点的大
// 这时需要进行一次向上调整
siftUp(i, replacement);
}
return true;
} finally {
lock.unlock();
}
}
siftDown
出队列时，调用siftDown：1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27// 指定位置元素的值设为k，并进行比较、下移；可用于替换指定位置的元素
private void siftDown(int k, RunnableScheduledFuture<?> key) {
// 根据二叉树的特性，数组长度除以2，表示取有子节点的索引
int half = size >>> 1;
// 判断索引为k的节点是否有子节点
while (k < half) {
// 左子节点的索引
int child = (k << 1) + 1;
RunnableScheduledFuture<?> c = queue[child];
// 右子节点的索引
int right = child + 1;
// 如果有右子节点并且左子节点的时间间隔大于右子节点，取时间间隔最小的节点
if (right < size && c.compareTo(queue[right]) > 0)
c = queue[child = right];
// 如果key的时间间隔小于等于c的时间间隔，跳出循环
if (key.compareTo(c) <= 0)
break;
// 将字节点上移
queue[k] = c;
setIndex(c, k);
// 设置索引
k = child;
}
// 将key放入索引为k的位置
queue[k] = key;
setIndex(key, k);
}
FutureTask
jdk1.6基于AQS来实现的，从jdk1.7开始是基于volatile来实现的。
1 | public interface RunnableFuture<V> extends Runnable, Future<V> |
另外，使用线程池submit的task，实际上被包裹成FutureTask：1
2
3
4
5
6
7
8
9
10public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
RunnableFuture<Void> ftask = newTaskFor(task, null);
execute(ftask);
return ftask;
}
protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
return new FutureTask<T>(runnable, value);
}
Constructor
1 | /** |
run
1 | public void run() { |
run()方法首先会
- 判断当前任务的state是否等于NEW,如果不为NEW则说明任务或者已经执行过，或者已经被取消，直接返回。
- 如果状态为NEW则接着会通过unsafe类把任务执行线程引用CAS的保存在runner字段中，如果保存失败，则直接返回。
- 执行任务。
- 如果任务执行发生异常，则调用setException()方法保存异常信息。setException()方法如下：
get
1 | public V get() throws InterruptedException, ExecutionException { |
awaitDone()中有个死循环，每一次循环都会
- 判断调用get()的线程是否被其他线程中断，如果是的话则在等待队列中删除对应节点然后抛出InterruptedException异常。
- 获取任务当前状态，如果当前任务状态大于COMPLETING则表示任务执行完成，则把thread字段置null并返回结果。
- 如果任务处于COMPLETING状态，则表示任务已经处理完成(正常执行完成或者执行出现异常)，但是执行结果或者异常原因还没有保存到outcome字段中。这个时候调用线程让出执行权让其他线程优先执行。
- 如果等待节点为空，则构造一个等待节点WaitNode。
- 如果第四步中新建的节点还没如队列，则CAS的把该节点加入waiters队列的首节点。
- 阻塞等待。
假设当前state=NEW且waiters为NULL,也就是说还没有任何一个线程调用get()获取执行结果，这个时候有两个线程threadA和threadB先后调用get()来获取执行结果。再假设这两个线程在加入阻塞队列进行阻塞等待之前任务都没有执行完成且threadA和threadB都没有被中断的情况下(因为如果threadA和threadB在进行阻塞等待结果之前任务就执行完成或线程本身被中断的话，awaitDone()就执行结束返回了)，执行过程是这样的，以threadA为例:
- 第一轮for循环，执行的逻辑是q == null,所以这时候会新建一个节点q。第一轮循环结束。
- 第二轮for循环，执行的逻辑是!queue，这个时候会把第一轮循环中生成的节点的netx指针指向waiters，然后CAS的把节点q替换waiters。也就是把新生成的节点添加到waiters链表的首节点。如果替换成功，queued=true。第二轮循环结束。
- 第三轮for循环，进行阻塞等待。要么阻塞特定时间，要么一直阻塞知道被其他线程唤醒。
cancel(boolean)
1 | public boolean cancel(boolean mayInterruptIfRunning) { |
cancel()方法会做下面几件事:
- 判断任务当前执行状态，如果任务状态不为NEW，则说明任务或者已经执行完成，或者执行异常，不能被取消，直接返回false表示执行失败。
- 判断需要中断任务执行线程，则把任务状态从NEW转化到INTERRUPTING。这是个中间状态。中断任务执行线程。修改任务状态为INTERRUPTED。这个转换过程对应上图中的四。
- 如果不需要中断任务执行线程，直接把任务状态从NEW转化为CANCELLED。如果转化失败则返回false表示取消失败。这个转换过程对应上图中的四。
- 调用finishCompletion()。
当调用cancel(true)方法的时候，实际执行还是Thread.interrupt()方法，而interrupt()方法只是设置中断标志位，如果被中断的线程处于sleep()、wait()或者join()逻辑中则会抛出InterruptedException异常。
finishCompletion()
根据前面的分析，不管是任务执行异常还是任务正常执行完毕，或者取消任务，最后都会调用finishCompletion()方法，该方法实现如下:1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24private void finishCompletion() {
// assert state > COMPLETING;
for (WaitNode q; (q = waiters) != null;) {
if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
for (;;) {
Thread t = q.thread;
if (t != null) {
q.thread = null;
LockSupport.unpark(t);
}
WaitNode next = q.next;
if (next == null)
break;
q.next = null; // unlink to help gc
q = next;
}
break;
}
}
done();
callable = null; // to reduce footprint
}
这个方法的实现比较简单，依次遍历waiters链表，唤醒节点中的线程，然后把callable置空。
被唤醒的线程会各自从awaitDone()方法中的LockSupport.park*()阻塞中返回，然后会进行新一轮的循环。在新一轮的循环中会返回执行结果(或者更确切的说是返回任务的状态)。