CompletableFuture | 隔叶黄莺 Yanbin Blog

Spring 5 响应式编程研究

2023-05-08 | 阅读(423)

Spring 5.0 发布之时(2017-09-28) WebFlux 是它的一大亮点，即响应式 Web 编程。因为同一时代的 RxJava 2 和 Akka Actor 具备一定的流行度，Spring 5 也来赶这一趟时髦。于是多线程编程大致两种模式

CompletableFuture, runAsync, supplyAsync, whenComplete...
Obervable, observeOn, subscribe, subscribeOn...

以及 PlayFramework 的 Action 方法无论返回 Result 还是 CompletableStage<Result>, 内部都是异步的模式。

Akka Actor 比 CompletableFuture, RxJava，以及本文将要讨论的 Reactor 更高级的是 Akka System 可以分布式部署，Actor 分布在不同的进程，主机上。

那时候业界已行成了一个 Reactive Stream 规范 org.reactivestreams(Publisher, Subscriber, Subscription, Processor), JDK 9 也奈不住寂寞，无法对 Reactive Stream 置若罔闻，在 2017-09-21 发布时加入了 java.util.concurrent.flow 包(Publisher, Subscriber, Processor, Subscription) 作为自己的 Reactive Stream 规范。

然而随着云计算的普及，基于消息系统解耦合的任务分解让代码变得更清晰，编码中甚至不用考虑多线程的行为，部署方式能解决任务执行的效率。

阅读全文 >>

等待所有的 CompletableFuture 完成

2019-09-23 | 阅读(4,927)

现实中有这样的用法，创建一批在线程池中运行的 CompletableFuture 实例，然后等待它们全部执行完再继续后面的操作。比如说 AWS 的 Lambda, 单单提交任务到线程池，不等待所有任务全部完成便退出主线程的话，AWS 便认为 Lambda 执行完毕，无视线程池中正在执行的任务而强行结束该 Lambda 实例。

以往我们通常的作法如下

ExecutorService threadPool = Executors.newFixedThreadPool(10);
List<CompletableFuture<Void>> futures = IntStream.rangeClosed(1, 10000)
    .mapToObj(n ->
    CompletableFuture.runAsync(() -> {
        System.out.println("done " + n);
    }, threadPool)).collect(toList());

futures.forEach(CompletableFuture::join);

System.out.println("all done");

ExecutorService threadPool = Executors.newFixedThreadPool(10);

List<CompletableFuture<Void>> futures = IntStream.rangeClosed(1, 10000)

.mapToObj(n ->

CompletableFuture.runAsync(() -> {

System.out.println("done " + n);

}, threadPool)).collect(toList());

futures.forEach(CompletableFuture::join);

System.out.println("all done");

如果所有的任务均无异常，上面的代码能得到预想的结果，只要上面打印出 all done 的话真的就是表明所有的任务都完成了。但是在循环 join futures 中的每一个 CompletableFuture 时，只要碰到任意一个任务有异常时，便立即抛出给外部线程，不在乎是否还有其他任务正在执行。此时，如果外部未予捕获，当然阅读全文 >>

CompletableFuture 的并发性能研究

2016-12-29 | 阅读(5,774)

今天继续探讨 CompletableFuture 的特性，它并发时的性能如何呢？我们知道集合的 stream() 后的操作是序列化进行的，parallelStream()是能够并发执行的，而用 CompletableFuture 可以更灵活的控制并发。

我们先可以对比一下 parallelStream() 与 CompletableFuture 的性能差异

假设一个这样的耗时 1000 毫秒的计算任务

private static int getJob() {
    try {
        Thread.sleep(1000);
    } catch (InterruptedException e) {
    }
    return 50;
}

private static int getJob() {

try {

Thread.sleep(1000);

} catch (InterruptedException e) {

}

return 50;

}

分别用下面两个方法来测试，任务数可以通过参数来控制阅读全文 >>

理解 CompletableFuture 的任务与回调函数的线程

2016-12-23 | 阅读(5,320)

继续对 CompletableFuture 的学习，本然依然不对它的众多方法的介绍，其实也不容易通过一篇述说完所有 CompletableFuture 的操作。此处重点了解下 CompletableFuture 几类操作时所使用的线程，CompletableFuture 的方法重点在它的静态方法以及实现自 CompletionStage 接口的方法，如果是意图异步化编程，反而自我标榜的 Future 中的方法用的少了。

