Java「并发编程」Future模式（内有CompletableFuture的详解）

Future

Future是Java5增加的类，它用来描述一个异步计算的结果。你可以使用 isDone 方法检查计算是否完成，或者使用 get 方法阻塞住调用线程，直到计算完成返回结果。你也可以使用 cancel 方法停止任务的执行。下面来一个栗子：

public class FutureDemo {
public static void main(String[] args) {
ExecutorService es = Executors.newFixedThreadPool(10);
Future<Integer> f = es.submit(() ->{
Thread.sleep(10000);
// 结果
return 100;
});
// do something
Integer result = f.get();
System.out.println(result);
// while (f.isDone()) {
// System.out.println(result);
// }
}
}
在这个例子中，我们往线程池中提交了一个任务并立即返回了一个Future对象，接着可以做一些其他操作，最后利用它的 get 方法阻塞等待结果或 isDone 方法轮询等待结果（关于Future的原理可以参考之前的文章：【并发编程】Future模式及JDK中的实现）

虽然这些方法提供了异步执行任务的能力，但是对于结果的获取却还是很不方便，只能通过阻塞或者轮询的方式得到任务的结果。

阻塞的方式显然和我们的异步编程的初衷相违背，轮询的方式又会耗费无谓的CPU资源，而且也不能及时的得到计算结果，为什么不能用观察者设计模式当计算结果完成及时通知监听者呢？

很多语言，比如Node.js，采用Callback的方式实现异步编程。Java的一些框架，比如Netty，自己扩展了Java的 Future 接口，提供了 addListener 等多个扩展方法。Google的guava也提供了通用的扩展Future：ListenableFuture 、 SettableFuture 以及辅助类 Futures 等，方便异步编程。为此，Java终于在JDK1.8这个版本中增加了一个能力更强的Future类：CompletableFuture 。它提供了非常强大的Future的扩展功能，可以帮助我们简化异步编程的复杂性，提供了函数式编程的能力，可以通过回调的方式处理计算结果。下面来看看这几种方式。

Netty-Future

引入Maven依赖：

<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.29.Final</version>
</dependency>

public class NettyFutureDemo {

public static void main(String[] args) throws InterruptedException {

EventExecutorGroup group = new DefaultEventExecutorGroup(4);
System.out.println("开始:" + DateUtils.getNow());
Future<Integer> f = group.submit(new Callable<Integer>() {
@Override
public Integer call() throws Exception {
System.out.println("开始耗时计算:" + DateUtils.getNow());
Thread.sleep(10000);
System.out.println("结束耗时计算:" + DateUtils.getNow());
return 100;
}
});
f.addListener(new FutureListener<Object>() {
@Override
public void operationComplete(Future<Object> objectFuture) throws Exception {
System.out.println("计算结果:" + objectFuture.get());
}
});
System.out.println("结束:" + DateUtils.getNow());
// 不让守护线程退出
new CountDownLatch(1).await();
}
}
输出结果：

开始:2019-05-16 08:25:40:779
结束:2019-05-16 08:25:40:788
开始耗时计算:2019-05-16 08:25:40:788
结束耗时计算:2019-05-16 08:25:50:789
计算结果:100
从结果可以看出，耗时计算结束后自动触发Listener的完成方法，避免了主线程无谓的阻塞等待，那么它究竟是怎么做到的呢？下面看源码

DefaultEventExecutorGroup 实现了 EventExecutorGroup 接口，而 EventExecutorGroup 则是实现了JDK ScheduledExecutorService 接口的线程组接口，所以它拥有线程池的所有方法。然而它却把所有返回 java.util.concurrent.Future 的方法重写为返回 io.netty.util.concurrent.Future ，把所有返回 java.util.concurrent.ScheduledFuture 的方法重写为返回 io.netty.util.concurrent.ScheduledFuture 。

