本文共 6410 字，大约阅读时间需要 21 分钟。
在前面的博客：
我们讲解了如何使用线程池 Executor + Future + @Async 实现异步调用。今天讲解一个 JDK1.8 的异步类：CompletableFuture。
所谓非阻塞异步编程方法，简单地说，就是不用等待返回结果的多线程的回调方法的封装。相比于阻塞式异步回调方法，非阻塞异步编程方法使用一个监听器，这样在使用回调的过程中，就能够自动感知到调用结果，从而实现了高并发的调用方法。
使用阻塞式调用，会有大量的时间耗费在等待之中。Java 开发语 Java1.8 之后 ，提供了非阻塞的回调处理方法 它封装在 CompletableFuture 类中。CompletableFuture 提供了非常强大的程序扩展能力可以帮助我们简化异步编程的复杂性，并通过非阻塞的回调方式处理调用结果。 CompletableFuture不但提供了非阻塞的异步调用的方法，还可以将处理结果进行转换和结合使用。
我们来实验一下。
定义线程池配置类：
import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;import java.util.concurrent.Executor;import java.util.concurrent.ThreadPoolExecutor;//线程池配置类@Configurationpublic class TaskPoolConfig { @Bean("myTaskExecutor") public Executor myTaskExecutor() { ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); executor.setCorePoolSize(10);//核心线程数量，线程池创建时候初始化的线程数 executor.setMaxPoolSize(15);//最大线程数，只有在缓冲队列满了之后才会申请超过核心线程数的线程 executor.setQueueCapacity(200);//缓冲队列，用来缓冲执行任务的队列 executor.setKeepAliveSeconds(60);//当超过了核心线程出之外的线程在空闲时间到达之后会被销毁 executor.setThreadNamePrefix("myTask-");//设置好了之后可以方便我们定位处理任务所在的线程池 executor.setWaitForTasksToCompleteOnShutdown(true);//用来设置线程池关闭的时候等待所有任务都完成再继续销毁其他的Bean executor.setAwaitTerminationSeconds(60);//该方法用来设置线程池中任务的等待时间，如果超过这个时候还没有销毁就强制销毁，以确保应用最后能够被关闭，而不是阻塞住。 //线程池对拒绝任务的处理策略：这里采用了CallerRunsPolicy策略，当线程池没有处理能力的时候，该策略会直接在 execute 方法的调用线程中运行被拒绝的任务；如果执行程序已关闭，则会丢弃该任务 executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); return executor; }} 定义一个 Service，模拟要处理的业务逻辑：
@Servicepublic class UserService { public String getUserName(){ return "biandan"; } public String getSign(){ return "让天下没有难写的代码"; }} @RestControllerpublic class UserController { @Autowired private UserService userService; @Autowired @Qualifier("myTaskExecutor") private Executor executor; //案例1：使用 supplyAsync 异步获取返回结果 @GetMapping("/getUserInfo") public String getUserInfo() throws ExecutionException, InterruptedException { CompletableFuture supplyAsync = CompletableFuture.supplyAsync(() -> { System.out.println("当前线程名：" + Thread.currentThread().getName()); return userService.getUserName(); }, executor); String result = supplyAsync.get(); return result; }} 说明：
1、创建 CompletableFuture 对象有几种方式：
方式1：public static CompletableFuturerunAsync(Runnable runnable)方式2：public static CompletableFuture runAsync(Runnable runnable,Executor executor)方式3：public static CompletableFuture supplyAsync(Supplier supplier)方式4：public static CompletableFuture supplyAsync(Supplier supplier,Executor executor)
其中方式1和2是不需要返回类型，方式3和4需要返回类型。
方式1和3没有使用 Executor 线程池，则会默认使用 ForkJoinPool.commonPool()作为它的线程池执行异步代码。案例1中我们使用自定义的 Executor 线程池，因此在调用的时候，会看到我们的线程池前缀。
2、静态方法 runAsync 和 supplyAsync 允许我们相应地使用Runnable和Supplier函数类型创建一个可完成的未来实例。
Runnable和Supplier都是函数接口，由于新的 java1.8 特性，它们允许将实例作为lambda表达式传递。 Runnable接口不允许返回值。Supplier接口是一个通用函数接口，它有一个方法，该方法没有参数，并且返回一个参数化类型的值。//案例2：使用 thenApply 方法将结果返回给其它函数调用 @GetMapping("/getUserInfo2") public String getUserInfo2() throws ExecutionException, InterruptedException { CompletableFuture supplyAsync = CompletableFuture.supplyAsync(() -> { System.out.println("getUserName 当前线程名：" + Thread.currentThread().getName()); return userService.getUserName(); }, executor); CompletableFuture thenApply = supplyAsync.thenApply(r -> r + " 提供给函数"); String result = thenApply.get(); return result; } 说明：使用上面的对象，调用 thenApply 即可将返回结果给其它函数调用。
①如果其它函数需要获取结果，但该函数并不需要返回值，可以定义成 Void，并使用的是 thenAccept 方法如下：
CompletableFuturethenAccept = supplyAsync.thenAccept(s->{});
②如果其它函数既不需要获取结果，也不需要返回值，只想单纯的在前面的方法执行完成后，再执行一段逻辑，可以使用 thenRun。如下：
CompletableFuturethenRun = supplyAsync.thenRun(()->{});
//案例3：使用 thenCompose 链接两个调用 @GetMapping("/getUserInfo3") public String getUserInfo3() throws ExecutionException, InterruptedException { CompletableFuture supplyAsync = CompletableFuture.supplyAsync(() -> { System.out.println("getUserName 当前线程名：" + Thread.currentThread().getName()); return userService.getUserName(); }, executor) .thenCompose(result -> CompletableFuture.supplyAsync(() -> { System.out.println("getSign 当前线程名：" + Thread.currentThread().getName()); return result + " 说 " + userService.getSign(); }, executor)); return supplyAsync.get(); } 测试结果：biandan 说 让天下没有难写的代码
说明：使用 thenCompose 可以将上一个的返回结果给下一个函数使用。
如果要链接可完成的 CompletableFuture 方法，那么最好使用 thenCompose()
//案例4：使用 thenCombine 将两个结果结合起来，给另一个调用 @GetMapping("/getUserInfo4") public String getUserInfo4() throws ExecutionException, InterruptedException { CompletableFuture result = CompletableFuture.supplyAsync(() -> { return "biandan"; }).thenCombine(CompletableFuture.supplyAsync(() -> " 说 "), (r1, r2) -> r1 + r2) .thenCompose(p -> CompletableFuture.supplyAsync(() -> p + "让天下没有难写的代码")); return result.get(); } //案例5：allof 多种结合的调用 @GetMapping("/getUserInfo5") public String getUserInfo5() throws ExecutionException, InterruptedException { CompletableFuture future1 = CompletableFuture.supplyAsync(() -> "biandan"); CompletableFuture future2 = CompletableFuture.supplyAsync(() -> "说"); CompletableFuture future3 = CompletableFuture.supplyAsync(() -> "让天下没有难写的代码"); //返回的是 void 类型 CompletableFuture combinedFuture_1 = CompletableFuture.allOf(future1, future2, future3); //使用 join 方法手动获取结果 String combinedFuture_2 = Stream.of(future1, future2, future3) .map(CompletableFuture::join) .collect(Collectors.joining(" ")); return combinedFuture_2; } 注意 CompletableFuture.allOf 的返回类型是CompletableFuture。这种方法的局限性在于它不能返回所有 Supplier 的组合结果。我们必须从未来手动获取结果。使用 CompletableFuture.join() 方法获取。
转载地址：http://ikuhf.baihongyu.com/