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

3次阅读

共计 11970 个字符，预计需要花费 30 分钟才能阅读完成。

在 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 的一些函数的性能，上面的例子是一个实际场景：

首先异步调用 cars 办法取得 Car 的列表，它返回 CompletionStage 场景。cars生产一个近程的 REST API。
而后咱们复合一个 CompletionStage 填写每个汽车的评分，通过 rating(manufacturerId) 返回一个 CompletionStage, 它会异步地获取汽车的评分(可能又是一个 REST API 调用)
当所有的汽车填好评分后，咱们完结这个列表，所以咱们调用 allOf 失去最终的阶段，它在后面阶段所有阶段实现后才实现。
在最终的阶段调用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 开发手册（嵩山版）》最新公布，速速下载！

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

正文完

java

发表至： java

2021-03-25

0

java基于虹软sdk实现人脸识别（demo）

Spring-BeanDefinition的加载

关于java:解决-idea-web项目没有小蓝点的问题

关于java:Java-中-this-和-super-的用法详解

关于javascript:G6-TreeGraph自动收起其相邻节点

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

站内搜索