那些年我们追过的异步骚操作：ListenableFuture、CompletableFuture、RxJava(Observable)

- 日理万妓 2022-03-25 16:50 234阅读 0赞

[Futures][]在Java 5(2004)中引入。它们是承诺在操作完成后保留操作结果的对象。调用者可以使用future对象来检查操作[isDone()][isDone]，或者等待它完成使用[get()][get]。Future模式一个最大的问题是何时调用问题（**过早地阻塞Future.get()，这消除了异步执行的好处**）。guava的 ListenableFuture、java8的 CompletableFuture、RxJava的 Observable 都给出了相应的解决方案，该如何选择这些，本文将就此展开探讨。

### 1. ListenableFuture ###

guava在很早的时候（估计Java 5）已经开源，它的增强主要是通过 MoreExecutor 和 Futures 两个类，如果你要研究源码也可以从他们下手。

*  ListenableFuture 继承了Future (jdk)，额外新增了一个方法（任务结束后的回调方法），void addListener(Runnable listener, Executor executor);  其中executor是回调方法的执行器(通常是线程池)。需要注意的是不加Executor的情况，只适用于轻型的回调方法，如果回调方法很耗时占资源，会造成线程阻塞， 因为DirectExecutor有可能在主线程中执行回调
 *  ListenableFutureTask 继承了FutureTask (jdk)并实现了ListenableFuture
 *  ListeningExecutorService 继承 ExecutorService (jdk)接口，重写了submit方法，修改返回值类型为ListenableFuture

它提供的功能也特别多：

*  监听任务执行结果并执行回调方法，Futures.addCallback
 *  提供方便的任务接口转换，Futures.transform()、Futures.transformAsync()
 *  多线程并发执行取结果集合，Futures.allAsList()

注：Futures.transform()和Futures.addCallback()都是对addListener做了封装，进行回调的设置，但是**transform更适合用在链式处理的中间过程**，addCallback更适合用在处理最终的结果上。

//线程池
 ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(5, 5, 0, TimeUnit.SECONDS, new ArrayBlockingQueue<>(100), new CustomizableThreadFactory("demo"), new ThreadPoolExecutor.DiscardPolicy());
 ListeningExecutorService listeningExecutor = MoreExecutors.listeningDecorator(poolExecutor);
 
 //获得一个随着jvm关闭而关闭的线程池，通过Runtime.getRuntime().addShutdownHook(hook)实现,修改ThreadFactory为创建守护线程，默认jvm关闭时最多等待120秒关闭线程池，重载方法可以设置时间
 ExecutorService newPoolExecutor = MoreExecutors.getExitingExecutorService(poolExecutor);
 
 //只增加关闭线程池的钩子，不改变ThreadFactory
 MoreExecutors.addDelayedShutdownHook(poolExecutor, 120, TimeUnit.SECONDS);
 
 //提交任务
 ListenableFuture<String> listenableFuture = listeningExecutor.submit(new Callable<String>() {
 @Override
 public String call() throws Exception {
 return "";
 }
 });
 //注册回调
 Futures.addCallback(listenableFuture, new FutureCallback<String>() {
 @Override
 public void onSuccess(String result) {
 }
 @Override
 public void onFailure(Throwable t) {
 }
 });
 /**
 * 链式语法
 **/
 ListenableFutureTask<String> task1 = ListenableFutureTask.create(new Callable<String>() {
 @Override
 public String call() throws Exception {
 return "";
 }
 });
 new Thread(task1).start();
 task1.addListener(new Runnable() {
 @Override
 public void run() {
 ListenableFutureTask<String> task2 = ListenableFutureTask.create(new Callable<String>() {
 @Override
 public String call() throws Exception {
 return "";
 }
 });
 task2.addListener(new Runnable() {
 @Override
 public void run() {
 ...
 }
 }, MoreExecutors.directExecutor());
 new Thread(task2).start();
 }
 }, MoreExecutors.directExecutor());
 ListenableFuture<String> task2 = Futures.transform(task1, new Function<String, String>() {
 @Override
 public String apply(String input) {
 return "";
 }
 });
 ListenableFuture<String> task3 = Futures.transform(task2, new Function<String, String>() {
 @Override
 public String apply(String input) {
 return "";
 }
 });
 //处理最终的异步任务
 Futures.addCallback(task3, new FutureCallback<String>() {
 @Override
 public void onSuccess(String result) { 
 }
 @Override
 public void onFailure(Throwable t) {
 }
 });

### 2. CompletableFuture ###