public interface EventExecutorGroup extends ScheduledExecutorService, Iterable<EventExecutor> {
/**

  • 返回一个EventExecutor
    */
    EventExecutor next();
    Iterator<EventExecutor> iterator();
    Future<?> submit(Runnable task);
    <T> Future<T> submit(Runnable task, T result);
    <T> Future<T> submit(Callable<T> task);
    ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit);
    <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit);
    ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit);
    ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit);
    }
    EventExecutorGroup 的submit方法因为 newTaskFor 的重写导致返回了netty的 Future 实现类，而这个实现类正是 PromiseTask 。

@Override
public <T> Future<T> submit(Callable<T> task) {
return (Future<T>) super.submit(task);
}
@Override
protected final <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
return new PromiseTask<T>(this, callable);
}
PromiseTask 的实现很简单，它缓存了要执行的 Callable 任务，并在run方法中完成了任务调用和Listener的通知。

@Override
public void run() {
try {
if (setUncancellableInternal()) {
V result = task.call();
setSuccessInternal(result);
}
} catch (Throwable e) {
setFailureInternal(e);
}
}
@Override
public Promise<V> setSuccess(V result) {
if (setSuccess0(result)) {
notifyListeners();
return this;
}
throw new IllegalStateException("complete already: " + this);
}
@Override
public Promise<V> setFailure(Throwable cause) {
if (setFailure0(cause)) {
notifyListeners();
return this;
}
throw new IllegalStateException("complete already: " + this, cause);
}
任务调用成功或者失败都会调用 notifyListeners 来通知Listener，所以大家得在回调的函数里调用 isSuccess 方法来检查状态。

这里有一个疑惑，会不会 Future 在调用 addListener 方法的时候任务已经执行完成了，这样子会不会通知就会失败了啊？

@Override
public Promise<V> addListener(GenericFutureListener<? extends Future<? super V>> listener) {
synchronized (this) {
addListener0(listener);
}
if (isDone()) {
notifyListeners();
}
return this;
}
可以发现，在Listener添加成功之后，会立即检查状态，如果任务已经完成立刻进行回调，所以这里不用担心啦。OK，下面看看Guava-Future的实现。

Guava-Future

首先引入guava的Maven依赖：

<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
<version>22.0</version>
</dependency>

public class GuavaFutureDemo {

public static void main(String[] args) throws InterruptedException {

System.out.println("开始:" + DateUtils.getNow());

ExecutorService executorService = Executors.newFixedThreadPool(10);
ListeningExecutorService service = MoreExecutors.listeningDecorator(executorService);
ListenableFuture<Integer> future = service.submit(new Callable<Integer>() {
@Override
public Integer call() throws Exception {
System.out.println("开始耗时计算:" + DateUtils.getNow());
Thread.sleep(10000);
System.out.println("结束耗时计算:" + DateUtils.getNow());
return 100;
}
});

future.addListener(new Runnable() {
@Override
public void run() {
System.out.println("调用成功");
}
}, executorService);
System.out.println("结束:" + DateUtils.getNow());
new CountDownLatch(1).await();
}
}
ListenableFuture 可以通过 addListener 方法增加回调函数，一般用于不在乎执行结果的地方。如果需要在执行成功时获取结果或者执行失败时获取异常信息，需要用到 Futures 工具类的 addCallback 方法：

Futures.addCallback(future, new FutureCallback<Integer>() {
@Override
public void onSuccess(@Nullable Integer result) {
System.out.println("成功，计算结果:" + result);
}
@Override
public void onFailure(Throwable t) {
System.out.println("失败");
}
}, executorService);
前面提到除了 ListenableFuture 外，还有一个 SettableFuture 类也支持回调能力。它实现自 ListenableFuture ，所以拥有 ListenableFuture 的所有能力。

