乐趣区

关于java:Java-8-中的-CompletableFuture-太好用了20-个示例全分享…

jiezi

在 Java 中异步编程，不肯定非要应用 rxJava, Java 自身的库中的 CompletableFuture 能够很好的应答大部分的场景。

这篇文章介绍 Java 8 的 CompletionStage API 和它的规范库的实现 CompletableFuture。API 通过例子的形式演示了它的行为，每个例子演示一到两个行为。

既然 CompletableFuture 类实现了 CompletionStage 接口，首先咱们须要了解这个接口的契约。它代表了一个特定的计算的阶段，能够同步或者异步的被实现。你能够把它看成一个计算流水线上的一个单元，最终会产生一个最终后果，这意味着几个 CompletionStage 能够串联起来，一个实现的阶段能够触发下一阶段的执行，接着触发下一次，接着……

除了实现 CompletionStage 接口，CompletableFuture也实现了 future 接口, 代表一个未实现的异步事件。CompletableFuture提供了办法，可能显式地实现这个 future, 所以它叫CompletableFuture。

1、创立一个实现的 CompletableFuture

最简略的例子就是应用一个预约义的后果创立一个实现的 CompletableFuture, 通常咱们会在计算的开始阶段应用它。

static void completedFutureExample() {CompletableFuture cf = CompletableFuture.completedFuture("message");
    assertTrue(cf.isDone());
    assertEquals("message", cf.getNow(null));
}

getNow(null)办法在 future 实现的状况下会返回后果，就比方下面这个例子，否则返回 null (传入的参数)。

2、运行一个简略的异步阶段

这个例子创立一个一个异步执行的阶段：

static void runAsyncExample() {CompletableFuture cf = CompletableFuture.runAsync(() -> {assertTrue(Thread.currentThread().isDaemon());
        randomSleep();});
    assertFalse(cf.isDone());
    sleepEnough();
    assertTrue(cf.isDone());
}

通过这个例子能够学到两件事件：

CompletableFuture 的办法如果以 Async 结尾，它会异步的执行(没有指定 executor 的状况下)，异步执行通过 ForkJoinPool 实现，它应用守护线程去执行工作。留神这是 CompletableFuture 的个性，其它 CompletionStage 能够 override 这个默认的行为。

3、在前一个阶段上利用函数

上面这个例子应用后面 #1 的实现的 CompletableFuture，#1 返回后果为字符串message, 而后利用一个函数把它变成大写字母。

static void thenApplyExample() {CompletableFuture cf = CompletableFuture.completedFuture("message").thenApply(s -> {assertFalse(Thread.currentThread().isDaemon());
        return s.toUpperCase();});
    assertEquals("MESSAGE", cf.getNow(null));
}

留神 thenApply 办法名称代表的行为。

then意味着这个阶段的动作产生以后的阶段失常实现之后。本例中，以后节点实现，返回字符串message。

Apply意味着返回的阶段将会对后果前一阶段的后果利用一个函数。

函数的执行会被 阻塞 ，这意味着getNow() 只有打斜操作被实现后才返回。

4、在前一个阶段上异步利用函数

通过调用异步办法(办法后边加 Async 后缀)，串联起来的 CompletableFuture 能够异步地执行（应用 ForkJoinPool.commonPool()）。

static void thenApplyAsyncExample() {CompletableFuture cf = CompletableFuture.completedFuture("message").thenApplyAsync(s -> {assertTrue(Thread.currentThread().isDaemon());
        randomSleep();
        return s.toUpperCase();});
    assertNull(cf.getNow(null));
    assertEquals("MESSAGE", cf.join());
}

5、应用定制的 Executor 在前一个阶段上异步利用函数

异步办法一个十分有用的个性就是可能提供一个 Executor 来异步地执行 CompletableFuture。这个例子演示了如何应用一个固定大小的线程池来利用大写函数。

static ExecutorService executor = Executors.newFixedThreadPool(3, new ThreadFactory() {
    int count = 1;
 
    @Override
    public Thread newThread(Runnable runnable) {return new Thread(runnable, "custom-executor-" + count++);
    }
});
 
static void thenApplyAsyncWithExecutorExample() {CompletableFuture cf = CompletableFuture.completedFuture("message").thenApplyAsync(s -> {assertTrue(Thread.currentThread().getName().startsWith("custom-executor-"));
        assertFalse(Thread.currentThread().isDaemon());
        randomSleep();
        return s.toUpperCase();}, executor);
 
    assertNull(cf.getNow(null));
    assertEquals("MESSAGE", cf.join());
}

6、生产前一阶段的后果