CompletableFuture 根据任务的主从关系为

提交任务的方法，如静态方法 supplyAsync(supplier[, executor]), runAsync(runnable[, executor])
回调函数，即对任务执行后所作出回应的方法，多数方法了，如 thenRun(action), thenRunAsync(action[, executor]), whenComplete(action), whenCompleteAsync(action[, executor]) 等

根据执行方法可分为同步与异步方法，任务都是要被异步执行，所以提交任务的方法都是异步的。而对任务作出回应的方法很多分为两个版本，如

同步方法，如 thenRun(action), whenComplete(action)
异步方法，如 thenRunAsync(action[, executor]), whenCompleteAsync(action[, executor]), 异步方法可以传入线程池，否则用默认的

因此所要理解的 CompletableFuture 的线程会涉及到任务与回调函数所使用的线程。阅读全文 >>

Java 8 CompletableFuture 浅入

2016-12-23 | 阅读(3,354)

Java 1.5 有了 Future, 可谓是跨了一大步，继而 Java 1.8 新加入一个 Future 的实现 CompletableFuture, 从此线程与线程之间可以愉快的对话了。最初两个线程间的协调我采用过 Object 的 wait() 和 notify() , Thread 的 join() 方法，那可算是很低级的 API 了，是否很多 Java 程序都不知道它们的存在，或根本没用过它们。

如果是简单的等待所有线程完成可使用 Java 1.5 的 CountDownLatch, 这里有一篇介绍 CountDownLatch 协调线程, 就是实现的 waitAll(threads) 功能。而 Java 8 的 CompletableFuture 的功能就多去，可简单使用它实现异步方法。虽说 CompletableFuture 实现了 Future 接口，但它多数方法源自于 CompletionStage, 所以还里氏代换，用 Future 来引用 CompletableFuture 实例就很牵强了; 这也是为什么 PlayFramework 自 2.5 开始直接暴露的类型是 CompletionStage 而非其他两个。

顾名思义，CompletableFuture 代表着一个 Future 完成后该干点什么，具体大致有：

Future 完成后执行动作，或求取下一个 Future 的值。then...
多个 Future 的协调; 同时完成该怎么，其中一个完成该如何。allOf, anyOf

有时候可以把 Future 想像成与线程是一一对应的。阅读全文 >>

巩固 Java Future 的使用

2016-12-22 | 阅读(914)

Future 还是一 Java 1.5 带进来的产物，但过去那么多年实际代码中却很少有直接接触, 大约它多是隐匿在各种现成框架中默默的为我们服务。Future 本身不代表着多线程，而是代表着需异步计算的结果, 将来的一个期待，至于后来真正的结果不可知。在此之前想要获得一个 Runnable 在其他线程中的计算结果颇费波折，有了 Future 加之它身后的 Callable 一切就变得简单了。

对比一下 Java 1.5 前后的下面几个概念

Callable 相当于之前的 Runnable, 只是 Callable 是有返回值的
ExecuteService.submit(callable): Future 就类似于之前的 Thread(runnable)
只是前者 submit 后立即执行，通过 get() 获得结果，后者用 start() 方法启动，runnable 是没有结果的。如果你也不想关心 Future 的结果也能 ExecuteService.submit(runnable)

只有 callable 被提交(到线程池) 后返回的 Future 才可能会有结果, 所以下面的代码永远等不到结果

Future<String> future = new FutureTask<>(() -> "Never");
String result = future.get();

最容易理解的 Future 基本使用代码如下：阅读全文 >>

seetimee on 体验 Python FastAPI 的并发能力及线, 进程模型感谢
Yanbin on Mockito 3.4.0 开始可 Mock 静态方法有一个补救，新写了一篇 https://yanbin.blog/mockito-mock-static-method-in-multiple...
Yanbin on 升级到 Spring Boot 3 后 javax.inject.Named 不可用怎么，被抄袭了！算是被机器翻译引用的？
oleksi on 升级到 Spring Boot 3 后 javax.inject.Named 不可用https://www.springcloud.io/post/2023-02/springboot-3-javax-inject-name...
Zhe on 体验 Python FastAPI 的并发能力及线, 进程模型你在trio下得用 trio.sleep()，调用asyncio.sleep() 就会报错。