JDK8中开始引入的[CompletableFuture][]，实际上是由Google的ListenableFuture 启发的常规期货的演变，所以在一定程度上他们非常类似，但它提供了更强大的 Future 的扩展功能。比如：ListenableFuture 的addCallback监听回调，相当于thenRun或者whenComplete操作原语。

*  CompletionStage，代表异步计算过程中的某一个阶段，一个阶段完成以后可能会触发另外一个阶段，一个阶段的计算执行可以是一个Function，Consumer或者Runnable
 *  CompletableFuture，它实现了Future 与 CompletionStage，它提供了函数式编程的能力，可以通过回调的方式处理计算结果，也提供了转换和组合的方法。原理:内部通过阻塞队列+FutureTask，实现了**任务先完成可优先获取到，即结果按照完成先后顺序排序。**

**回调函数的线程**

提供了非异步的完成操作，没有显式入参Executor的所有async方法都使用 `ForkJoinPool.commonPool()`为了简化监视、调试和跟踪，所有生成的异步任务都是标记接口 `AsynchronousCompletionTask`的实例。`注意：这些线程都是Daemon线程，主线程结束Daemon线程不结束，只有JVM关闭时，生命周期终止`。

public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
 /**
 * 四种创建异步任务(*Async)
 * runAsync则用于没有返回值的异步任务
 * supplyAsync用于有返回值的异步任务
 **/
 public static CompletableFuture supplyAsync(Supplier supplier)
 public static CompletableFuture supplyAsync(Supplier supplier, Executor executor)
 public static CompletableFuture<Void> runAsync(Runnable runnable) 
 public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)
 /**
 * 本stage出现了异常，会进入，可以对异常处理再返回新的CompletionStage
 **/
 public CompletableFuture<T> exceptionally(Function<Throwable, ? extends T> fn) 
 /**
 * 本stage正常完成后或出现了异常，来执行无需返回值的Action
 * 不以Async结尾的方法由原来的线程计算，以Async结尾的方法由默认的线程池ForkJoinPool.commonPool()
 **/
 public CompletableFuture<T> whenComplete(BiConsumer<? super T,? super Throwable> action)
 public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T,? super Throwable> action)
 public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T,? super Throwable> action, Executor executor)
 public CompletableFuture<T> exceptionally(Function<Throwable,? extends T> fn) 
 /**
 * 转换
 * 本stage完成后（转换并不是马上执行的，也不会阻塞，而是在前一个stage完成后继续执行），用其结果构建一个新的stage并返回
 **/
 public CompletableFuture thenApply(Function<? super T,? extends U> fn)
 public CompletableFuture thenApplyAsync(Function<? super T,? extends U> fn)
 public CompletableFuture thenApplyAsync(Function<? super T,? extends U> fn, Executor executor)
 /**
 * 转换和异常处理的两个功能
 * 相当于thenApply和exceptionally加强版，可操作空间大
 **/
 public CompletableFuture handle(BiFunction<? super T,Throwable,? extends U> fn)
 public CompletableFuture handleAsync(BiFunction<? super T,Throwable,? extends U> fn)
 public CompletableFuture handleAsync(BiFunction<? super T,Throwable,? extends U> fn, Executor executor) 
 /**
 * 纯消费(执行Void Action)
 * 本stage完成后，来执行无需返回值的 Consumer
 **/
 public CompletableFuture<Void> thenAccept(Consumer<? super T> action)
 public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action)
 public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action, Executor executor)
 //与thenAccept不同,下面三个不使用计算的结果
 public CompletableFuture<Void> thenRun(Runnable action)
 public CompletableFuture<Void> thenRunAsync(Runnable action)
 public CompletableFuture<Void> thenRunAsync(Runnable action, Executor executor)
 /**
 * 纯消费2
 * 当两个CompletionStage都正常完成计算的时候，会执行提供的action/Runnable
 **/
 public CompletableFuture<Void> thenAcceptBoth(CompletionStage<? extends U> other, BiConsumer<? super T,? super U> action)
 public CompletableFuture<Void> thenAcceptBothAsync(CompletionStage<? extends U> other, BiConsumer<? super T,? super U> action)
 public CompletableFuture<Void> thenAcceptBothAsync(CompletionStage<? extends U> other, BiConsumer<? super T,? super U> action, Executor executor)
 public CompletableFuture<Void> runAfterBoth(CompletionStage<?> other, Runnable action)
 public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other, Runnable action) 
 /**
 * 组合
 * 本stage正常完成后，用其结果构造一个新的stage，和thenApply类似,只不过这个需要自己构造stage，thenApply会自动构造，它对于构建异步的pipeline非常有用
 **/
 public CompletableFuture thenCompose(Function<? super T,? extends CompletionStage> fn)
 public CompletableFuture thenComposeAsync(Function<? super T,? extends CompletionStage> fn)
 public CompletableFuture thenComposeAsync(Function<? super T,? extends CompletionStage> fn, Executor executor)
 /**
 * 组合2
 * 本stage和给定的stage都正常完成后，执行一个BiFunction
 * 两个stage是并行执行的，它们之间并没有先后依赖顺序，与thenAcceptBoth区别在于有返回值
 **/
 public <U,V> CompletableFuture<V> thenCombine(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn)
 public <U,V> CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn)
 public <U,V> CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn, Executor executor)		
 
 /**
 * Either
 * 当任意一个stage完成的时候，action这个消费者就会被执行
 * applyToEither方法是当任意一个stage完成的时候，fn会被执行，它的返回值会当作新的CompletableFuture的计算结果
 **/
 public CompletableFuture<Void> acceptEither(CompletionStage<? extends T> other, Consumer<? super T> action)
 public CompletableFuture<Void> acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action)
 public CompletableFuture<Void> acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action, Executor executor)
 public CompletableFuture applyToEither(CompletionStage<? extends T> other, Function<? super T,U> fn)
 public CompletableFuture applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T,U> fn)
 public CompletableFuture applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T,U> fn, Executor executor)
 /**
 * allOf 和 anyOf
 * allOf方法是当所有的CompletableFuture都执行完后执行计算
 * anyOf方法是当任意一个CompletableFuture执行完后就会执行计算，计算的结果相同
 **/ 
 public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs)
 public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs)
 }