如果下一阶段接管了以后阶段的后果，然而在计算的时候不须要返回值(它的返回类型是 void)，那么它能够不利用一个函数，而是一个消费者，调用办法也变成了thenAccept:

static void thenAcceptExample() {StringBuilder result = new StringBuilder();
    CompletableFuture.completedFuture("thenAccept message")
            .thenAccept(s -> result.append(s));
    assertTrue("Result was empty", result.length() > 0);
}

本例中消费者同步地执行，所以咱们不须要在 CompletableFuture 调用 join 办法。

7、异步地生产迁徙阶段的后果

同样，能够应用 thenAcceptAsync 办法，串联的 CompletableFuture 能够异步地执行。

static void thenAcceptAsyncExample() {StringBuilder result = new StringBuilder();
    CompletableFuture cf = CompletableFuture.completedFuture("thenAcceptAsync message")
            .thenAcceptAsync(s -> result.append(s));
    cf.join();
    assertTrue("Result was empty", result.length() > 0);
}

8、实现计算异样

当初咱们来看一下异步操作如何显式地返回异样，用来批示计算失败。咱们简化这个例子，操作解决一个字符串，把它转换成答谢，咱们模仿提早一秒。

咱们应用 thenApplyAsync(Function, Executor) 办法，第一个参数传入大写函数，executor 是一个 delayed executor, 在执行前会提早一秒。

static void completeExceptionallyExample() {CompletableFuture cf = CompletableFuture.completedFuture("message").thenApplyAsync(String::toUpperCase,
            CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS));
    CompletableFuture exceptionHandler = cf.handle((s, th) -> {return (th != null) ? "message upon cancel" : ""; });
    cf.completeExceptionally(new RuntimeException("completed exceptionally"));
assertTrue("Was not completed exceptionally", cf.isCompletedExceptionally());
    try {cf.join();
        fail("Should have thrown an exception");
    } catch(CompletionException ex) { // just for testing
        assertEquals("completed exceptionally", ex.getCause().getMessage());
    }
 
    assertEquals("message upon cancel", exceptionHandler.join());
}

让咱们看一下细节。

首先咱们创立了一个 CompletableFuture, 实现后返回一个字符串 message, 接着咱们调用thenApplyAsync 办法，它返回一个 CompletableFuture。这个办法在第一个函数实现后，异步地利用转大写字母函数。

这个例子还演示了如何通过 delayedExecutor(timeout, timeUnit) 提早执行一个异步工作。

咱们创立了一个拆散的 handler 阶段：exceptionHandler，它解决异样异样，在异常情况下返回message upon cancel。

下一步咱们显式地用异样实现第二个阶段。在阶段上调用 join 办法，它会执行大写转换，而后抛出 CompletionException（失常的 join 会期待 1 秒，而后失去大写的字符串。不过咱们的例子还没等它执行就实现了异样），而后它触发了 handler 阶段。

9、勾销计算

和实现异样相似，咱们能够调用 cancel(boolean mayInterruptIfRunning) 勾销计算。对于 CompletableFuture 类，布尔参数并没有被应用，这是因为它并没有应用中断去勾销操作，相同，cancel等价于completeExceptionally(new CancellationException())。

static void cancelExample() {CompletableFuture cf = CompletableFuture.completedFuture("message").thenApplyAsync(String::toUpperCase,
            CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS));
    CompletableFuture cf2 = cf.exceptionally(throwable -> "canceled message");
    assertTrue("Was not canceled", cf.cancel(true));
    assertTrue("Was not completed exceptionally", cf.isCompletedExceptionally());
    assertEquals("canceled message", cf2.join());
}

10、在两个实现的阶段其中之一上利用函数

上面的例子创立了 CompletableFuture, applyToEither 解决两个阶段，在其中之一上利用函数(包保障哪一个被执行)。本例中的两个阶段一个是利用大写转换在原始的字符串上，另一个阶段是利用小些转换。

static void applyToEitherExample() {
    String original = "Message";
    CompletableFuture cf1 = CompletableFuture.completedFuture(original)
            .thenApplyAsync(s -> delayedUpperCase(s));
    CompletableFuture cf2 = cf1.applyToEither(CompletableFuture.completedFuture(original).thenApplyAsync(s -> delayedLowerCase(s)),
            s -> s + "from applyToEither");
    assertTrue(cf2.join().endsWith("from applyToEither"));
}

11、在两个实现的阶段其中之一上调用生产函数

和前一个例子很相似了，只不过咱们调用的是消费者函数 (Function 变成 Consumer):

