Java异步编程 2 CompletableFuture

所谓异步调用其实就是实现一个可无需等待被调用函数的返回值而让操作继续运行的方法。

在 Java 语言中，简单的讲就是另启一个线程来完成调用中的部分计算，使调用继续运行或返回，而不需要等待计算结果。但调用者仍需要取线程的计算结果。

JDK5 新增了 Future 接口，用于描述一个异步计算的结果。虽然 Future 以及相关使用方法提供了异步执行任务的能力，但是对于结果的获取却是很不方便，只能通过阻塞或者轮询的方式得到任务的结果。 例如：

public static void main(String[] args) throws Exception {
    ExecutorService es = Executors.newSingleThreadExecutor();

    // 在 Java8 中，推荐使用 Lambda 来替代匿名 Callable 实现类
    Future<Integer> f = es.submit(() -> {
        System.out.println(Thread.currentThread().getName());

        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        return 123;
    });

    // 当前 main 线程阻塞，直至 future 得到值
    System.out.println(f.get());

    es.shutdown();
}

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

关于 Future 接口，还有如下一段描述：

The Future interface was added in Java 5 to serve as a result of an asynchronous computation, but it did not have any methods to combine these computations or handle possible errors.
不能很好地组合多个异步任务，也不能处理可能的异常。

CompletableFuture

Java 8 中, 新增加了一个包含 50 个方法左右的类 CompletableFuture，它提供了非常强大的 Future 的扩展功能，可以帮助我们简化异步编程的复杂性，并且提供了函数式编程的能力，可以通过回调的方式处理计算结果，也提供了转换和组合 CompletableFuture 的方法。

对于阻塞或者轮询方式，依然可以通过 CompletableFuture 类的 CompletionStage 和 Future 接口方式支持。

CompletableFuture 类声明了 CompletionStage 接口，CompletionStage 接口实际上提供了同步或异步运行计算的舞台，所以我们可以通过实现多个 CompletionStage 命令，并且将这些命令串联在一起的方式实现多个命令之间的触发。

我们可以通过 CompletableFuture.supplyAsync(this::sendMsg); 这么一行代码创建一个简单的异步计算。在这行代码中，supplyAsync 支持异步地执行我们指定的方法，这个例子中的异步执行方法是 sendMsg。当然，我们也可以使用 Executor 执行异步程序，默认是 ForkJoinPool.commonPool()。

我们也可以在异步计算结束之后指定回调函数，例如 CompletableFuture.supplyAsync(this::sendMsg) .thenAccept(this::notify); 这行代码中的 thenAccept 被用于增加回调函数，在我们的示例中 notify 就成了异步计算的消费者，它会处理计算结果。

CompletionStage<T> 接口

A stage of a possibly asynchronous computation, that performs an action or computes a value when another CompletionStage completes.
A stage completes upon termination of its computation, but this may in turn trigger other dependent stages.
一个可能执行的异步计算的某个阶段，在另一个CompletionStage完成时执行一个操作或计算一个值。
一个阶段完成后，其计算结束。但是，该计算阶段可能会触发下一个计算阶段。

最简单的例子

CompletableFuture 实际上也实现了 Future 接口：

public class CompletableFuture<T> implements Future<T>, CompletionStage<T>

所以我们也可以利用 CompletableFuture 来实现基本的 Future 功能，例如：

public static void main(String[] args) throws Exception {
    CompletableFuture future = new CompletableFuture();

    // 在 Java8 中，推荐使用 Lambda 来替代匿名 Runnable 实现类
    new Thread(
            () -> {
                try {
                    // 模拟一段耗时的操作
                    Thread.sleep(2000);

                    future.complete("I have completed");
                } catch (Exception e) {
                }
            }
    ).start();

    System.out.println(future.get());
}

此时此刻主线程 future.get() 将得到字符串的结果 I have completed，同时完成回调以后将会立即生效。注意 complete() 方法只能调用一次，后续调用将被忽略。

注意：get() 方法可能会抛出异常 InterruptedException 和 ExecutionException。

如果我们已经知道了异步任务的结果，我们也可以直接创建一个已完成的 future，如下：

public static void main(String[] args) throws Exception {
    // Returns a new CompletableFuture that is already completed with the given value.
    CompletableFuture future = CompletableFuture.completedFuture("I have completed");

    System.out.println(future.get());
}

