首页 > 基础资料 博客日记
【并发编程】Java的Future机制详解（Future接口和FutureTask类）
2024-07-11 19:00:05基础资料围观234次
目录
一、彻底理解Java的Future模式
先上一个场景：假如你突然想做饭，但是没有厨具，也没有食材。网上购买厨具比较方便，食材去超市买更放心。
实现分析：在快递员送厨具的期间，我们肯定不会闲着，可以去超市买食材。所以，在主线程里面另起一个子线程去网购厨具。
但是，子线程执行的结果是要返回厨具的，而run方法是没有返回值的。所以，这才是难点，需要好好考虑一下。
模拟代码1：
package test;
public class CommonCook {
public static void main(String[] args) throws InterruptedException {
long startTime = System.currentTimeMillis();
// 第一步 网购厨具
OnlineShopping thread = new OnlineShopping();
thread.start();
thread.join(); // 保证厨具送到
// 第二步 去超市购买食材
Thread.sleep(2000); // 模拟购买食材时间
Shicai shicai = new Shicai();
System.out.println("第二步：食材到位");
// 第三步 用厨具烹饪食材
System.out.println("第三步：开始展现厨艺");
cook(thread.chuju, shicai);
System.out.println("总共用时" + (System.currentTimeMillis() - startTime) + "ms");
}
// 网购厨具线程
static class OnlineShopping extends Thread {
private Chuju chuju;
@Override
public void run() {
System.out.println("第一步：下单");
System.out.println("第一步：等待送货");
try {
Thread.sleep(5000); // 模拟送货时间
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("第一步：快递送到");
chuju = new Chuju();
}
}
// 用厨具烹饪食材
static void cook(Chuju chuju, Shicai shicai) {}
// 厨具类
static class Chuju {}
// 食材类
static class Shicai {}
}
运行结果：
第一步：下单
第一步：等待送货
第一步：快递送到
第二步：食材到位
第三步：开始展现厨艺
总共用时7013ms
可以看到，多线程已经失去了意义。在厨具送到期间，我们不能干任何事。对应代码，就是调用join方法阻塞主线程。
有人问了，不阻塞主线程行不行？？？
不行！！！
从代码来看的话，run方法不执行完，属性chuju就没有被赋值，还是null，后面执行就会出现空指针异常。换句话说，没有厨具，怎么做饭。
Java现在的多线程机制，核心方法run是没有返回值的；如果要保存run方法里面的计算结果，必须等待run方法计算完，无论计算过程多么耗时。
面对这种尴尬的处境，程序员就会想：在子线程run方法计算的期间，能不能在主线程里面继续异步执行？？？
Where there is a will，there is a way！！！
这种想法的核心就是Future模式，下面先应用一下Java自己实现的Future模式。
模拟代码2：
package test;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;
public class FutureCook {
public static void main(String[] args) throws InterruptedException, ExecutionException {
long startTime = System.currentTimeMillis();
// 第一步 网购厨具
Callable<Chuju> onlineShopping = new Callable<Chuju>() {
@Override
public Chuju call() throws Exception {
System.out.println("第一步：下单");
System.out.println("第一步：等待送货");
Thread.sleep(5000); // 模拟送货时间
System.out.println("第一步：快递送到");
return new Chuju();
}
};
FutureTask<Chuju> task = new FutureTask<Chuju>(onlineShopping);
new Thread(task).start();
// 第二步 去超市购买食材
Thread.sleep(2000); // 模拟购买食材时间
Shicai shicai = new Shicai();
System.out.println("第二步：食材到位");
// 第三步 用厨具烹饪食材
if (!task.isDone()) { // 联系快递员，询问是否到货
System.out.println("第三步：厨具还没到，心情好就等着（心情不好就调用cancel方法取消订单）");
}
Chuju chuju = task.get();
System.out.println("第三步：厨具到位，开始展现厨艺");
cook(chuju, shicai);
System.out.println("总共用时" + (System.currentTimeMillis() - startTime) + "ms");
}
// 用厨具烹饪食材
static void cook(Chuju chuju, Shicai shicai) {}
// 厨具类
static class Chuju {}
// 食材类
static class Shicai {}
}
运行结果：
第一步：下单
第一步：等待送货
第二步：食材到位
第三步：厨具还没到，心情好就等着（心情不好就调用cancel方法取消订单）
第一步：快递送到
第三步：厨具到位，开始展现厨艺
总共用时5005ms
可以看见，在快递员送厨具的期间，我们没有闲着，可以去买食材；而且我们知道厨具到没到，甚至可以在厨具没到的时候，取消订单不要了。
好神奇，有没有。
当然你可能会说模拟代码1中，只要join在cook(thread.chuju, shicai);就可以了，但是用future应该来说从编程角度更加自然。
比如模拟二的逻辑：
if (!task.isDone()) { // 联系快递员，询问是否到货
System.out.println("第三步：厨具还没到，心情好就等着（心情不好就调用cancel方法取消订单）");
}
Chuju chuju = task.get();
等价于模拟一：
thread.start();
// 第二步 去超市购买食材
Thread.sleep(2000); // 模拟购买食材时间
Shicai shicai = new Shicai();
System.out.println("第二步：食材到位");
thread.join(); // 保证厨具送到 ---------------join应该在购买食材开始后，而不是之前
用join这种写法也可以达到future的效果，但是使用Future不仅仅是更自然，而且确实是增加了灵活性，比如任务完成与否的判断，任务的取消等。这些是join不能做到的。
下面具体分析一下第二段代码：
1）把耗时的网购厨具逻辑，封装到了一个Callable的call方法里面。
public interface Callable<V> {
/**
* Computes a result, or throws an exception if unable to do so.
*
* @return computed result
* @throws Exception if unable to compute a result
*/
V call() throws Exception;
}Callable接口可以看作是Runnable接口的补充，call方法带有返回值，并且可以抛出异常。
2）把Callable实例当作参数，生成一个FutureTask的对象，然后把这个对象当作一个Runnable，作为参数另起线程。
public class FutureTask<V> implements RunnableFuture<V>
public interface RunnableFuture<V> extends Runnable, Future<V>
public interface Future<V> {
boolean cancel(boolean mayInterruptIfRunning);
boolean isCancelled();
boolean isDone();
V get() throws InterruptedException, ExecutionException;
V get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException;
}这个继承体系中的核心接口是Future。Future的核心思想是：一个方法f，计算过程可能非常耗时，等待f返回，显然不明智。可以在调用f的时候，立马返回一个Future，可以通过Future这个数据结构去控制方法f的计算过程。
这里的控制包括：
get方法：获取计算结果（如果还没计算完，也是必须等待的）
cancel方法：还没计算完，可以取消计算过程
isDone方法：判断是否计算完
isCancelled方法：判断计算是否被取消
这些接口的设计很完美，FutureTask的实现注定不会简单，后面再说。
3）在第三步里面，调用了isDone方法查看状态，然后直接调用task.get方法获取厨具，不过这时还没送到，所以还是会等待3秒。对比第一段代码的执行结果，这里我们节省了2秒。这是因为在快递员送货期间，我们去超市购买食材，这两件事在同一时间段内异步执行。
二、为什么出现Future机制
常见的两种创建线程的方式。一种是直接继承Thread，另外一种就是实现Runnable接口。
这两种方式都有一个缺陷就是：在执行完任务之后无法获取执行结果。
从Java 1.5开始，就提供了Callable和Future，通过它们可以在任务执行完毕之后得到任务执行结果。
Future模式的核心思想是能够让主线程将原来需要同步等待的这段时间用来做其他的事情。（因为可以异步获得执行结果，所以不用一直同步等待去获得执行结果）
上图简单描述了不使用Future和使用Future的区别，不使用Future模式，主线程在invoke完一些耗时逻辑之后需要等待，这个耗时逻辑在实际应用中可能是一次RPC调用，可能是一个本地IO操作等。B图表达的是使用Future模式之后，我们主线程在invoke之后可以立即返回，去做其他的事情，回头再来看看刚才提交的invoke有没有结果。
2.1 Future 类有什么用？
Future 类是异步思想的典型运用，主要用在一些需要执行耗时任务的场景，避免程序一直原地等待耗时任务执行完成，执行效率太低。具体来说是这样的：当我们执行某一耗时的任务时，可以将这个耗时任务交给一个子线程去异步执行，同时我们可以干点其他事情，不用傻傻等待耗时任务执行完成。等我们的事情干完后，我们再通过 Future 类获取到耗时任务的执行结果。这样一来，程序的执行效率就明显提高了。
这其实就是多线程中经典的 Future 模式，你可以将其看作是一种设计模式，核心思想是异步调用，主要用在多线程领域，并非 Java 语言独有。
在 Java 中，Future 类只是一个泛型接口，位于 java.util.concurrent 包下，其中定义了 5 个方法，主要包括下面这 4 个功能：
- 取消任务；
- 判断任务是否被取消;
- 判断任务是否已经执行完成;
- 获取任务执行结果。
// V 代表了Future执行的任务返回值的类型
public interface Future<V> {
// 取消任务执行
// 成功取消返回 true，否则返回 false
boolean cancel(boolean mayInterruptIfRunning);
// 判断任务是否被取消
boolean isCancelled();
// 判断任务是否已经执行完成
boolean isDone();
// 获取任务执行结果
V get() throws InterruptedException, ExecutionException;
// 指定时间内没有返回计算结果就抛出 TimeOutException 异常
V get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutExceptio
}简单理解就是：我有一个任务，提交给了 Future 来处理。任务执行期间我自己可以去做任何想做的事情。并且，在这期间我还可以取消任务以及获取任务的执行状态。一段时间之后，我就可以 Future 那里直接取出任务执行结果。
三、Future的相关类图
2.1 Future 接口
首先，我们需要清楚，Futrue是个接口。Future就是对于具体的Runnable或者Callable任务的执行结果进行取消、查询是否完成、获取结果。必要时可以通过get方法获取执行结果，该方法会阻塞直到任务返回结果。
接口定义行为，我们通过上图可以看到实现Future接口的子类会具有哪些行为：
- 我们可以取消这个执行逻辑，如果这个逻辑已经正在执行，提供可选的参数来控制是否取消已经正在执行的逻辑。
- 我们可以判断执行逻辑是否已经被取消。
- 我们可以判断执行逻辑是否已经执行完成。
- 我们可以获取执行逻辑的执行结果。
- 我们可以允许在一定时间内去等待获取执行结果，如果超过这个时间，抛TimeoutException。
2.2 FutureTask 类
类图如下：
FutureTask是Future的具体实现。FutureTask实现了RunnableFuture接口。RunnableFuture接口又同时继承了Future 和 Runnable 接口。所以FutureTask既可以作为Runnable被线程执行，又可以作为Future得到Callable的返回值。
五、FutureTask源码分析
5.1 state字段
volatile修饰的state字段；表示FutureTask当前所处的状态。可能的转换过程见注释。
/**
* Possible state transitions:
* NEW -> COMPLETING -> NORMAL（创建到正常运行结束的状态变化轨迹）
* NEW -> COMPLETING -> EXCEPTIONAL（创建到异常运行结束的状态变化轨迹）
* NEW -> CANCELLED （创建到取消的状态变化轨迹）
* NEW -> INTERRUPTING -> INTERRUPTED（创建到中断结束的状态变化轨迹）
*/
private volatile int state;
// NEW 新建状态，表示这个 FutureTask还没有开始运行
private static final int NEW = 0;
// COMPLETING 完成状态， 表示 FutureTask 任务已经计算完毕了
// 但是还有一些后续操作，例如唤醒等待线程操作，还没有完成。
private static final int COMPLETING = 1;
// FutureTask 任务完结，正常完成，没有发生异常
private static final int NORMAL = 2;
// FutureTask 任务完结，因为发生异常。
private static final int EXCEPTIONAL = 3;
// FutureTask 任务完结，因为取消任务
private static final int CANCELLED = 4;
// FutureTask 任务完结，也是取消任务，不过发起了中断运行任务线程的中断请求
private static final int INTERRUPTING = 5;
// FutureTask 任务完结，也是取消任务，已经完成了中断运行任务线程的中断请求
private static final int INTERRUPTED = 6;
5.2 其他变量
/** 任务 */
private Callable<V> callable;
/** 储存结果*/
private Object outcome; // non-volatile, protected by state reads/writes
/** 执行任务的线程*/
private volatile Thread runner;
/** get方法阻塞的线程队列 */
private volatile WaitNode waiters;
//FutureTask的内部类，get方法的等待队列
static final class WaitNode {
volatile Thread thread;
volatile WaitNode next;
WaitNode() { thread = Thread.currentThread(); }
}
5.3 CAS工具初始化
// Unsafe mechanics
private static final sun.misc.Unsafe UNSAFE;
private static final long stateOffset;
private static final long runnerOffset;
private static final long waitersOffset;
static {
try {
UNSAFE = sun.misc.Unsafe.getUnsafe();
Class<?> k = FutureTask.class;
stateOffset = UNSAFE.objectFieldOffset
(k.getDeclaredField("state"));
runnerOffset = UNSAFE.objectFieldOffset
(k.getDeclaredField("runner"));
waitersOffset = UNSAFE.objectFieldOffset
(k.getDeclaredField("waiters"));
} catch (Exception e) {
throw new Error(e);
}
}
这段代码是为了后面使用CAS而准备的。可以这么理解：
一个java对象可以看成是一段内存，各个字段都得按照一定的顺序放在这段内存里，同时考虑到对齐要求，可能这些字段不是连续放置的，用这个UNSAFE.objectFieldOffset（）方法能准确地告诉你某个字段相对于对象的起始内存地址的字节偏移量，因为是相对偏移量，所以它其实跟某个具体对象又没什么太大关系，跟class的定义和虚拟机的内存模型的实现细节更相关。
5.4 构造函数
FutureTask有两个构造函数：
public FutureTask(Callable<V> callable) {
if (callable == null)
throw new NullPointerException();
this.callable = callable;
this.state = NEW; // ensure visibility of callable
}
public FutureTask(Runnable runnable, V result) {
this.callable = Executors.callable(runnable, result);
this.state = NEW; // ensure visibility of callable
}这两个构造函数区别在于，如果使用第一个构造函数最后获取线程实行结果就是callable的执行的返回结果；而如果使用第二个构造函数那么最后获取线程实行结果就是参数中的result，接下来让我们看一下FutureTask的run方法。
同时两个构造函数都把当前状态设置为NEW。
5.5 jdk1.8和之前版本的区别
run方法：
public void run() {
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return;
try {
Callable<V> c = callable; // 这里的callable是从构造方法里面传人的
if (c != null && state == NEW) {
V result;
boolean ran;
try {
result = c.call();
ran = true;
} catch (Throwable ex) {
result = null;
ran = false;
setException(ex); // 保存call方法抛出的异常
}
if (ran)
set(result); // 保存call方法的执行结果
}
} finally {
// runner must be non-null until state is settled to
// prevent concurrent calls to run()
runner = null;
// state must be re-read after nulling runner to prevent
// leaked interrupts
int s = state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}
1.8:
get方法的逻辑很简单，如果call方法的执行过程已完成，就把结果给出去；如果未完成，就将当前线程挂起等待。awaitDone方法里面死循环的逻辑，推演几遍就能弄懂；它里面挂起线程的主要创新是定义了WaitNode类，来将多个等待线程组织成队列，这是与JDK6的实现最大的不同。
1.6：
JDK6的FutureTask的基本操作都是通过自己的内部类Sync来实现的，而Sync继承自AbstractQueuedSynchronizer这个出镜率极高的并发工具类
也就是说1.8自己定义了一个WaitNode类，来将多个等待线程组织成队列，在以前使用AQS来实现的
这个get方法里面处理等待线程队列的方式是调用了acquireSharedInterruptibly方法，看过我之前几篇博客文章的读者应该非常熟悉了。其中的等待线程队列、线程挂起和唤醒等逻辑，这里不再赘述，如果不明白，请出门左转。
六、Callable 和 Future 有什么关系？
我们可以通过 FutureTask 来理解 Callable 和 Future 之间的关系。
FutureTask 提供了 Future 接口的基本实现，常用来封装 Callable 和 Runnable，具有取消任务、查看任务是否执行完成以及获取任务执行结果的方法。ExecutorService.submit() 方法返回的其实就是 Future 的实现类 FutureTask 。
<T> Future<T> submit(Callable<T> task);
Future<?> submit(Runnable task);FutureTask 不光实现了 Future接口，还实现了Runnable 接口，因此可以作为任务直接被线程执行。
FutureTask 有两个构造函数，可传入 Callable 或者 Runnable 对象。实际上，传入 Runnable 对象也会在方法内部转换为Callable 对象。
public FutureTask(Callable<V> callable) {
if (callable == null)
throw new NullPointerException();
this.callable = callable;
this.state = NEW;
}
public FutureTask(Runnable runnable, V result) {
// 通过适配器RunnableAdapter来将Runnable对象runnable转换成Callable对象
this.callable = Executors.callable(runnable, result);
this.state = NEW;
}
FutureTask相当于对Callable 进行了封装，管理着任务执行的情况，存储了 Callable 的 call 方法的任务执行结果。
七、CompletableFuture 类
Future 在实际使用过程中存在一些局限性，比如不支持异步任务的编排组合、获取计算结果的 get() 方法为阻塞调用。
Java 8 才被引入CompletableFuture 类可以解决Future 的这些缺陷。CompletableFuture 除了提供了更为好用和强大的 Future 特性之外，还提供了函数式编程、异步任务编排组合（可以将多个异步任务串联起来，组成一个完整的链式调用）等能力。
下面我们来简单看看 CompletableFuture 类的定义。
public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
}
可以看到，CompletableFuture 同时实现了 Future 和 CompletionStage 接口。
CompletionStage 接口描述了一个异步计算的阶段。很多计算可以分成多个阶段或步骤，此时可以通过它将所有步骤组合起来，形成异步计算的流水线。
CompletionStage 接口中的方法比较多，CompletableFuture 的函数式能力就是这个接口赋予的。从这个接口的方法参数你就可以发现其大量使用了 Java8 引入的函数式编程。
本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若内容造成侵权/违法违规/事实不符，请联系邮箱：jacktools123@163.com进行投诉反馈，一经查实，立即删除！
标签：
相关文章
最新发布
点击排行
本站推荐