public class GuavaFutureDemo {

public static void main(String[] args) throws InterruptedException {

System.out.println("开始:" + DateUtils.getNow());
ExecutorService executorService = Executors.newFixedThreadPool(10);
ListenableFuture<Integer> future = submit(executorService);
Futures.addCallback(future, new FutureCallback<Integer>() {
@Override
public void onSuccess(@Nullable Integer result) {
System.out.println("成功，计算结果:" + result);
}
@Override
public void onFailure(Throwable t) {
System.out.println("失败:" + t.getMessage());
}
}, executorService);
Thread.sleep(1000);
System.out.println("结束:" + DateUtils.getNow());
new CountDownLatch(1).await();
}
private static ListenableFuture<Integer> submit(Executor executor) {
SettableFuture<Integer> future = SettableFuture.create();
executor.execute(new Runnable() {
@Override
public void run() {
System.out.println("开始耗时计算:" + DateUtils.getNow());
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("结束耗时计算:" + DateUtils.getNow());
// 返回值
future.set(100);
// 设置异常信息
// future.setException(new RuntimeException("custom error!"));
}
});
return future;
}
}
看起来用法上没有太多差别，但是有一个很容易被忽略的重要问题。当 SettableFuture 的这种方式最后调用了 cancel 方法后，线程池中的任务还是会继续执行，而通过 submit 方法返回的 ListenableFuture 方法则会立即取消执行，这点尤其要注意。下面看看源码：

和Netty的Future一样，Guava也是通过实现了自定义的 ExecutorService 实现类 ListeningExecutorService 来重写了 submit 方法。

public interface ListeningExecutorService extends ExecutorService {
<T> ListenableFuture<T> submit(Callable<T> task);
ListenableFuture<?> submit(Runnable task);
<T> ListenableFuture<T> submit(Runnable task, T result);
}
同样的，newTaskFor 方法也被进行了重写，返回了自定义的Future类：TrustedListenableFutureTask

@Override
protected final <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
return TrustedListenableFutureTask.create(runnable, value);
}
@Override
protected final <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
return TrustedListenableFutureTask.create(callable);
}
任务调用会走 TrustedFutureInterruptibleTask 的run方法：

@Override
public void run() {
TrustedFutureInterruptibleTask localTask = task;
if (localTask != null) {
localTask.run();
}
}
@Override
public final void run() {
if (!ATOMIC_HELPER.compareAndSetRunner(this, null, Thread.currentThread())) {
return; // someone else has run or is running.
}
try {
// 抽象方法，子类进行重写
runInterruptibly();
} finally {
if (wasInterrupted()) {
while (!doneInterrupting) {
Thread.yield();
}
}
}
}
最终还是调用到 TrustedFutureInterruptibleTask 的 runInterruptibly 方法，等待任务完成后调用 set 方法。

@Override
void runInterruptibly() {
if (!isDone()) {
try {
set(callable.call());
} catch (Throwable t) {
setException(t);
}
}
}
protected boolean set(@Nullable V value) {
Object valueToSet = value == null ? NULL : value;
// CAS设置值
if (ATOMIC_HELPER.casValue(this, null, valueToSet)) {
complete(this);
return true;
}
return false;
}
在 complete 方法的最后会获取到Listener进行回调。

上面提到的 SettableFuture 和 ListenableFuture 的 cancel 方法效果不同，原因在于一个重写了 afterDone 方法而一个没有。

下面是 ListenableFuture 的 afterDone 方法：

@Override
protected void afterDone() {
super.afterDone();
if (wasInterrupted()) {
TrustedFutureInterruptibleTask localTask = task;
if (localTask != null) {
localTask.interruptTask();
}
}
this.task = null;
}
wasInterrupted 用来判断是否调用了 cancel （cancel方法会设置一个取消对象Cancellation到value中）

protected final boolean wasInterrupted() {
final Object localValue = value;
return (localValue instanceof Cancellation) && ((Cancellation) localValue).wasInterrupted;
}
interruptTask 方法通过线程的 interrupt 方法真正取消线程任务的执行：

final void interruptTask() {
Thread currentRunner = runner;
if (currentRunner != null) {
currentRunner.interrupt();
}
doneInterrupting = true;
}
由 Callback Hell 引出 Promise 模式

如果你对 ES6 有所接触，就不会对 Promise 这个模式感到陌生，如果你对前端不熟悉，也不要紧，我们先来看看回调地狱（Callback Hell）是个什么概念。