如果在异步执行过程中，我们觉得执行会超时或者会出现问题，我们也可以通过 cancle() 方法取消，此时调用 get() 方法时会产生异常 java.util.concurrent.CancellationException，代码如下：

public static void main(String[] args) throws Exception {
    CompletableFuture future = new CompletableFuture();

    // 在 Java8 中，推荐使用 Lambda 来替代匿名 Runnable 实现类
    new Thread(
            () -> {
                try {
                    // 模拟一段耗时的操作
                    Thread.sleep(2000);

                    future.cancel(false);
                } catch (Exception e) {
                }
            }
    ).start();

    System.out.println(future.get());
}

使用工厂方法创建 CompletableFuture

在上述的代码中，我们手动地创建 CompletableFuture，并且手动的创建一个线程（或者利用线程池）来启动异步任务，这样似乎有些复杂。

其实我们可以利用 CompletableFuture 的工厂方法，传入 Supplier 或者 Runnable 的实现类，直接得到一个 CompletableFuture 的实例：

  • public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
  • public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)
  • public static CompletableFuture<Void> runAsync(Runnable runnable)
  • public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)

第一个和第三个方法，没有 Executor 参数，将会使用 ForkJoinPool.commonPool() (全局的，在 JDK8 中介绍的通用池），这适用于 CompletableFuture 类中的大多数的方法。

  • Runnable 接口方法 public abstract void run(); 没有返回值
  • Supplier 接口方法 T get(); 有返回值。如果你需要处理异步操作并返回结果，使用前两种 Supplier<U> 方法

一个小的 Tips：

Both Runnable and Supplier are functional interfaces that allow passing their instances as lambda expressions thanks to the new Java 8 feature. 使用 Lambda 表达式来传入 Supplier 或者 Runnable 的实现类。

一个示例代码如下：

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

    // 在 Java8 中，推荐使用 Lambda 来替代匿名 Supplier 实现类
    CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
        try {
            Thread.sleep(2000);
        } catch (Exception e) {
        }

        return "I have completed";
    });

    System.out.println(future.get());
}

转换和作用于异步任务的结果 (thenApply)

我们可以叠加功能，把多个 future 组合在一起等

  • public <U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn)
    • 该方法的作用是在该计算阶段正常完成后，将该计算阶段的结果作为参数传递给参数 fn 值的函数Function，并会返回一个新的 CompletionStage
  • public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn)
    • 该方法和上面的方法 thenApply 功能类似，不同的是对该计算阶段的结果进行计算的函数 fn 的执行时异步的。
  • public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn, Executor executor)
    • 该方法和上面的方法 thenApplyAsync 功能类似，不同的是对该计算阶段的结果进行计算的函数 fn 的执行时异步的， 并且是在调用者提供的线程池中执行的。

Function 接口方法 R apply(T t); 包含一个参数和一个返回值

一个示例代码如下：

public static void main(String[] args) throws Exception {
    // 在 Java8 中，推荐使用 Lambda 来替代匿名 Supplier 实现类
    CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
        try {
            Thread.sleep(2000);
        } catch (Exception e) {
        }

        return "I have completed";
    });

    CompletableFuture<String> upperfuture = future.thenApply(String::toUpperCase);

    System.out.println(upperfuture.get());
}

运行完成的异步任务的结果 (thenAccept/thenRun)

在 future 的管道里有两种典型的“最终”阶段方法。他们在你使用 future 的值的时候做好准备，当
thenAccept() 提供最终的值时，thenRun 执行 Runnable。

  • public CompletableFuture<Void> thenAccept(Consumer<? super T> action)
  • public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action)
  • public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action, Executor executor)
  • public CompletableFuture<Void> thenRun(Runnable action)
  • public CompletableFuture<Void> thenRunAsync(Runnable action)
  • public CompletableFuture<Void> thenRunAsync(Runnable action, Executor executor)

Consumer 接口方法 void accept(T t); 包含一个参数，但是没有返回值

一个示例代码如下：