static void acceptEitherExample() {
    String original = "Message";
    StringBuilder result = new StringBuilder();
    CompletableFuture cf = CompletableFuture.completedFuture(original)
            .thenApplyAsync(s -> delayedUpperCase(s))
            .acceptEither(CompletableFuture.completedFuture(original).thenApplyAsync(s -> delayedLowerCase(s)),
                    s -> result.append(s).append("acceptEither"));
    cf.join();
    assertTrue("Result was empty", result.toString().endsWith("acceptEither"));
}

12、在两个阶段都执行完后运行一个 Runnable

这个例子演示了依赖的 CompletableFuture 如果期待两个阶段实现后执行了一个 Runnable。留神上面所有的阶段都是同步执行的，第一个阶段执行大写转换，第二个阶段执行小写转换。

static void runAfterBothExample() {
    String original = "Message";
    StringBuilder result = new StringBuilder();
    CompletableFuture.completedFuture(original).thenApply(String::toUpperCase).runAfterBoth(CompletableFuture.completedFuture(original).thenApply(String::toLowerCase),
            () -> result.append("done"));
    assertTrue("Result was empty", result.length() > 0);
}

13、应用 BiConsumer 解决两个阶段的后果

下面的例子还能够通过 BiConsumer 来实现:

static void thenAcceptBothExample() {
    String original = "Message";
    StringBuilder result = new StringBuilder();
    CompletableFuture.completedFuture(original).thenApply(String::toUpperCase).thenAcceptBoth(CompletableFuture.completedFuture(original).thenApply(String::toLowerCase),
            (s1, s2) -> result.append(s1 + s2));
    assertEquals("MESSAGEmessage", result.toString());
}

14、应用 BiFunction 解决两个阶段的后果

如果 CompletableFuture 依赖两个后面阶段的后果，它复合两个阶段的后果再返回一个后果，咱们就能够应用 thenCombine() 函数。整个流水线是同步的，所以 getNow() 会失去最终的后果，它把大写和小写字符串连接起来。

static void thenCombineExample() {
    String original = "Message";
    CompletableFuture cf = CompletableFuture.completedFuture(original).thenApply(s -> delayedUpperCase(s))
            .thenCombine(CompletableFuture.completedFuture(original).thenApply(s -> delayedLowerCase(s)),
                    (s1, s2) -> s1 + s2);
    assertEquals("MESSAGEmessage", cf.getNow(null));
}

15、异步应用 BiFunction 解决两个阶段的后果

相似下面的例子，然而有一点不同：依赖的前两个阶段异步地执行，所以 thenCombine() 也异步地执行，即时它没有 Async 后缀。

Javadoc 中有正文：

Actions supplied for dependent completions of non-async methods may be performed by the thread that completes the current CompletableFuture, or by any other caller of a completion method

所以咱们须要 join 办法期待后果的实现。

static void thenCombineAsyncExample() {
    String original = "Message";
    CompletableFuture cf = CompletableFuture.completedFuture(original)
            .thenApplyAsync(s -> delayedUpperCase(s))
            .thenCombine(CompletableFuture.completedFuture(original).thenApplyAsync(s -> delayedLowerCase(s)),
                    (s1, s2) -> s1 + s2);
    assertEquals("MESSAGEmessage", cf.join());
}

16、组合 CompletableFuture

咱们能够应用 thenCompose() 实现下面两个例子。这个办法期待第一个阶段的实现(大写转换)，它的后果传给一个指定的返回 CompletableFuture 函数，它的后果就是返回的 CompletableFuture 的后果。

有点拗口，然而咱们看例子来了解。函数须要一个大写字符串做参数，而后返回一个 CompletableFuture, 这个 CompletableFuture 会转换字符串变成小写而后连贯在大写字符串的前面。

static void thenComposeExample() {
    String original = "Message";
    CompletableFuture cf = CompletableFuture.completedFuture(original).thenApply(s -> delayedUpperCase(s))
            .thenCompose(upper -> CompletableFuture.completedFuture(original).thenApply(s -> delayedLowerCase(s))
                    .thenApply(s -> upper + s));
    assertEquals("MESSAGEmessage", cf.join());
}

17、当几个阶段中的一个实现，创立一个实现的阶段

上面的例子演示了当任意一个 CompletableFuture 实现后，创立一个实现的 CompletableFuture.

待处理的阶段首先创立，每个阶段都是转换一个字符串为大写。因为本例中这些阶段都是同步地执行 (thenApply), 从anyOf 中创立的 CompletableFuture 会立刻实现，这样所有的阶段都已实现，咱们应用 whenComplete(BiConsumer<? super Object, ? super Throwable> action) 解决实现的后果。

