JUC 之 CompletableFuture

——CompletableFuture

Future

• Future 接口（FutureTask 实现类） 定义了操作异步任务执行的一些方法，如获取异步的执行结果、取消任务的执行、判断任务是否被取消、判断任务执行是否完毕
• 例如： 主线程让一个子线程去执行任务，子线程可能比较耗时，启动子线程开始执行任务后，主线程就去做其他事情了，忙其他事情或者先执行完，过了一会才去获取子任务的执行结果或变更的任务状态
• Future 接口可以为主线程开一个分支任务，专门为主线程处理耗时和费力的复杂业务
• Future 是 Java5 新加的一个接口，提供了一种异步并行计算的功能。如果主线程需要执行一个很耗时的计算任务，就可以通过 Future 把这个任务放到异步线程中执行。主线程继续处理其他任务或者先行结束，再通过 Future 获取计算结果
• 异步多线程任务执行且返回有结果（特点：多线程、有返回、异步任务）

优点

• Future + 线程池异步多线程任务配合，能显著提高程序的执行效率

缺点

• futureTask.get() 容易导致阻塞，一般建议放在程序后面，一旦调用，非要等到结果才会离开，不管你是否计算完成；假如不愿意等待很长时间，规定时间内自动离开，futureTask.get(3, TimeUnit.SECENDS)
• isDone()轮询的方式会耗费无畏的CPU资源，而且也不见得能及时地得到计算结果；如果想要异步获取结果，通常都会以轮询的方式获取结果，尽量不要阻塞

问题 / 改进

• 简单的业务场景使用 Future 完全 OK
• 通过轮询的方式去判断任务是否完成，非常占 CPU 并且代码也不优雅
• 假设业务需求，想将多个异步任务的计算结果组合起来，后一个异步任务的计算结果需要前一个异步任务的值；将两个或多个异步计算合成一个异步计算，这几个异步计算互相独立，同时后面这个又依赖前一个处理的结果
• 当 Future 集合中某个任务最快结束时，返回第一名处理结果
• 至此引出 CompletableFuture，以声明式的方式优雅处理这些需求，Future 能干的，CompletableFuture 都能干

CompletableFuture 概念

为什么会出现

• 对于真正的异步处理，我们希望是可以通过传入回调函数，在Future结束时自动调用该函数，这样就不用等待结果
• 阻塞的方式和异步编程的设计理念互相违背，而轮询的方式会耗费无畏的 CPU 资源
• CompletableFuture 提供了一种观察者模式类似的机制，可以让任务执行完成后通知监听的一方

是什么

• 在 Java 8 中，CompletableFuture 提供了非常强大的 Future 的扩展功能，可以帮我们简化异步编程的复杂性，并且提供了函数式编程的能力，可以通过回调的方式处理计算结果，也提供了转换和组合 CompletableFuture 的方法
• 它可能代表一个明确完成的 Future，也可能代表一个完成阶段（CompletionStage），它支持在计算完成以后触发一些函数或执行某些动作

实现 Future 和 CompletionStage 接口

• CompletionStage
  • 代表异步计算过程中的某一个阶段，一个阶段完成以后可能会触发另一个阶段，有些类似 Linux 系统的管道分隔符传参数
  • 一个阶段的计算执行可以是一个 Function，Consumer 或者 Runnable
  • 一个阶段的执行可能是被单个阶段的完成触发，也可能是由多个阶段一起触发

核心静态方法

• runAsync 无返回值
  • runAsync(Runnable runnable)
  • runAsync(Runnable runnable, Executor executor)
• supplyAsync 有返回值
  • supplyAsync(Supplier<U> supplier)
  • supplyAsync(Supplier<U> supplier, Executor executor)
• 上述 Executor executor 参数说明
  • 没有指定 Executor 的方法，直接使用默认的 ForkJoinPool.commonPool()
  • 如果指定线程池，则使用我们自定义或者特别指定的线程池执行异步代码

优点

• 异步任务结束时，会自动回调某个对象的方法
• 主线程设置好回调后，不再关心异步任务的执行，异步任务之间可以顺序执行
• 异步任务出错时，会自动回调某个对象的方法

get() 和 join() 方法区别

• 调用 join 方法时，编译时不会报出检查时的异常
• 调用 get 方法时，编译时会报出检查时的异常

线程池说明

• 没有传入自定义线程池，都用默认线程池 ForkJoinPool
• 如果你执行第一个任务的时候，传入了一个自定义线程池
  • 调用 thenRun 方法执行第二个任务时，则第二个任务和第一个任务是共用同一个线程池
  • 调用 thenRunAsync 执行第二个任务时，则第一个任务使用的是你自己传入的线程池，第二个任务使用的是 ForkJoinPool 线程池
• 有可能处理太快，系统优化切换原则，直接使用 main 线程处理

常用方法API

获得结果和触发计算

• get() ：调用时阻塞
• get(long timeout, TimeUnit unit) ： 超过时间不等待
• join() ：不抛异常
• getNow(T valueIfAbsent) ：提供默认值，计算完成返回结果，不阻塞
• complete(T vlaue)：是否打断 get 方法立即返回括号值

对计算结果进行处理

• thenApply：计算结果存在依赖关系，这两个线程串行化；由于存在依赖关系，当前步骤有异常就叫停
• handle：计算结果存在依赖关系，这两个线程串行化；有异常也可以往下一步走，根据带的异常参数可以进一步处理

对计算结果进行消费

• thenRun：任务 A 执行完执行 B，并且 B 不需要 A 的结果
• thenAccept：任务 A 执行完执行 B，B 需要 A 的结果，但是任务 B 无返回值
• thenApply：任务 A 执行完执行 B， B 需要 A 的结果，同时任务 B 有返回值

对计算速度进行选用

• applyToEither： 对于两个线程相比较，哪个比较快就用哪个

对计算结果进行合并

• thenCombine：两个 CompletionStage 任务都完成后，最终能把两个任的结果一起交给 thenCombine 来处理，先完成的等待另一个完成