public static void main(String[] args) throws Exception {
    // 在 Java8 中，推荐使用 Lambda 来替代匿名 Supplier 实现类
    CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
        try {
            Thread.sleep(2000);
        } catch (Exception e) {
        }

        return "I have completed";
    });

    future.thenAccept(s -> {
        System.out.println(s);
    });

    // Waits if necessary for this future to complete, and then returns its result.
    future.get();
}

结合两个 CompletableFuture

The best part of the CompletableFuture API is the ability to combine CompletableFuture instances in a chain of computation steps.
这就是 CompletableFuture 最大的优势。

  • public <U> CompletableFuture<U> thenCompose(Function<? super T, ? extends CompletionStage<U>> fn)
    • 传入前一个 CompletableFuture 的返回值，返回另外一个 CompletableFuture 实例
  • public <U> CompletableFuture<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn)
  • public <U> CompletableFuture<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn, Executor executor)
  • public <U,V> CompletableFuture<V> thenCombine(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn)
  • public <U,V> CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn)
  • public <U,V> CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn, Executor executor)

一个示例代码如下：

public static void main(String[] args) throws Exception {
    CompletableFuture<String> future
            = CompletableFuture.supplyAsync(() -> {
        try {
            Thread.sleep(1000);
        } catch (Exception e) {
        }

        return "Hello ";
    }).thenCompose(s -> CompletableFuture.supplyAsync(() -> {
        try {
            Thread.sleep(2000);
        } catch (Exception e) {
        }

        return s + "World";
    }));

    System.out.println(future.get()); // Hello World
}

上述功能也可以通过 thenCombine() 方法实现，传入一个 BiFunction 接口的实例（以 Lambda 形式） 例如：

public static void main(String[] args) throws Exception {
    CompletableFuture<String> future
            = CompletableFuture.supplyAsync(() -> {
        try {
            Thread.sleep(1000);
        } catch (Exception e) {
        }

        return "Hello ";
    }).thenCombine(CompletableFuture.supplyAsync(() -> {
        try {
            Thread.sleep(2000);
        } catch (Exception e) {
        }

        return "World";
    }), (s1, s2) -> s1 + s2);

    System.out.println(future.get());
}

并行执行多个异步任务

有时候我们可能需要等待所有的异步任务都执行完毕，然后组合他们的结果。我们可以使用 allOf() 方法：

  • public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs)

一个示例代码如下：

public static void main(String[] args) throws Exception {
    CompletableFuture<String> future1
            = CompletableFuture.supplyAsync(() -> "Hello");
    CompletableFuture<String> future2
            = CompletableFuture.supplyAsync(() -> "World");

    CompletableFuture<Void> combinedFuture
            = CompletableFuture.allOf(future1, future2);

    // 这个方法不会合并结果，可以看到他的返回值是 Void 类型
    combinedFuture.get();

    // 我们需要手动来处理每一个并行异步任务的结果
    String combined = Stream.of(future1, future2)
            .map(CompletableFuture::join)
            .collect(Collectors.joining(" "));

    System.out.println(combined); // Hello World
}

有时候我们可能不需要等待所有的异步任务都执行完毕，只要任何一个任务完成就返回结果。我们可以使用 anyOf() 方法：

  • public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs)

一个示例代码如下：

public static void main(String[] args) throws Exception {
    CompletableFuture<String> future1
            = CompletableFuture.supplyAsync(() -> "Hello");
    CompletableFuture<String> future2
            = CompletableFuture.supplyAsync(() -> {
                try {
                    Thread.sleep(2000);
                } catch (Exception e) {
                }

                return "World";
            }
    );

    CompletableFuture<Object> combinedFuture
            = CompletableFuture.anyOf(future1, future2);

    System.out.println(combinedFuture.get()); // Hello
}

异常的处理

我们可以在 handle() 方法里处理异常：

  • public <U> CompletableFuture<U> handle(BiFunction<? super T, Throwable, ? extends U> fn)
    • 第一个参数为 CompletableFuture 返回的结果
    • 第二个参数为抛出的异常

一个示例代码如下：

public static void main(String[] args) throws Exception {
    String name = null;

    CompletableFuture<String> future
            = CompletableFuture.supplyAsync(() -> {
        if (name == null) {
            throw new RuntimeException("Computation error!");
        }

        return "Hello, " + name;
    }).handle((s, t) -> s != null ? s : "Hello, Stranger!");

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