回调是一种我们推崇的异步调用方式，但也会遇到问题，也就是回调的嵌套。当需要多个异步回调一起书写时，就会出现下面的代码(以 js 为例):

asyncFunc1(opt, (...args1) => {
asyncFunc2(opt, (...args2) => {
asyncFunc3(opt, (...args3) => {
asyncFunc4(opt, (...args4) => {
// some operation
});
});
});
});
虽然在 JAVA 业务代码中很少出现回调的多层嵌套，但总归是个问题，这样的代码不易读，嵌套太深修改也麻烦。于是 ES6 提出了 Promise 模式来解决回调地狱的问题。可能就会有人想问：java 中存在 Promise 模式吗？答案是肯定的。

前面提到了 Netty 和 Guava 的扩展都提供了 addListener 这样的接口，用于处理 Callback 调用，但其实 jdk1.8 已经提供了一种更为高级的回调方式：CompletableFuture。首先尝试用 CompletableFuture 来重写上面回调的问题。

public class CompletableFutureTest {
public static void main(String[] args) throws InterruptedException {
System.out.println("开始:" + DateUtils.getNow());
CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
System.out.println("开始耗时计算:" + DateUtils.getNow());
try {
Thread.sleep(10000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("结束耗时计算:" + DateUtils.getNow());
return 100;
});
completableFuture.whenComplete((result, e) -> {
System.out.println("回调结果:" + result);
});
System.out.println("结束:" + DateUtils.getNow());
new CountDownLatch(1).await();
}
}
使用CompletableFuture耗时操作没有占用主线程的时间片，达到了异步调用的效果。我们也不需要引入任何第三方的依赖，这都是依赖于 java.util.concurrent.CompletableFuture 的出现。CompletableFuture 提供了近 50 多个方法，大大便捷了 java 多线程操作，和异步调用的写法。

使用 CompletableFuture 解决回调地狱问题：

public class CompletableFutureDemo {
public static void main(String[] args) throws InterruptedException {
long l = System.currentTimeMillis();
CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
System.out.println("在回调中执行耗时操作...");
Thread.sleep(10000);
return 100;
});
completableFuture = completableFuture.thenCompose(i -> {
return CompletableFuture.supplyAsync(() -> {
System.out.println("在回调的回调中执行耗时操作...");
Thread.sleep(10000);
return i + 100;
});
});
completableFuture.whenComplete((result, e) -> {
System.out.println("计算结果:" + result);
});
System.out.println("主线程运算耗时:" + (System.currentTimeMillis() - l) + " ms");
new CountDownLatch(1).await();
}
}
输出：

在回调中执行耗时操作...主线程运算耗时:58 ms在回调的回调中执行耗时操作...计算结果:200
使用 thenCompose 或者 thenComposeAsync 等方法可以实现回调的回调，且写出来的方法易于维护。

总的看来，为Future模式增加回调功能就不需要阻塞等待结果的返回并且不需要消耗无谓的CPU资源去轮询处理状态，JDK8之前使用Netty或者Guava提供的工具类，JDK8之后则可以使用自带的 CompletableFuture 类。Future 有两种模式：将来式和回调式。而回调式会出现回调地狱的问题，由此衍生出了 Promise 模式来解决这个问题。这才是 Future 模式和 Promise 模式的相关性。

©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 212,294评论 6赞 493
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 90,493评论 3赞 385
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 157,790评论 0赞 348
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 56,595评论 1赞 284
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 65,718评论 6赞 386
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 49,906评论 1赞 290
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 39,053评论 3赞 410
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 37,797评论 0赞 268
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 44,250评论 1赞 303
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 36,570评论 2赞 327
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 38,711评论 1赞 341
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 34,388评论 4赞 332
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 40,018评论 3赞 316
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,796评论 0赞 21
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 32,023评论 1赞 266
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 46,461评论 2赞 360
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 43,595评论 2赞 350

推荐阅读更多精彩内容