8.线程池
- 8.1 什么是线程池
- 8.2 线程池作用
- 8.3 线程池体系结构
- Executor框架的两级调度模型
- Executor框架任务执行详细示意图
- 8.4 ThreadPoolExecutor
- 8.4.1 阻塞队列
- 8.4.2 拒绝策略
- 8.4.3 ThreadFactory
- 8.5 线程池原理剖析
- 8.6 Executors线程池工具类
- 8.6.1 newCachedThreadPool
- 介绍
- 构造源码
- 执行示意图
- 示例代码
- 8.6.2 newFixedThreadPool
- 构造源码
- FixedThreadPool 执行示意图
- 示例代码
- 8.6.3 newSingleThreadExecutor
- 介绍
- 构造源码
- SingleThreadExecutor执行示意图
- 示例代码
- 8.6.4 newScheduledThreadPool
- 1.基本使用
- 介绍
- api方法
- 代码实现
- 2.运行机制
- 三种调度执行方式
- schedule方法调度源码分析
- 执行示意图
- 3.DelayedWorkQueue 延迟队列
- 介绍
- 3.6.小结
- 1.基本使用
- 小结
- 8.6.5 向线程池提交任务
- 介绍
- execute方法
- submit方法
- 8.6.1 newCachedThreadPool
- 8.7 监控线程池API
- 8.8 扩展线程池
- 8.9 优化线程池大小
- 8.10 线程池死锁
- 8.11 线程池中的异常处理
- 8.12 ForkJoinPool 线程池
- 4.FutureTask详解
- 4.1.目标
- 4.2.FutureTask基础知识
- FutureTask简介
- api介绍
- 4.FutureTask详解
- 8.13 FutureTask的使用
- FutureTask的应用场景
- 案例—最优的“烧水泡茶”程序
- 目标
- 案例需求
- 解决方案
- FutureTask的实现原理
- 小结
8.1 什么是线程池
Java中的线程池是运用场景最多的并发框架，几乎所有需要异步或并发执行任务的程序都可以使用线程池。在开发过程中，合理地使用线程池能够带来3个好处。
第一：降低资源消耗。通过重复利用已创建的线程降低线程创建和销毁造成的消耗。
第二：提高响应速度。当任务到达时，任务可以不需要等到线程创建就能立即执行。
第三：提高线程的可管理性。线程是稀缺资源，如果无限制地创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一分配、调优和监控。但是，要做到合理利用线程池，必须对其实现原理了如指掌。
8.2 线程池作用
线程池是为突然大量爆发的线程设计的，通过有限的几个固定线程为大量的操作服务，减少了创建和销毁线程所需的时间，从而提高效率。
如果一个线程的时间非常长，就没必要用线程池了(不是不能作长时间操作，而是不宜)，况且我们还不能控制线程池中线程的开始、挂起、和中止。
8.3 线程池体系结构
线程池就是线程的集合，线程池集中管理线程，以实现线程的重用，降低资源消耗，提高响应速度等。线程用于执行异步任务，单个的线程既是工作单元也是执行机制，从JDK1.5开始，为了把工作单元与执行机制分离开，Executor框架诞生了，它是一个用于统一创建任务与运行任务的接口。框架就是异步执行任务的线程池框架。
线程池的体系结构：
java.util.concurrent.Executor 负责线程的使用和调度的 （根接口）
|--ExecutorService 子接口： 线程池的主要接口
|--ThreadPoolExecutor 线程池的实现类（核心）
|--ScheduledExceutorService 子接口： 负责线程的调度
|--ScheduledThreadPoolExecutor : 继承ThreadPoolExecutor，实现了ScheduledExecutorService
| 关键类或接口 | 含义 |
|---|---|
| Executor | 是一个接口，它是Executor框架的基础，它将任务的提交与任务的执行分离开来 |
| ExecutorService | 线程池的主要接口,是Executor的子接口 |
| ThreadPoolExecutor | 是线程池的核心实现类，用来执行被提交的任务 |
| ScheduledThreadPoolExecutor | 另一个关键实现类，可以进行延迟或者定期执行任务。ScheduledThreadPoolExecutor比Timer定时器更灵活，功能更强大 |
| Future接口与FutureTask实现类 | 代表异步计算的结果 |
| Runnable接口和Callable接口的实现类 | 都可以被ThreadPoolExecutor或ScheduledThreadPoolExecutor执行的任务 |
| Executors | 线程池的工具类，可以快捷的创建线程池 |
Executor框架的两级调度模型
Java多线程程序通常把应用分解为若干个任务，任务的执行分为两级调度器
- 用户级的调度器（Executor框架）将这些任务映射为固定数量的线程进行执行；
- 系统内核调度器（系统内核），将这些线程映射到硬件处理器cpu上进行执行。
从图中可以看出
-
应用程序通过Executor框架控制上层的调度；
-
而下层的调度由操作系统内核控制，下层的调度不受应用程序的控制
Executor框架任务执行详细示意图
8.4 ThreadPoolExecutor
Executor框架的最核心实现是ThreadPoolExecutor类，通过传入不同的参数，就可以构造出适用于不同应用场景下的线程池，那么它的底层原理是怎样实现的呢，下面就来介绍下ThreadPoolExecutor线程池的运行过程。
ThreadPoolExecutor 的构造方法：
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
}
核心构造器参数
| 组件 | 含义 |
|---|---|
| corePoolSize | 指定线程池中核心线程的数量 |
| maximumPoolSize | 指定线程池中最大线程数量 |
| keepAliveTime | 当线程池线程的数量超过 corePoolSize 时，多余的空闲线程的存活时长，即空闲线程在多长时长内销毁 |
| unit | 是 keepAliveTime 时长单位 |
| workQueue | 用来暂时保存任务的工作队列（阻塞队列），把任务提交到该任务队列中等待执行 |
| threadFactory | 指定创建线程的线程工厂，用于创建线程 |
| handler | 拒绝策略，当任务太多来不及处理时，如何拒绝。 |
说明：
workQueue 工作队列是指提交未执行的任务队列，它是 BlockingQueue 接口的对象，仅用于存储 Runnable 任务。
8.4.1 阻塞队列
根据队列功能分类，在 ThreadPoolExecutor 构造方法中可以使用以下几种阻塞队列：
1.SynchronousQueue
1) 直接提交队列,由 SynchronousQueue 对象提供,该队列没有容量,提交给线程池的任务不会被真实的保存,总是将新的任务提交给线程执行,如果没有空闲线程,则尝试创建新的线程,如果线程数量已经达到 maxinumPoolSize 规定的最大值,则执行拒绝策略.
SynchronousQueue没有容量，是无缓冲等待队列，是一个不存储元素的阻塞队列，会直接将任务交给消费者，必须等队列中的添加元素被消费后才能继续添加新的元素。
使用SynchronousQueue阻塞队列一般要求maximumPoolSizes为无界，避免线程拒绝执行操作。
2.ArrayBlockingQueue
2) 有界任务队列,由 ArrayBlockingQueue 实现,在创建ArrayBlockingQueue 对象时,可以指定一个容量. 当有任务需要执行时,如果线程池中线程数小于 corePoolSize 核心线程数则创建新的线程; 如果大于 corePoolSize 核心线程数则加入等待队列。如果队列已满则无法加入,在线程数小于 maxinumPoolSize 指定的最大线程数前提下会创建新的线程来执 行, 如果线程数大于 maxinumPoolSize 最大线程数则执行拒绝策略.
ArrayBlockingQueue是一个有界缓存等待队列，可以指定缓存队列的大小，当正在执行的线程数等于corePoolSize时，多余的元素缓存在ArrayBlockingQueue队列中等待有空闲的线程时继续执行，当ArrayBlockingQueue已满时，加入ArrayBlockingQueue失败，会开启新的线程去执行，当线程数已经达到最大的maximumPoolSizes时，再有新的元素尝试加入ArrayBlockingQueue时会报错。
3.LinkedBlockingQueue
3) 无界任务队列,由 LinkedBlockingQueue 对象实现,与有界队列相比,除非系统资源耗尽,否则无界队列不存在任务入队失败的情况. 当有新的任务时,在系统线程数小于 corePoolSize 核心线程数则创建新的线程来执行任务;当线程池中线程数量大于corePoolSize 核心线程数则把任务加入阻塞队列.
LinkedBlockingQueue是一个无界缓存等待队列。当前执行的线程数量达到corePoolSize的数量时，剩余的元素会在阻塞队列里等待。（所以在使用此阻塞队列时maximumPoolSizes就相当于无效了），每个线程完全独立于其他线程。生产者和消费者使用独立的锁来控制数据的同步，即在高并发的情况下可以并行操作队列中的数据。
4.PriorityBlockingQueue
4) 优先任务队列,由 PriorityBlockingQueue 实现的,是带有任务优先级的队列, 是一个特殊的无界队列. 不管是
ArrayBlockingQueue 队列还是 LinkedBlockingQueue 队列都是按照先进先出算法处理任务的.在PriorityBlockingQueue 队列中可以根据任务优先级顺序先后执行.
8.4.2 拒绝策略
ThreadPoolExecutor 构造方法的最后一个参数指定了拒绝策略，当提交给线程池的任务量超过实际承载能力时，如何处理? 即线程池中的线程已经用完了，等待队列也满了，无法为新提交的任务服务，可以通过拒绝策略来处理这个问题。
JDK 提供了四种拒绝策略：
AbortPolicy 策略，会抛出异常
CallerRunsPolicy 策略，只要线程池没关闭，会在调用者线程中运行当前被丢弃的任务，交给调用者执行；??：main方法中创建线程池，就会使用main方法执行任务
DiscardOldestPolicy 策略，将任务队列中最老的任务丢弃，尝试再次提交新任务
DiscardPolicy 策略，直接丢弃这个无法处理的任务，不抛异常
这个四个策略是ThreadPoolExecutor中静态内部类，都实现了RejectedExecutionHandler接口。
Executors 工具类提供的静态方法返回的线程池默认的拒绝策略是 AbortPolicy 抛出异常。
ThreadPoolExecutor源码：
/**
* The default rejected execution handler
*/
private static final RejectedExecutionHandler defaultHandler = new AbortPolicy();
/**
* 被拒绝任务的处理程序，它抛出一个RejectedExecutionException 异常
*/
public static class AbortPolicy implements RejectedExecutionHandler {
public AbortPolicy() { }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
throw new RejectedExecutionException("Task " + r.toString() +
" rejected from " +
e.toString());
}
}
如果内置的拒绝策略无法满足实际需求，可以扩展 RejectedExecutionHandler 接口进行自定义。
/**
* 自定义拒绝策略
*/
public class Test03 {
public static void main(String[] args) {
//定义任务
Runnable r = new Runnable() {
@Override
public void run() {
int num = new Random().nextInt(5);
System.out.println(Thread.currentThread().getId() + "--" +
System.currentTimeMillis() + "开始睡眠" + num + "秒");
try {
TimeUnit.SECONDS.sleep(num);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
};
//创建线程池, 自定义拒绝策略
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 2, 0,
TimeUnit.SECONDS, new LinkedBlockingQueue<>(5), Executors.defaultThreadFactory(),
new RejectedExecutionHandler() {
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
//r 就是请求的任务, executor 就是当前线程池
System.out.println(r + " is discarding====");
}
});
//向线程池提交若干任务
for (int i = 0; i < Integer.MAX_VALUE; i++) {
threadPoolExecutor.submit(r);
}
}
}
8.4.3 ThreadFactory
线程池中的线程来自线程工厂，ThreadFactory 是一个接口，只有一个用来创建线程的方法:
public interface ThreadFactory {
/**
* Constructs a new {@code Thread}. Implementations may also initialize
* priority, name, daemon status, {@code ThreadGroup}, etc.
*
* @param r a runnable to be executed by new thread instance
* @return constructed thread, or {@code null} if the request to
* create a thread is rejected
*/
Thread newThread(Runnable r);
}
当线程池中需要创建线程时就会调用该方法。
/**
* 自定义线程工厂
*/
public class Test04 {
public static void main(String[] args) throws InterruptedException {
//定义任务
Runnable r = new Runnable() {
@Override
public void run() {
int num = new Random().nextInt(10);
System.out.println(Thread.currentThread().getId() + "--" +
System.currentTimeMillis() + "开始睡眠:" + num + "秒");
try {
TimeUnit.SECONDS.sleep(num);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
};
//创建线程池, 使用自定义线程工厂, 采用默认的拒绝策略是抛出异常
ExecutorService executorService = new ThreadPoolExecutor(5, 5, 0,
TimeUnit.SECONDS, new SynchronousQueue<>(),
new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
//根据参数 r 接收的任务,创建一个线程
Thread t = new Thread(r);
t.setDaemon(true); //设置为守护线程, 当主线程运行结束,线程池中的线程会自动退出
System.out.println("创建了线程: " + t);
return t;
}
});
//提交 5 个任务, 当给当前线程池提交的任务超过 5 个时,线程池默认抛出异常
for (int i = 0; i < 5; i++) {
executorService.submit(r);
}
//主线程睡眠
Thread.sleep(10000);
//主线程睡眠超时, 主线程结束, 线程池中的线程会自动退出
}
}
8.5 线程池原理剖析
提交一个任务到线程池中，线程池的处理流程如下：
1、判断线程池里的核心线程是否都在执行任务，如果不是（核心线程空闲或者还有核心线程没有被创建）则创建一个新的工作线程来执行任务。如果核心线程都在执行任务，则进入下个流程。
2、线程池判断工作队列是否已满，如果工作队列没有满，则将新提交的任务存储在这个工作队列里。如果工作队列满了，则进入下个流程。
3、判断线程池里的线程是否都处于工作状态，如果没有，则创建一个新的工作线程来执行任务。如果已经满了，则交给饱和策略来处理这个任务。
8.6 Executors线程池工具类
Executors是线程池的工具类，提供了四种快捷创建线程池的方法：
newCachedThreadPool
创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。
newFixedThreadPool
创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。
newSingleThreadExecutor
创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。
newScheduledThreadPool
创建一个定长线程池，支持定时及周期性任务执行。
Excutors 工具类中返回线程池的方法底层都使用了ThreadPoolExecutor 线程池.
ExecutorService newFixedThreadPool() : 创建固定大小的线程池。
ExecutorService newCachedThreadPool() : 缓存线程池，线程池的数量不固定，可以根据需求自动的更改数量。
ExecutorService newSingleThreadExecutor() : 创建单个线程池，线程池中只有一个线程。
ScheduledExecutorService newScheduledThreadPool() : 创建固定大小的线程，可以延迟或定时的执行任务。
8.6.1 newCachedThreadPool
介绍
CachedThreadPool是一个"无限"容量的线程池，它会根据需要创建新线程。特点是可以根据需要来创建新的线程执行任务，没有特定的corePool。
创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。
构造源码
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue());
}
- CachedThreadPool的corePoolSize被设置为0，即corePool为空；maximumPoolSize被设置为Integer.MAX_VALUE，即maximum是无界的。这里keepAliveTime设置为60秒，意味着空闲的线程最多可以等待任务60秒，否则将被回收。
- CachedThreadPool使用没有容量的SynchronousQueue作为主线程池的工作队列，它是一个没有容量的阻塞队列。每个插入操作必须等待另一个线程的对应移除操作。这意味着，如果主线程提交任务的速度高于线程池中处理任务的速度时，CachedThreadPool会不断创建新线程。极端情况下，CachedThreadPool会因为创建过多线程而耗尽CPU资源
执行示意图
执行过程如下：
-
首先执行SynchronousQueue.offer(Runnable task)。如果在当前的线程池中有空闲的线程正在执行SynchronousQueue.poll()，那么主线程执行的offer操作与空闲线程执行的poll操作配对成功，主线程把任务交给空闲线程执行。，execute()方法执行成功，否则执行步骤2
-
当线程池为空(初始maximumPool为空)或没有空闲线程时，配对失败，将没有线程执行SynchronousQueue.poll操作。这种情况下，线程池会创建一个新的线程执行任务。
-
在创建完新的线程以后，将会执行poll操作。当步骤2的线程执行完成后，将等待60秒，如果此时主线程提交了一个新任务，那么这个空闲线程将执行新任务，否则被回收。因此长时间不提交任务的CachedThreadPool不会占用系统资源。
SynchronousQueue是一个不存储元素阻塞队列，每次要进行offer操作时必须等待poll操作，否则不能继续添加元素。
示例代码
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class ThreadPoolExecutorTest4_newCachedThreadPool {
public static void main(String[] args) {
ExecutorService executorService = Executors.newCachedThreadPool();
for (int i = 0; i < 10; i++) {
executorService.execute(()->{
System.out.println(Thread.currentThread().getName());
});
}
}
}
运行代码
10个任务就创建了10个线程
8.6.2 newFixedThreadPool
创建固定长度的线程池，每次提交任务创建一个线程，直到达到线程池的最大数量，线程池的大小不再变化。
这个线程池可以创建固定线程数的线程池。特点就是可以重用固定数量线程的线程池
创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。示例代码如下：
public static void main(String[] args) {
// 创建线程池
ExecutorService es = Executors.newFixedThreadPool(2);
// 会创建出10个线程 分别执行任务
for (int i = 0; i < 10; i++) {
es.execute(()->{
try {
Thread.sleep(400);
} catch (InterruptedException e) {
e.printStackTrace();
}
for (int j = 0; j < 10; j++) {
System.out.println(Thread.currentThread().getName() + ":" + j);
}
});
}
es.shutdown();
}
定长线程池的大小最好根据系统资源进行设置。如Runtime.getRuntime().availableProcessors()
构造源码
Executors.newFixedThreadPool(int)；
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads, 0L,
TimeUnit.MILLISECONDS,
new LinkedBlockingQueue());
}
- FixedThreadPool的corePoolSize和maxiumPoolSize都被设置为创建FixedThreadPool时指定的参数nThreads。
- 当线程池中的线程数大于corePoolSize时，keepAliveTime为多余的空闲线程等待新任务的
最长时间，超过这个时间后多余的线程将被终止。这里把keepAliveTime设置为0L，意味着多余
的空闲线程会被立即终止。 - 最后一个参数表示FixedThreadPool使用了无界队列LinkedBlockingQueue作为线程池的做工队列，由于是无界的，当线程池的线程数达到corePoolSize后，新任务将在无界队列中等待，因此线程池的线程数量不会超过corePoolSize，同时maxiumPoolSize也就变成了一个无效的参数，并且运行中的线程池并不会拒绝任务。
FixedThreadPool 执行示意图
执行过程如下：
1.如果当前工作中的线程数量少于corePool的数量，就创建新的线程来执行任务。
2.当线程池的工作中的线程数量达到了corePool，则将任务加入LinkedBlockingQueue。
3.线程执行完1中的任务后会从队列中去任务。
注意：LinkedBlockingQueue是无界队列，所以可以一直添加新任务到线程池。
示例代码
import java.util.concurrent.*;
import java.util.concurrent.ThreadPoolExecutor.AbortPolicy;
public class ThreadPoolExecutorTest2_newFixedThreadPool {
public static void main(String[] args) {
ExecutorService executorService = Executors.newFixedThreadPool(3);
for (int i = 0; i < 10; i++) {
executorService.execute(()->{
System.out.println(Thread.currentThread().getName());
});
}
}
}
运行效果
上面只有数字1,2,3号线程，说明线程池里面有3个线程
8.6.3 newSingleThreadExecutor
介绍
SingleThreadExecutor是使用单个worker线程的Executor。特点是使用单个工作线程执行任务
构造源码
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue()));
}
SingleThreadExecutor的corePoolSize和maxiumPoolSize都被设置1。 其他参数均与FixedThreadPool相同
SingleThreadExecutor执行示意图
执行过程如下：
1.如果当前工作中的线程数量少于corePool的数量，就创建一个新的线程来执行任务。
2.当线程池的工作中的线程数量达到了corePool，则将任务加入LinkedBlockingQueue。
3.线程执行完1中的任务后会从队列中去任务。
注意：由于在线程池中只有一个工作线程，所以任务可以按照添加顺序执行。
示例代码
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class ThreadPoolExecutorTest3_newSingleThreadExecutor {
public static void main(String[] args) {
ExecutorService executorService = Executors.newSingleThreadExecutor();
for (int i = 0; i < 10; i++) {
executorService.execute(()->{
System.out.println(Thread.currentThread().getName());
});
}
}
}
运行效果
8.6.4 newScheduledThreadPool
创建一个定长线程池，支持定时及周期性任务执行。
1.基本使用
介绍
ScheduledExecutorService在普通执行器接口（ExecutorService）的基础上引入了Future模式，使得可以限时延迟或周期性地调度任务。
api方法
创建周期线程池构造函数
Executors使用newScheduledThreadPool工厂方法创建ScheduledThreadPoolExecutor
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
return new ScheduledThreadPoolExecutor(corePoolSize);
}
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize, ThreadFactory threadFactory) {
return new ScheduledThreadPoolExecutor(corePoolSize, threadFactory);
}
构造函数源码
public ScheduledThreadPoolExecutor(int corePoolSize) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS, new DelayedWorkQueue());
}
public ScheduledThreadPoolExecutor(int corePoolSize, ThreadFactory threadFactory) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS, new DelayedWorkQueue(), threadFactory);
}
public ScheduledThreadPoolExecutor(int corePoolSize, RejectedExecutionHandler handler) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS, new DelayedWorkQueue(), handler);
}
public ScheduledThreadPoolExecutor(int corePoolSize, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS, new DelayedWorkQueue(), threadFactory, handler);
}
这里最需要注意的就是任务队列的选择——DelayedWorkQueue，底层基于优先无界阻塞队列（PriorityQueue）实现
类关系图
从上图中可以看到，ScheduledThreadPoolExecutor其实是继承了ThreadPoolExecutor这个普通线程池，我们知道ThreadPoolExecutor中提交的任务都是实现了Runnable接口，但是ScheduledThreadPoolExecutor比较特殊，由于要满足任务的延迟/周期调度功能，它会对所有的Runnable任务都进行包装，包装成一个RunnableScheduledFuture任务.
延迟/周期执行任务的api方法
ScheduledFuture<?> scheduleAtFixedRate(Runnable command, //执行的任务
long initialDelay,//延迟指定时间执行
long period, //周期或间隔固定时间循环执行
TimeUnit unit) //设置时间单位
源码
RunnableScheduledFuture是Future模式中的一个接口，关于Future模式，我们后续讲解，这里只要知道RunnableScheduledFuture的作用就是可以异步地执行【延时/周期任务】。
另外，我们知道在ThreadPoolExecutor中，需要指定一个阻塞队列作为任务队列。ScheduledThreadPoolExecutor中也一样，不过特殊的是，ScheduledThreadPoolExecutor中的任务队列是一种特殊的延时队列（DelayQueue）。
我们曾经在juc中，分析过该种阻塞队列，DelayQueue底层基于优先无界阻塞队列（PriorityQueue）实现，通过该种阻塞队列可以实现任务的延迟到期执行（即每次从队列获取的任务都是最先到期的任务）。
ScheduledThreadPoolExecutor在内部定义了DelayQueue的变种—DelayedWorkQueue，它和DelayQueue类似，只不过要求所有入队元素必须实现RunnableScheduledFuture接口
代码实现
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
public class _04DemoTest {
public static void main(String[] args) {
//1.创建延时或周期的线程池，设置核心线程数2个
ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(2);
//2.循环运行10个任务
for (int i = 0; i < 2; i++) {
scheduledExecutorService.scheduleAtFixedRate(
()->{
System.out.println(Thread.currentThread().getName());
}, //任务
2,//执行任务的延迟秒数
5,//设置周期执行任务的秒数
TimeUnit.SECONDS //设置时间的单位
);
}
}
}
运行效果
2.运行机制
三种调度执行方式
-
ScheduledThreadPoolExecutor的schedule()方法方式
-
ScheduledThreadPoolExecutor的scheduleAtFixedRate()方法方式
-
ScheduledThreadPoolExecutor的scheduleWithFixedDelay方法方式
2个方式都会向ScheduledThreadPoolExecutor的DelayQueue添加一个实现了
RunnableScheduledFutur接口的ScheduledFutureTask。线程池中的线程从DelayQueue中获取ScheduledFutureTask，然后执行任务。DelayQueue是一个无界队列，所以ThreadPoolExecutor的maximumPoolSize在Scheduled-
ThreadPoolExecutor中没有什么意义（设置maximumPoolSize的大小没有什么效果）。
schedule方法调度源码分析
public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
if (command == null || unit == null) {
throw new NullPointerException();
}
RunnableScheduledFuture<?> t = decorateTask(command, new ScheduledFutureTask(command, null, triggerTime(delay, unit)));
delayedExecute(t);
return t;
}
上述的decorateTask方法把Runnable任务包装成ScheduledFutureTask
protected RunnableScheduledFuture decorateTask(Runnable runnable,
RunnableScheduledFuture task) {
return task;
}
注意：ScheduledFutureTask是RunnableScheduledFuture接口的实现类，任务通过period字段来表示任务类型
private class ScheduledFutureTask extends FutureTask implements RunnableScheduledFuture {
/**
* 任务序号, 自增唯一
*/
private final long sequenceNumber;
/**
* 首次执行的时间点
*/
private long time;
/**
* 0: 非周期任务
* >0: fixed-rate任务
* <0: fixed-delay任务
*/
private final long period;
/**
* 在堆中的索引
*/
int heapIndex;
ScheduledFutureTask(Runnable r, V result, long ns) {
super(r, result);
this.time = ns;
this.period = 0;
this.sequenceNumber = sequencer.getAndIncrement();
}
// ...
}
ScheduledThreadPoolExecutor中的任务队列——DelayedWorkQueue，保存的元素就是ScheduledFutureTask。DelayedWorkQueue是一种堆结构，time最小的任务会排在堆顶（表示最早过期），每次出队都是取堆顶元素，这样最快到期的任务就会被先执行。如果两个ScheduledFutureTask的time相同，就比较它们的序号——sequenceNumber，序号小的代表先被提交，所以就会先执行。
schedule的核心是其中的delayedExecute方法：
private void delayedExecute(RunnableScheduledFuture<?> task) {
if (isShutdown()) // 线程池已关闭
reject(task); // 任务拒绝策略
else {
super.getQueue().add(task); // 将任务入队
// 如果线程池已关闭且该任务是非周期任务, 则将其从队列移除
if (isShutdown() && !canRunInCurrentRunState(task.isPeriodic()) && remove(task))
task.cancel(false); // 取消任务
else
ensurePrestart(); // 添加一个工作线程
}
}
执行示意图
通过delayedExecute可以看出，ScheduledThreadPoolExecutor的整个任务调度流程大致如下图
-
首先，任务被提交到线程池后，会判断线程池的状态，如果不是RUNNING状态会执行拒绝策略。
-
然后，将任务添加到阻塞队列中。
（注意，由于DelayedWorkQueue是无界队列，所以一定会add成功）
-
然后，会创建一个工作线程，加入到线程池：
void ensurePrestart() { int wc = workerCountOf(ctl.get()); if (wc < corePoolSize) addWorker(null, true); else if (wc == 0) addWorker(null, false); }通过ensurePrestart()方法可以看到，如果核心线程池未满，则新建的工作线程会被放到核心线程池中。如果核心线程池已经满了，ScheduledThreadPoolExecutor不会像ThreadPoolExecutor那样再去创建归属于非核心线程池的工作线程，而是直接返回。也就是说，在ScheduledThreadPoolExecutor中，一旦核心线程池满了，就不会再去创建工作线程。
什么时候会执行else if (wc == 0)创建一个归属于非核心线程池的工作线程？
答案是，当通过setCorePoolSize方法设置核心线程池大小为0时，这里必须要保证任务能够被执行，所以会创建一个工作线程，放到非核心线程池中。
最后，线程池中的工作线程会去任务队列获取任务并执行，当任务被执行完成后，如果该任务是周期任务，则会重置time字段，并重新插入队列中，等待下次执行。
3.DelayedWorkQueue 延迟队列
介绍
DelayedWorkQueue，该队列和已经介绍过的DelayQueue区别不大，只不过队列元素是RunnableScheduledFuture
static class DelayedWorkQueue extends AbstractQueue implements BlockingQueue {
private static final int INITIAL_CAPACITY = 16;
private RunnableScheduledFuture<?>[] queue = new RunnableScheduledFuture<?>[INITIAL_CAPACITY];
private int size = 0;
private final ReentrantLock lock = new ReentrantLock();
private final Condition available = lock.newCondition();
private Thread leader = null;
// ...
}
DelayedWorkQueue是一个无界队列，在队列元素满了以后会自动扩容，它并没有像DelayQueue那样，将队列操作委托给PriorityQueue，而是自己重新实现了一遍堆的核心操作——上浮、下沉。
我们关键来看下add、take、poll这三个队列方法，因为ScheduledThreadPoolExecutor的核心调度流程中使用到了这三个方法：
public boolean add(Runnable e) {
return offer(e);
}
public boolean offer(Runnable e, long timeout, TimeUnit unit) {
return offer(e);
}
add、offer内部都调用了下面这个方法：
public 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; // 队列已满, 扩容
if (i >= queue.length)
grow();
size = i + 1;
if (i == 0) {
queue[0] = e;
setIndex(e, 0);
} else {
siftUp(i, e); // 堆上浮操作
}
if (queue[0] == e) { // 当前元素是首个元素
leader = null;
available.signal(); // 唤醒一个等待线程
}
} finally {
lock.unlock();
}
return true;
}
take方法：
public RunnableScheduledFuture<?> take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
for (; ; ) {
RunnableScheduledFuture<?> first = queue[0];
if (first == null) // 队列为空
available.await(); // 等待元素入队
else {
long delay = first.getDelay(NANOSECONDS);
if (delay <= 0) // 元素已到期
return finishPoll(first);
// 执行到此处, 说明队首元素还未到期
first = null;
if (leader != null)
available.await();
else {
// 当前线程成功leader线程
Thread thisThread = Thread.currentThread();
leader = thisThread;
try {
available.awaitNanos(delay);
} finally {
if (leader == thisThread)
leader = null;
}
}
}
}
} finally {
if (leader == null && queue[0] != null)
available.signal();
lock.unlock();
}
}
每次出队元素时，如果队列为空或者队首元素还未到期，线程就会在condition条件队列等待。一般的思路是无限等待，直到出现一个入队线程，入队元素后将一个出队线程唤醒。
为了提升性能，当队列非空时，用leader保存第一个到来并尝试出队的线程，并设置它的等待时间为队首元素的剩余期限，这样当元素过期后，线程也就自己唤醒了，不需要入队线程唤醒。这样做的好处就是提升一些性能。
3.6.小结
-
ScheduledThreadPoolExecutor的介绍与主要特点
它是对普通线程池ThreadPoolExecutor的扩展，增加了延时调度、周期调度任务的功能。
概括下ScheduledThreadPoolExecutor的主要特点：
-
对Runnable任务进行包装，封装成
ScheduledFutureTask，该类任务支持任务的周期执行、延迟执行； -
采用
DelayedWorkQueue作为任务队列。该队列是无界队列，所以任务一定能添加成功，但是当工作线程尝试从队列取任务执行时，只有最先到期的任务会出队，如果没有任务或者队首任务未到期，则工作线程会阻塞； -
ScheduledThreadPoolExecutor的任务调度流程与ThreadPoolExecutor略有区别，最大的区别就是，先往队列添加任务，然后创建工作线程执行任务。
-
-
java实现定时器周期执行任务的技术有哪些？有什么区别？
ScheduledThreadPoolExecutor是多线程的周期执行任务，Timer是单线程周期执行任务；ScheduledThreadPoolExecutor性能高于Timer
小结
-
ThreadPoolExecutor包含几种线程池类型
- FixedThreadPool，有界固定大小线程池
- SingleThreadExecutor，有界单个线程串行执行的线程池
- CachedThreadPool，无界线程池
-
FixedThreadPool
适用于为了满足资源管理需求，而需要限制当前线程的数量的应用场景，它适用于负载比较重的服务器。
-
SingleThreadExecutor
适用于需要保证执行顺序地执行各个任务；并且在任意时间点，不会有多个线程是活动的场景。
-
CachedThreadPool
大小无界的线程池，适用于执行很多的短期异步任务的小程序，或者负载较轻的服务器。
8.6.5 向线程池提交任务
介绍
可以使用两个方法向线程池提交任务，分别为execute()和submit()方法。
execute方法
threadsPool.execute(
()->{
//任务代码
}
);
execute()方法用于提交不需要返回值的任务，所以无法判断任务是否被线程池执行成功。
通过以下代码可知execute()方法输入的任务是一个Runnable类的实例。
submit方法
Futuresubmit()方法用于提交需要返回值的任务。线程池会返回一个future类型的对象，通过这个future对象可以判断任务是否执行成功，并且可以通过future的get()方法来获取返回值，get()方法会阻塞当前线程直到任务完成，而使用get（long timeout，TimeUnit unit）方法则会阻塞当前线程一段时间后立即返回，这时候有可能任务没有执行完。
8.7 监控线程池API
ThreadPoolExecutor 提供了一组方法用于监控线程池。
int getActiveCount() 获得线程池中当前活动线程的数量
long getCompletedTaskCount() 返回线程池完成任务的数量
int getCorePoolSize() 线程池中核心线程的数量
int getLargestPoolSize() 返回线程池曾经达到的线程的最大数
int getMaximumPoolSize() 返回线程池的最大容量
int getPoolSize() 当前线程池的大小
BlockingQueue getQueue() 返回阻塞队列
long getTaskCount() 返回线程池收到的任务总数
??：java在线执行
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
/**
* 监控线程池
*/
public class Test05 {
public static void main(String[] args) throws InterruptedException {
//先定义任务
Runnable r = new Runnable() {
@Override
public void run() {
System.out.println(Thread.currentThread().getId() + " 编号的线程开始执行: " + System.currentTimeMillis());
try {
Thread.sleep(5000); //线程睡眠 20 秒,模拟任务执行时长
} catch (InterruptedException e) {
e.printStackTrace();
}
}
};
//定义线程池
ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(1, 2, 0,
TimeUnit.SECONDS, new ArrayBlockingQueue<>(2), Executors.defaultThreadFactory(),
new ThreadPoolExecutor.DiscardPolicy());
//向线程池提交 30 个任务
for (int i = 0; i < 10; i++) {
poolExecutor.submit(r);
System.out.println(" 当前线程池核心线程数量: " + poolExecutor.getCorePoolSize()
+ ",最大线程数:" + poolExecutor.getMaximumPoolSize()
+ ",当前线程池大小:" + poolExecutor.getPoolSize()
+ ",活动线程数量:" + poolExecutor.getActiveCount()
+ ",收到任务数量:" + poolExecutor.getTaskCount()
+ ",完成任务数: " + poolExecutor.getCompletedTaskCount()
+ ",等待任务数:" + poolExecutor.getQueue().size());
TimeUnit.MILLISECONDS.sleep(500);
}
System.out.println("-----------------------------------------------");
while (poolExecutor.getActiveCount() >= 0) {
System.out.println(" 当前线程池核心线程数量: " + poolExecutor.getCorePoolSize()
+ ",最大线程数:" + poolExecutor.getMaximumPoolSize()
+ ",当前线程池大小 :" + poolExecutor.getPoolSize()
+ ",活动线程数量 :" + poolExecutor.getActiveCount()
+ ",收到任务数量:" + poolExecutor.getTaskCount()
+ ",完成任务数 : " + poolExecutor.getCompletedTaskCount()
+ ",等待任务数 :" + poolExecutor.getQueue().size());
Thread.sleep(1000);
}
}
}
8.8 扩展线程池
有时需要对线程池进行扩展，如在监控每个任务的开始和结束时间，或者自定义一些其他增强的功能。
ThreadPoolExecutor 线程池提供了两个方法：
protected void afterExecute(Runnable r, Throwable t) { }
protected void beforeExecute(Thread t, Runnable r) { }
在线程池执行某个任务前会调用 beforeExecute()方法，在任务结束后(任务异常退出)会执行 afterExecute()方法。
查看 ThreadPoolExecutor 源码，在该类中定义了一个内部类 Worker。
/** Delegates main run loop to outer runWorker */
public void run() {
runWorker(this);
}
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
while (task != null || (task = getTask()) != 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 = null;
try {
task.run();
} catch (RuntimeException x) {
thrown = x; throw x;
} catch (Error x) {
thrown = x; throw x;
} catch (Throwable x) {
thrown = x; throw new Error(x);
} finally {
afterExecute(task, thrown);
}
} finally {
task = null;
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
}
}
ThreadPoolExecutor 线程池中的工作线程就是 Worker 类的实例，Worker 实例在执行时会调用 beforeExecute()与 afterExecute()方法。
举个??：
/**
* 扩展线程池
*/
public class Test06 {
//定义任务类
private static class MyTask implements Runnable {
String name;
public MyTask(String name) {
this.name = name;
}
@Override
public void run() {
System.out.println(name + "任务正在被线程 " + Thread.currentThread().getId() + " 执行");
try {
Thread.sleep(1000); //模拟任务执行时长
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
public static void main(String[] args) {
//定义扩展线程池 , 可以定义线程池类继承 ThreadPoolExecutor,在子类中重写beforeExecute()/afterExecute()方法
//也可以直接使用 ThreadPoolExecutor 的内部类
ExecutorService executorService = new ThreadPoolExecutor(5, 5, 0,
TimeUnit.SECONDS, new LinkedBlockingQueue<>()) {
//在内部类中重写任务开始方法
@Override
protected void beforeExecute(Thread t, Runnable r) {
System.out.println(t.getId() + "线程准备执行任务: " + ((MyTask) r).name);
}
@Override
protected void afterExecute(Runnable r, Throwable t) {
System.out.println(((MyTask) r).name + "任务执行完毕");
}
@Override
protected void terminated() {
System.out.println("线程池退出");
}
};
//向线程池中添加任务
for (int i = 0; i < 5; i++) {
MyTask task = new MyTask("task-" + i);
executorService.execute(task);
}
//关闭线程池
executorService.shutdown();
//关闭线程池仅仅是说线程池不再接收新的任务 , 线程池中已接收的任务正常执行完毕
}
}
8.9 优化线程池大小
线程池大小对系统性能是有一定影响的，过大或者过小都会无法发挥最优的系统性能，线程池大小不需要非常精确，只要避免极大或者极小的情况即可，一般来说，线程池大小需要考虑 CPU 数量，内存大小等因素。
在《Java Concurrency in Practice》书中给出一个估算线程池大小的公式:
线程池大小 = CPU 的数量 * 目标 CPU 的使用率 * ( 1 + 等待时间与计算时间的比)
8.10 线程池死锁
如果在线程池中执行的 任务 A 在执行过程中又向线程池提交了任务 B，任务 B 添加到了线程池的等待队列中，如果任务 A 的结束需要等待任务 B 的执行结果，就有可能会出现这种情况：
线程池中所有的工作线程都处于等待任务处理结果，而这些任务在阻塞队列中等待执行，线程池中没有可以对阻塞队列中的任务进行处理的线程，这种等待会一直持续下去，从而造成死锁。
适合给线程池提交相互独立的任务，而不是彼此依赖的任务。对于彼此依赖的任务，可以考虑分别提交给不同的线程池来执行。
8.11 线程池中的异常处理
在使用 ThreadPoolExecutor 进行 submit 提交任务时，有的任务抛出了异常，但是线程池并没有进行提示，即线程池把任务中的异常给吞掉了，可以把 submit 提交改为 execute 执行；也可以对 ThreadPoolExecutor线程池进行扩展.对提交的任务进行包装。
/**
* 线程池异常处理
*
*/
public class Test07 {
static class DivideTask implements Runnable {
private int a;
public DivideTask(int a) {
this.a = a;
}
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + "执行 10/" + a + " = " + 10 / a);
}
}
public static void main(String[] args) {
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(1, 5, 5
, TimeUnit.SECONDS, new SynchronousQueue());
/*
* submit()方法 运行程序，只有四条计算结果，实际上向线程池提交了5个计算任务，
* 分析结果发现当i == 0时，提交的任务会产生算术异常，线程池
* 把该异常吞掉了，导致我们对该异常一无所知
*
* 解决方法：
* ① 把submit()方法替换成 execute() 方法执行
* ② 对线程池进行扩展，对submit()方法进行包装
* */
for (int i = 0; i < 5; i++) {
// threadPoolExecutor.submit(new DivideTask(i));
threadPoolExecutor.execute(new DivideTask(i));
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
自定义线程池类，对 ThreadPoolExecutor 进行扩展：
/**
* 自定义线程池类,对 ThreadPoolExecutor 进行扩展
*/
public class Test08 {
//自定义线程池类
private static class TraceThreadPollExecutor extends ThreadPoolExecutor {
public TraceThreadPollExecutor(int corePoolSize, int maximumPoolSize, long
keepAliveTime, TimeUnit unit, BlockingQueue workQueue) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
}
//定义方法,对执行的任务进行包装,接收两个参数,第一个参数接收要执行的任务,第二个参数是一个 Exception 异常
public Runnable wrap(Runnable task, Exception exception) {
return new Runnable() {
@Override
public void run() {
try {
task.run();
} catch (Exception e) {
exception.printStackTrace();
throw e;
}
}
};
}
//重写 submit 方法
@Override
public Future<?> submit(Runnable task) {
return super.submit(wrap(task, new Exception("客户跟踪异常")));
}
@Override
public void execute(Runnable command) {
super.execute(wrap(command, new Exception("客户跟踪异常")));
}
}
//定义类实现 Runnable 接口,用于计算两个数相除
private static class DivideTask implements Runnable {
private int x;
private int y;
public DivideTask(int x, int y) {
this.x = x;
this.y = y;
}
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + "计算:" + x + " / " + y + " = " + (x / y));
}
}
public static void main(String[] args) {
//创建线程池
// ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(0, Integer.MAX_VALUE,
// 0, TimeUnit.SECONDS, new SynchronousQueue<>());
//使用自定义的线程池
ThreadPoolExecutor poolExecutor = new TraceThreadPollExecutor(0,
Integer.MAX_VALUE, 0, TimeUnit.SECONDS, new SynchronousQueue<>());
//向线程池中添加计算两个数相除的任务
for (int i = 0; i < 5; i++) {
poolExecutor.submit(new DivideTask(10, i));
// poolExecutor.execute(new DivideTask(10, i));
}
}
}
8.12 ForkJoinPool 线程池
“分而治之”是一个有效的处理大数据的方法，著名的 MapReduce就是采用这种分而治之的思路。简单点说，如果要处理的 1000 个数据，但是我们不具备处理 1000个数据的能力，可以只处理 10个数据，可以把这 1000 个数据分阶段处理 100 次，每次处理 10 个，把 100 次的处理结果进行合成，形成最后这 1000 个数据的处理结果。
把一个大任务调用 fork()方法分解为若干小的任务，把小任务的处理结果进行 join()合并为大任务的结果。
系统对 ForkJoinPool 线程池进行了优化，提交的任务数量与线程的数量不一定是一对一关系。在多数情况下，一个物理线程实际上需要处理多个逻辑任务。
Java8-API文档
ForkJoinPool 线程池中最常用的方法是：
ForkJoinTask submit(ForkJoinTask task) // 提交一个ForkJoinTask执行
向线程池提交一个 ForkJoinTask 任务，ForkJoinTask 任务支持fork()分解与join()等待的任务。
ForkJoinTask 有两个重要的子类：
RecursiveAction, RecursiveTask
它们的区别在于RecursiveAction 任务没有返回值；RecursiveTask 任务可以带有返回值。
4.FutureTask详解
4.1.目标
-
理解什么是FutureTask
-
理解FutureTask的应用场景
4.2.FutureTask基础知识
FutureTask简介
前面我们提到的 Future 是一个接口，而 FutureTask 是接口实现类，
它可以获取异步任务计算的结果和取消此任务的执行
api介绍
创建FutureTask 构造函数
//构造方法一
public FutureTask(Callable callable) {
if (callable == null)
throw new NullPointerException();
this.callable = callable;
this.state = NEW; // ensure visibility of callable
}
//构造方法二
public FutureTask(Runnable runnable, V result) {
//将runnable包装成callable
this.callable = Executors.callable(runnable, result);
this.state = NEW; // ensure visibility of callable
}
第一个构造方法我们比较熟悉，第二个构造方法可以用 Runnable 构造 FutureTask，将 Runnable 使用适配器模式 构造成 FutureTask ，使其具有 FutureTask 的特性，如可在主线程捕获Runnable的子线程异常。
获取异步任务结果
V get();
V get(long timeout, TimeUnit unit); //设置超时时间获取结果
取消异步任务执行
boolean cancel(boolean mayInterruptIfRunning);//传递true代表取消
8.13 FutureTask的使用
使用方式1：使用线程池执行FutureTask任务
FutureTask 实现了 Runnable 和 Future 接口，由于实现了 Runnable 接口，所以可以将 FutureTask 对象作为任务提交给 ThreadPoolExecutor 去执行，也可以直接被 Thread 执行；又因为实现了 Future 接口，所以也能用来获得任务的执行结果。下面的示例代码是将 FutureTask 对象提交给 ThreadPoolExecutor 去执行。
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;
public class _05DemoTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
//1.定义FutureTask任务
FutureTask futureTask = new FutureTask<>(()->1+2);
//2.实例线程池
ExecutorService executorService = Executors.newCachedThreadPool();
//3.执行任务
executorService.execute(futureTask);
//4.获取任务的结果
Integer result = futureTask.get();
System.out.println("获取的结果："+result);
//5.关闭线程池
executorService.shutdown();
}
}
使用方式2：Thread执行FutureTask任务
FutureTask 对象直接被 Thread 执行的示例代码如下所示。相信你已经发现了，利用 FutureTask 对象可以很容易获取子线程的执行结果。
import java.util.concurrent.*;
public class _06DemoTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
//1.定义FutureTask任务
FutureTask futureTask = new FutureTask<>(()->{
TimeUnit.SECONDS.sleep(3);
return 1+2;
});
//2.启动一个线程
new Thread(futureTask).start();
//4.获取任务的结果
Integer result = futureTask.get();
System.out.println("获取的结果："+result);
}
}
FutureTask的应用场景
- 当一个线程需要等待另一个线程把某个任务执行完后它才能继续执行，此时可以使用FutureTask
- FutureTask可用于异步获取执行结果或取消执行任务的场景
案例—最优的“烧水泡茶”程序
目标
掌握FutureTask 的应用场景
案例需求
下面我们来用一个例子来说明如何使用FutureTask
烧水泡茶最优工序
下面我们用程序来模拟一下这个最优工序。但之前，我们先来回顾下
并发编程可以总结为三个核心问题：分工、同步和互斥。
编写并发程序，首先要做的就是分工，所谓分工指的是如何高效地拆解任务并分配给线程。对于烧水泡茶这个程序，一种最优的分工方案可以是下图所示的这样：用两个线程 T1 和 T2 来完成烧水泡茶程序，T1 负责洗水壶、烧开水、泡茶这三道工序，T2 负责洗茶壶、洗茶杯、拿茶叶三道工序，其中 T1 在执行泡茶这道工序时需要等待 T2 完成拿茶叶的工序。对于 T1 的这个等待动作，你应该可以想出很多种办法，例如 Thread.join()、CountDownLatch，甚至阻塞队列都可以解决，不过今天我们用 Future 特性来实现。
解决方案
下面的示例代码就是用这一章提到的 Future 特性来实现的。首先，我们创建了两个 FutureTask——ft1 和 ft2，ft1 完成洗水壶、烧开水、泡茶的任务，ft2 完成洗茶壶、洗茶杯、拿茶叶的任务；这里需要注意的是 ft1 这个任务在执行泡茶任务前，需要等待 ft2 把茶叶拿来，所以 ft1 内部需要引用 ft2，并在执行泡茶之前，调用 ft2 的 get() 方法实现等待。
import java.util.concurrent.*;
public class _07DemoTest {
static class T1Task implements Callable{
private FutureTask futureTask2;
public T1Task(FutureTask futureTask2){
this.futureTask2 = futureTask2;
}
@Override
public String call() throws Exception {
System.out.println("T1:洗水壶");
TimeUnit.SECONDS.sleep(1);
System.out.println("T1:烧开水");
TimeUnit.SECONDS.sleep(15);
//获取另一个任务的结果数据
String t2Data = futureTask2.get();
System.out.println("T1:拿到了茶叶："+t2Data);
System.out.println("T1:泡茶");
return "上茶："+t2Data;
}
}
static class T2Task implements Callable{
@Override
public String call() throws Exception {
System.out.println("T2:洗茶壶");
TimeUnit.SECONDS.sleep(1);
System.out.println("T2:洗茶杯");
TimeUnit.SECONDS.sleep(1);
System.out.println("T2:拿茶叶");
TimeUnit.SECONDS.sleep(1);
return "龙井";
}
}
public static void main(String[] args) throws ExecutionException, InterruptedException {
//1.创建FutureTask任务执行任务2
FutureTask futureTask2 = new FutureTask<>(new T2Task());
//2.创建FutureTask任务执行任务1
FutureTask futureTask1 = new FutureTask<>(new T1Task(futureTask2));
//3.创建线程池
ExecutorService executorService = Executors.newCachedThreadPool();
//4.执行上面2个任务
executorService.execute(futureTask1);
executorService.execute(futureTask2);
//获取任务1的结果输出
String t1Data = futureTask1.get();
System.out.println(t1Data);
//5.关闭线程池
executorService.shutdown();
}
}
运行结果
FutureTask的实现原理
跟踪源码get方法
private int awaitDone(boolean timed, long nanos)
throws InterruptedException {
final long deadline = timed ? System.nanoTime() + nanos : 0L;
WaitNode q = null;
boolean queued = false;
for (;;) {
if (Thread.interrupted()) {
removeWaiter(q);
throw new InterruptedException();
}
int s = state;
if (s > COMPLETING) {
if (q != null)
q.thread = null;
return s;
}
else if (s == COMPLETING) // cannot time out yet
Thread.yield();
else if (q == null)
q = new WaitNode();
else if (!queued)
queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
q.next = waiters, q);
else if (timed) {
nanos = deadline - System.nanoTime();
if (nanos <= 0L) {
removeWaiter(q);
return state;
}
LockSupport.parkNanos(this, nanos);
}
else
LockSupport.park(this);
}
}
观察核心代码
发现使用LockSupport进行线程挂起与唤醒，使用CAS更新等待阻塞线程操作
小结
-
什么是FutureTask？
前面我们提到的 Future 是一个接口，而 FutureTask 是接口实现类，
它可以获取异步任务计算的结果和取消此任务的执行
-
FutureTask的应用场景？
-
当一个线程需要等待另一个线程把某个任务执行完后它才能继续执行，此时可以使用FutureTask
-
FutureTask可用于异步获取执行结果或取消执行任务的场景
-
-
面试题：什么是 Runnable、Callable、Future、FutureTask ？
-
Callable接口类似于Runnable，从名字就可以看出来了，但是Runnable不会返回结果，并且无法抛出返回结果的异常，而Callable功能更强大一些，被线程执行后，可以返回值，这个返回值可以被Future拿到，也就是说，Future可以拿到异步执行任务的返回值。可以认为是带有回调的Runnable。
-
在Java并发程序中FutureTask表示一个可以取消的异步运算。它有启动和取消运算、查询运算是否完成和取回运算结果等方法。只有当运算完成的时候结果才能取回，如果运算尚未完成get方法将会阻塞。一个FutureTask对象可以对调用了Callable和Runnable的对象进行包装
-