static void anyOfExample() {StringBuilder result = new StringBuilder();
    List messages = Arrays.asList("a", "b", "c");
    List<CompletableFuture> futures = messages.stream()
            .map(msg -> CompletableFuture.completedFuture(msg).thenApply(s -> delayedUpperCase(s)))
            .collect(Collectors.toList());
    CompletableFuture.anyOf(futures.toArray(new CompletableFuture[futures.size()])).whenComplete((res, th) -> {if(th == null) {assertTrue(isUpperCase((String) res));
            result.append(res);
        }
    });
    assertTrue("Result was empty", result.length() > 0);
}

18、当所有的阶段都实现后创立一个阶段

上一个例子是当任意一个阶段实现后接着解决，接下来的两个例子演示当所有的阶段实现后才持续解决, 同步地形式和异步地形式两种。

static void allOfExample() {StringBuilder result = new StringBuilder();
    List messages = Arrays.asList("a", "b", "c");
    List<CompletableFuture> futures = messages.stream()
            .map(msg -> CompletableFuture.completedFuture(msg).thenApply(s -> delayedUpperCase(s)))
            .collect(Collectors.toList());
    CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()])).whenComplete((v, th) -> {futures.forEach(cf -> assertTrue(isUpperCase(cf.getNow(null))));
        result.append("done");
    });
    assertTrue("Result was empty", result.length() > 0);
}

19、当所有的阶段都实现后异步地创立一个阶段

应用 thenApplyAsync() 替换那些单个的 CompletableFutures 的办法，allOf()会在通用池中的线程中异步地执行。所以咱们须要调用 join 办法期待它实现。

static void allOfAsyncExample() {StringBuilder result = new StringBuilder();
    List messages = Arrays.asList("a", "b", "c");
    List<CompletableFuture> futures = messages.stream()
            .map(msg -> CompletableFuture.completedFuture(msg).thenApplyAsync(s -> delayedUpperCase(s)))
            .collect(Collectors.toList());
    CompletableFuture allOf = CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]))
            .whenComplete((v, th) -> {futures.forEach(cf -> assertTrue(isUpperCase(cf.getNow(null))));
                result.append("done");
            });
    allOf.join();
    assertTrue("Result was empty", result.length() > 0);
}

20、实在的例子

当初你曾经理解了 CompletionStage 和 CompletableFuture 的一些函数的性能，上面的例子是一个实际场景：

  1. 首先异步调用 cars 办法取得 Car 的列表，它返回 CompletionStage 场景。cars生产一个近程的 REST API。
  2. 而后咱们复合一个 CompletionStage 填写每个汽车的评分，通过 rating(manufacturerId) 返回一个 CompletionStage, 它会异步地获取汽车的评分(可能又是一个 REST API 调用)
  3. 当所有的汽车填好评分后，咱们完结这个列表，所以咱们调用 allOf 失去最终的阶段，它在后面阶段所有阶段实现后才实现。
  4. 在最终的阶段调用whenComplete(), 咱们打印出每个汽车和它的评分。
cars().thenCompose(cars -> {List<CompletionStage> updatedCars = cars.stream()
            .map(car -> rating(car.manufacturerId).thenApply(r -> {car.setRating(r);
                return car;
            })).collect(Collectors.toList());
 
    CompletableFuture done = CompletableFuture
            .allOf(updatedCars.toArray(new CompletableFuture[updatedCars.size()]));
    return done.thenApply(v -> updatedCars.stream().map(CompletionStage::toCompletableFuture)
            .map(CompletableFuture::join).collect(Collectors.toList()));
}).whenComplete((cars, th) -> {if (th == null) {cars.forEach(System.out::println);
    } else {throw new RuntimeException(th);
    }
}).toCompletableFuture().join();

因为每个汽车的实例都是独立的，失去每个汽车的评分都能够异步地执行，这会进步零碎的性能 (提早)，而且，期待所有的汽车评分被解决应用的是allOf 办法，而不是手工的线程期待(Thread#join() 或 a CountDownLatch)。

最初，关注公众号 Java 技术栈，在后盾回复：面试，能够获取我整顿的 Java 并发多线程系列面试题和答案，十分齐全。

原文：https://mahmoudanouti.wordpre…\
起源：https://colobu.com/2018/03/12…

参考文档：

https://medium.com/@johnmccle…\
http://javaday.org.ua/complet…\
https://blog.takipi.com/back-…\
https://github.com/spotify/co…

近期热文举荐：

1.600+ 道 Java 面试题及答案整顿(2021 最新版)

2. 终于靠开源我的项目弄到 IntelliJ IDEA 激活码了，真香！

3. 阿里 Mock 工具正式开源，干掉市面上所有 Mock 工具！

4.Spring Cloud 2020.0.0 正式公布，全新颠覆性版本！

5.《Java 开发手册（嵩山版）》最新公布，速速下载！

感觉不错，别忘了顺手点赞 + 转发哦！

退出移动版