public class CompletableFuture_test {
 private CompletableFuture<Integer> compFuture = CompletableFuture.supplyAsync(() -> {
 System.out.println("compFuture");
 return 100;
 });
 
 private long getTimeStamp() {
 long timeMillis = System.currentTimeMillis();
 if (timeMillis % 2 == 0) throw new RuntimeException("custom uncheck exception");
 return timeMillis;
 }
 
 //正常or异常
 @Test
 public void go_whenComplete() throws ExecutionException, InterruptedException {
 CompletableFuture<Long> compFuture2 = CompletableFuture.supplyAsync(this::getTimeStamp);
 Future<Long> future = compFuture2.whenComplete((v, e) -> {
 System.out.println(v);//exception时 null
 System.out.println("exception:" + e);
 });
 System.out.println("result:" + future.get());
 }
 
 //转换
 @Test
 public void go_thenApplyAsync() throws ExecutionException, InterruptedException {
 CompletableFuture<String> compFuture2 = compFuture.thenApplyAsync(x -> x * 10).thenApply(x -> x.toString());
 System.out.println(compFuture2.get());
 }
 
 //纯消费
 @Test
 public void go_thenAccept_thenAcceptBoth() throws ExecutionException, InterruptedException {
 CompletableFuture<Void> f = compFuture.thenAccept(System.out::println);
 System.out.println(f.get());//返回null
 //两个CompletionStage都正常完成时执行
 CompletableFuture<Void> f2 = compFuture.thenAcceptBoth(CompletableFuture.completedFuture(10),
 (x, y) -> System.out.println(x * y));
 System.out.println(f2.get());
 }
 
 @Test
 public void go_thenRun() throws ExecutionException, InterruptedException {
 CompletableFuture<Void> f = compFuture.thenRun(() -> System.out.println("finished"));
 System.out.println(f.get());
 }
 
