概述
Java8之前用 Future 处理异步请求, 当你需要获取任务结果时, 通常的做法是调用 get(long timeout, TimeUnit unit) 此方法会阻塞当前的线程, 如果任务处理超时, 就会抛出一个 TimeoutException
@Test
public void test1() throws InterruptedException, ExecutionException, TimeoutException {
ExecutorService executorService = Executors.newCachedThreadPool();
Future<String> f = executorService.submit(() -> "ceshishanghu");
String s = f.get(3, TimeUnit.SECONDS);
System.out.println(s);
}
在Java8中引入了 CompletableFuture, 使用它提供的API可以不用像之前那样阻塞式或轮询的获取某个异步任务的结果, CompletableFuture 会在异步任务处理完成后自动进行回调, 让你可以链式的组合多个异步任务。
CompletableFuture 类中提供了许多以 Async 后缀结尾的方法。通常而言，名称中不带 Async 的方法和它的前一个任务一样，在同一个线程中运行。而名称以 Async 结尾的方法会将后续的任务提交到一个线程池，所以每个任务是由不同的线程处理的。
静态工厂方法
- supplyAsync(): 异步处理任务, 有返回值
- runAsync(): 异步处理任务, 没有返回值
- allOf(): 需要等待所有的异步任务都执行完毕,才会返回一个新的CompletableFuture
- anyOf(): 任意一个异步任务执行完毕,就会返回一个新的CompletableFuture
- completedFuture(): 这种方式获取的 CompletableFuture 不是异步的，它会等待获取明确的返回结果之后再返回一个已经完成的 CompletableFuture
@Test
public void test2() {
//创建一个已经有任务结果的CompletableFuture
CompletableFuture<String> f1 = CompletableFuture.completedFuture("return value");
//异步处理任务,有返回值
CompletableFuture<String> f2 = CompletableFuture.supplyAsync(this::get);
//异步处理任务,没有返回值
CompletableFuture<Void> f3 = CompletableFuture.runAsync(System.out::println);
//需要等待所有的异步任务都执行完毕,才会返回一个新的CompletableFuture
// CompletableFuture<Void> all = CompletableFuture.allOf(f1, f2, f3);
//任意一个异步任务执行完毕,就会返回一个新的CompletableFuture
CompletableFuture<Object> any = CompletableFuture.anyOf(f1, f2, f3);
Object result = any.join();
System.out.println("result = " + result);//result = return value
} public String get() {
delay();
return "异步任务结果";
} public void delay() {
try {
Thread.sleep(1000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
上面的示例中, allOf() 因为要等待所有的异步任务执行完成，所以要延时1秒钟才会返回一个新的 CompletableFuture, 而 anyOf() 则不需要等待所有的异步任务, 因为第一个异步最先完成, 所以控制台输出 result = return value 。
链式调用
A任务执行完毕, 继续执行B任务, B任务执行完毕, 继续执行C任务...
@Test
public void test2() {
CompletableFuture<Void> f = CompletableFuture.supplyAsync(() -> {
//测试抛异常后,handle()方法接受并处理
//int x = 1 / 0;
return "这是一个栗子";
}).handle((res, ex) -> {
System.out.println("handle res = " + res);
if (Objects.nonNull(ex)) {
System.out.println("handle ex" + ex.getCause().getMessage());
}
return Objects.nonNull(ex) ? 0 : 1;
}).thenApply(res -> {
System.out.println("thenApply res = " + res);
return res == 1 ? "success" : "error";
}).thenAccept(res -> System.out.println("thenAccept res = " + res)
).thenRun(() -> System.out.println("没有参数, 异步执行一个没有返回值的任务"));
f.join();
}
输出结果:
handle res = 这是一个栗子
thenApply res = 1
thenAccept res = success
没有参数, 异步执行一个没有返回值的任务
将上面 int x = 1 / 0; 这行代码取消注释, 重新运行结果如下:
handle res = null
handle ex/ by zero
thenApply res = 0
thenAccept res = error
没有参数, 异步执行一个没有返回值的任务
可以看到, handle() 方法接受前一个 CompletableFuture 的返回结果或抛出的异常作为方法入参, 经过处理后再返回一个新的结果。
级联组合
- thenCompose(): 对两个异步操作进行组合，第一个操作完成时，将其结果作为参数传递给第二个操作, 第二个操作会返回一个新的CompletableFuture。
- thenCombine(): 将两个完全无关联的异步请求的结果整合起来, 计算出一个新的值并返回
@Test
public void test3() {
CompletableFuture<String> f = CompletableFuture.completedFuture("CompletableFuture 1");
CompletableFuture<String> f1 = f.thenCompose(res -> {
System.out.println("thenCompose res = " + res);
return CompletableFuture.supplyAsync(() -> "CompletableFuture 2");
});
System.out.println(f1.join());
CompletableFuture<Integer> f3 = CompletableFuture.completedFuture(998);
CompletableFuture<String> f4 = f.thenCombine(f3, (str, num) -> {
System.out.println("str = " + str + ", num= " + num);
return str + num;
});
System.out.println(f4.join());
}
输出结果:
thenCompose res = CompletableFuture 1
CompletableFuture 2
str = CompletableFuture 1, num= 998
CompletableFuture 1998
whenComplete
当前一个 CompletableFuture 计算完成或抛出异常时, 可以使用 whenComplete() 执行指定的任务。
@Test
public void test4() {
CompletableFuture<String> f = CompletableFuture.supplyAsync(() -> {
//测试抛异常后,whenComplete()方法接受并处理
int x = 1 / 0;
return "这是一个栗子";
}).whenComplete((res, ex) -> {
System.out.println("whenComplete res = " + res);
if (Objects.nonNull(ex)) {
System.out.println("whenComplete ex" + ex.getCause().getMessage());
}
});
System.out.println("f.join() = " + f.join());
}
输出结果如下，其中 res 对应前一个 CompletableFuture 的返回结果，ex 对应前一个 CompletableFuture 抛出的异常(如果发生异常)。
从控制台输出顺序看出，当前一个 CompletableFuture 计算完成或抛出异常时, whenComplete() 会接受它的返回结果或抛出的异常，来做一些其他的事情，最后再返回原来的返回结果或抛出异常。类比下 try/catch 语句块中的 final 语句块。
whenComplete res = null
whenComplete ex/ by zero java.util.concurrent.CompletionException: java.lang.ArithmeticException: / by zero at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:273)
at java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:280)
at java.util.concurrent.CompletableFuture$AsyncSupply.run$$$capture(CompletableFuture.java:1592)
at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java)
at java.util.concurrent.CompletableFuture$AsyncSupply.exec(CompletableFuture.java:1582)
at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157)
Caused by: java.lang.ArithmeticException: / by zero
at com.java8.action.ChapterTest.lambda$test4$0(ChapterTest.java:22)
异常处理
只有当前一个 CompletableFuture 发生异常时，才会进入到 exceptionally() 方法，并将产生的异常作为入参。
@Test
public void test5() {
CompletableFuture<String> f = CompletableFuture.supplyAsync(() -> {
//测试抛异常后,exceptionally()方法接受并处理
//int x = 1 / 0;
return "这是一个栗子";
}).exceptionally(ex -> ex.getCause().getMessage());
System.out.println("f.join() = " + f.join());
}
注释 int x = 1 / 0; ，输出如下：
f.join() = 这是一个栗子
取消注释 int x = 1 / 0; , 输出如下：
f.join() = / by zero
Both系列方法
- thenAcceptBoth(): 等待当前的 CompletableFuture 和另一个 CompletableFuture 执行完成，将它们的返回结果作为入参去执行一个操作，没有返回值
- runAfterBoth(): 等待当前的 CompletableFuture 和另一个 CompletableFuture 执行完成，然后去执行一个操作，没有返回值
代码清单一
@Test
public void test6() {
CompletableFuture<Integer> f1 = CompletableFuture.completedFuture(9523);
CompletableFuture<String> f2 = CompletableFuture.supplyAsync(this::get);
CompletableFuture<Void> both = f1.thenAcceptBoth(f2, (num, str) -> System.out.println("num = " + num + ", str = " + str));
both.join();
} public String get() {
delay();
return "CompletableFuture 2";
} public void delay() {
try {
Thread.sleep(1000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
代码清单一输出结果如下：
num = 9523, str = CompletableFuture 2
代码清单二
@Test
public void test7() {
CompletableFuture<Integer> f1 = CompletableFuture.completedFuture(9523);
CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> "");
CompletableFuture<Void> both = f1.runAfterBoth(f2, () -> System.out.println("执行一个任务，没有入参"));
both.join();
}
代码清单二输出结果如下：
执行一个任务，没有入参
Either系列
- acceptEither: 当前的 CompletableFuture 和另一个 CompletableFuture 任意一个执行完成，将对应的返回结果作为入参去执行一个操作，没有返回值
- applyToEither: 当前的 CompletableFuture 和另一个 CompletableFuture 任意一个执行完成，将对应的返回结果作为入参，使用 mapping 函数转换成一个新的值并返回
- runAfterEither: 当前的 CompletableFuture 和另一个 CompletableFuture 任意一个执行完成，然后去执行一个操作，没有返回值
代码清单三：
@Test
public void test8() {
CompletableFuture<String> f1 = CompletableFuture.completedFuture("CompletableFuture 1");
CompletableFuture<String> f2 = CompletableFuture.supplyAsync(this::get);
CompletableFuture<Void> both = f1.acceptEither(f2, System.out::println);
both.join();
} public String get() {
delay();
return "CompletableFuture 2";
} public void delay() {
try {
Thread.sleep(1000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
代码清单三输出结果：
CompletableFuture 1
代码清单四：
@Test
public void test9() {
CompletableFuture<String> f1 = CompletableFuture.supplyAsync(this::get);
CompletableFuture<String> f2 = CompletableFuture.completedFuture("CompletableFuture 2");
CompletableFuture<Integer> f3 = f1.applyToEither(f2, res -> {
System.out.println("res = " + res);
return res.length();
});
System.out.println("f3.join() = " + f3.join());
} public String get() {
delay();//这里会延时一秒钟
return "CompletableFuture 1";
}
代码清单四输出结果：
res = CompletableFuture 2
f3.join() = 19
代码清单五：
@Test
public void test10() {
CompletableFuture<String> f1 = CompletableFuture.supplyAsync(this::get);
CompletableFuture<Void> f2 = CompletableFuture.allOf();
CompletableFuture<Void> f3 = f1.runAfterEither(f2, () -> System.out.println("执行一个任务，没有入参"));
f3.join();
} public String get() {
delay();//这里会延时一秒钟
return "CompletableFuture 1";
}
代码清单五输出结果：
执行一个任务，没有入参
使用自定义的执行器来处理多个异步任务
在实际应用场景中可能会遇到这种情况，假如你需要同时处理大量的异步任务，且这些异步任务互相不依赖，你只要最后把它们的结果组装起来就行，这该怎么实现呢？
下面给出了一个使用默认执行器的示例，通过Stream流同时创建 9 个异步任务，获取它们的结果并组装后返回，其中 Runtime.getRuntime().availableProcessors() 表示Java虚拟机可用的处理器个数，在我之前的文章 Java8系列 (二) Stream流 中有介绍过。
代码清单六：
@Test
public void test11() {
List<String> list = Arrays.asList("王小波书店", "杭州沈记古旧书店", "猫的天空之城概念书店", "纯真年代书吧", "南山书屋", "西西弗书店", "新华书店", "钟书阁", "云门书屋");
System.out.println("当前机器有" + Runtime.getRuntime().availableProcessors() + "个可用的处理器");
long start = System.nanoTime();
List<CompletableFuture<String>> futures = list.stream()
.map(str -> CompletableFuture.supplyAsync(() -> this.calculateLength(str)))
.collect(Collectors.toList());
System.out.println("get futures "+(System.nanoTime() - start) / 1000_000 + " msecs");
String result = futures.stream()
.map(CompletableFuture::join)
.collect(Collectors.joining(",", "[", "]"));
System.out.println("get result "+(System.nanoTime() - start) / 1000_000 + " msecs");
System.out.println(result);
} public String calculateLength(String str) {
delay();
return str;
} public void delay() {
try {
Thread.sleep(1000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
运行代码清单六，输出结果：
当前机器有4个可用的处理器
get futures 95 msecs
get result 3098 msecs
[王小波书店,杭州沈记古旧书店,猫的天空之城概念书店,纯真年代书吧,南山书屋,西西弗书店,新华书店,钟书阁,云门书屋]
可以看到，虽然使用了异步处理，但还是花了 3098 毫秒才执行完成所有任务。这是因为 CompletableFuture 内部采用的是通用线程池 ForkJoinPool.commonPool() , 默认都使用固定数目的线程, 具体线程数取决于 Runtime.getRuntime().availableProcessors() 的返回值。
我这里测试的机器显示通用线程池中处于可用状态的线程数为 4，一次只能同时处理 4 个任务，后面的5个异步任务只能等到前面某一个操作完成释放出空闲线程才能继续, 因此总的会消耗约 3 秒钟的时间。
我们将上面的代码进行重构，使用自定义的执行器，通过自定义的执行器你可以指定线程池的大小。其中线程数的设定可以参考公式 Nthreads = NCPU * UCPU * (1 + W/C)
@Test
public void test12() {
List<String> list = Arrays.asList("王小波书店", "杭州沈记古旧书店", "猫的天空之城概念书店", "纯真年代书吧", "南山书屋", "西西弗书店", "新华书店", "钟书阁", "云门书屋");
final ExecutorService executor = Executors.newFixedThreadPool(Math.min(list.size(), 100), r -> {
Thread thread = new Thread(r);
//守护线程不会组织程序的终止
thread.setDaemon(true);
return thread;
});
System.out.println("当前机器有" + Runtime.getRuntime().availableProcessors() + "个可用的处理器, 当前处理异步请求的线程池大小为 " + Math.min(list.size(), 100));
long start = System.nanoTime();
List<CompletableFuture<String>> futures = list.stream()
.map(str -> CompletableFuture.supplyAsync(() -> this.calculateLength(str), executor))
.collect(Collectors.toList());
System.out.println("get futures " + (System.nanoTime() - start) / 1000_000 + " msecs");
String result = futures.stream()
.map(CompletableFuture::join)
.collect(Collectors.joining(",", "[", "]"));
System.out.println("get result " + (System.nanoTime() - start) / 1000_000 + " msecs");
System.out.println(result);
} public String calculateLength(String str) {
delay();
return str;
} public void delay() {
try {
Thread.sleep(1000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
输出结果如下：
当前机器有4个可用的处理器, 当前处理异步请求的线程池大小为 9
get futures 38 msecs
get result 1039 msecs
[王小波书店,杭州沈记古旧书店,猫的天空之城概念书店,纯真年代书吧,南山书屋,西西弗书店,新华书店,钟书阁,云门书屋]
可以看到，使用自定义的执行器调大线程池大小后，总的运行时间只要 1039 毫秒。
将CompletableFuture作为Controller的返回值
上面还存在一个问题，虽然现在可以同时处理多个异步任务，但是如果需要将异步结果返回给另一个服务，那不是还得通过 join() 阻塞的获取到返回值后才能再返回么？
自Spring Boot 1.3 (Spring 4.2) 之后开始支持 CompletableFuture 或 CompletionStage 作为 Controller 的返回值，她很好的解决了上面的异步阻塞问题，只要将 CompletableFuture 作为 Controller 的返回值，在异步任务执行完成后，它会自动响应结果给另一个服务。
@RestController
public class AsyncController { @GetMapping("/redirect")
public CompletableFuture<ModelAndView> redirect() {
return CompletableFuture.supplyAsync(() -> {
this.delay();
RedirectView redirectView = new RedirectView("https://www.cnblogs.com/qingshanli/");
redirectView.addStaticAttribute("hint", "CompletableFuture组装ModelAndView视图,异步返回结果");
return new ModelAndView(redirectView);
});
} @GetMapping("/async")
public CompletableFuture<String> async() {
System.out.println("async method start");
return CompletableFuture.supplyAsync(() -> {
this.delay();
return "CompletableFuture作为Controller的返回值,异步返回结果";
}).whenComplete((res, ex) -> System.out.println("async method completely, res = " + res + ", ex = " + ex));
} public void delay() {
try {
Thread.sleep(3000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
启动项目，Postman 访问 http://localhost:8080/async，截图如下：
Postman 访问 http://localhost:8080/redirect，截图如下：
参考资料
https://github.com/AndreasKl/spring-boot-mvc-completablefuture
https://nickebbitt.github.io/blog/2017/03/22/async-web-service-using-completable-future
https://www.humansreadcode.com/spring-boot-completablefuture/
作者：张小凡
出处：https://www.cnblogs.com/qingshanli/
本文版权归作者和博客园共有，欢迎转载，但未经作者同意必须保留此段声明，且在文章页面明显位置给出原文连接，否则保留追究法律责任的权利。如果觉得还有帮助的话，可以点一下右下角的【推荐】。