 //组合
 @Test
 public void go_thenCompose() throws ExecutionException, InterruptedException {
 //一个个排队
 CompletableFuture<String> f = compFuture.thenCompose(i -> {
 return CompletableFuture.supplyAsync(() -> {
 return (i * 10) + "";
 });
 });
 System.out.println(f.get());
 
 CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
 return "abc";
 });
 //并行
 CompletableFuture<String> f2 = compFuture.thenCombine(future2, (x, y) -> y + "-" + x);
 System.out.println(f2.get()); //abc-100
 }
 
 //当任意一个完成
 @Test
 public void go_applyToEither() throws ExecutionException, InterruptedException {
 CompletableFuture<Integer> compFuture2 = CompletableFuture.supplyAsync(() -> 2);
 CompletableFuture<String> f = compFuture.applyToEither(compFuture2, i -> i.toString());
 System.out.println(f.get());
 }
 
 @Test
 public void go_allOf_anyOf() throws ExecutionException, InterruptedException {
 CompletableFuture<Integer> compFuture2 = CompletableFuture.supplyAsync(() -> 2);
 CompletableFuture<Integer> compFuture3 = CompletableFuture.supplyAsync(() -> 4);
 //所有完成
 CompletableFuture<Object> f = CompletableFuture.anyOf(compFuture, compFuture2, compFuture3);
 //任何一个
 //CompletableFuture<Void> f = CompletableFuture.anyOf(compFuture,compFuture2,compFuture3);
 System.out.println(f.get());
 }
 // @Test
 // public void getProductById(String productId) throws InterruptedException, ExecutionException {
 // CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> getProductDetailInfo(productId));
 // CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> getProductSkuInfo(productId));
 // //等待三个数据源都返回后，再组装数据。这里会有一个线程阻塞
 // CompletableFuture.allOf(f1, f2).join();
 //
 // String detail = f1.get();
 // String sku = f2.get();
 // System.out.println(detail + "@" + sku);
 // }
 //汇总计算结果
 @Test
 public void go_sequence() throws ExecutionException, InterruptedException {
 CompletableFuture<Integer> compFuture2 = CompletableFuture.supplyAsync(() -> 2);
 CompletableFuture<Integer> compFuture3 = CompletableFuture.supplyAsync(() -> 4);
 List<CompletableFuture<Integer>> futures= Lists.newArrayList(compFuture,compFuture2,compFuture3);
 CompletableFuture<List<Integer>> f = sequence(futures);
 System.out.println(f.get());
 }
 //拓展：内部是通过allOf和lamda实现的
 public static <T> CompletableFuture<List<T>> sequence(List<CompletableFuture<T>> futures) {
 CompletableFuture<Void> allDoneFuture = CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]));
 return allDoneFuture.thenApply(v -> futures.stream().map(CompletableFuture::join).collect(Collectors.<T>toList()));
 }
 }

**小结**

`Runnable`类型的参数会忽略计算的结果，`Consumer`是纯消费计算结果，`BiConsumer`会组合另外一个`CompletionStage`纯消费，`Function`会对计算结果做转换，`BiFunction`会组合另外一个`CompletionStage`的计算结果做转换。

**thenApply 与handle方法的区别？**

handle方法会处理正常计算值和异常，因此它可以屏蔽异常，避免异常继续抛出。而thenApply 方法只是用来处理正常值，因此一旦有异常就会抛出。

### 3. **RxJava** ###

[RxJava][] 用于[reactive programming][]，它似乎与[Java 8's streams][Java 8_s streams]相似，但是它更强大。Java 9 Reactive Streams 也有它的身影。

与 Futures 类似，RxJava可用于将一堆同步或异步操作串联起来以产生一个或多个有意义的结果。然而，与一次性使用的Futures不同，RxJava可以处理零个或多个项目的流，包括具有无限数量项目的永无止境的流。由于令人难以置信的丰富[set of operators][]，它也更加灵活和流畅。

与Java 8的流不同，RxJava具有[backpressure][]机制，允许它处理处理流的不同部分以不同的速率在不同的线程中运行的情况。

RxJava的缺点是，尽管有相当好的文档（[之前也分享过][Link 1]），但由于涉及范式转换，它是一个**具有挑战性的库**。 Rx代码也可能是调试的噩梦，特别是如果涉及多个线程，更糟糕的是 - 如果需要背压。

### 4. 总结 ###

如果你想要写出高逼格的代码，更友好的链式调用，更方便的维护代码，要解决谜之缩进确实是个挑战，而RxJava正是符合你的意愿。![89576152][]

**参阅资料**

*  [https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CompletableFuture.html][CompletableFuture]
 *  [https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CompletionStage.html][https_docs.oracle.com_javase_8_docs_api_java_util_concurrent_CompletionStage.html]

[Futures]: https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Future.html
[isDone]: https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Future.html#isDone--
[get]: https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Future.html#get-long-java.util.concurrent.TimeUnit-
[CompletableFuture]: https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CompletableFuture.html
[RxJava]: https://github.com/ReactiveX/RxJava
[reactive programming]: https://en.wikipedia.org/wiki/Reactive_programming
[Java 8_s streams]: http://www.oracle.com/technetwork/articles/java/ma14-java-se-8-streams-2177646.html
[set of operators]: http://reactivex.io/documentation/operators.html
[backpressure]: http://reactivex.io/documentation/operators/backpressure.html
[Link 1]: https://blog.csdn.net/alex_xfboy/article/details/83089734
[89576152]: /images/20220325/2e5c547c4fd845f5b7a2bc85e7f932e8.png
[https_docs.oracle.com_javase_8_docs_api_java_util_concurrent_CompletionStage.html]: https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CompletionStage.html