经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » Java相关 » Java » 查看文章
并发编程-CompletableFuture解析
来源：cnblogs　　作者：京东云技术团队　　时间：2023/7/28 10:52:30　　对本文有异议

1、CompletableFuture介绍

CompletableFuture对象是JDK1.8版本新引入的类，这个类实现了两个接口，一个是Future接口，一个是CompletionStage接口。

CompletionStage接口是JDK1.8版本提供的接口，用于异步执行中的阶段处理，CompletionStage定义了一组接口用于在一个阶段执行结束之后，要么继续执行下一个阶段，要么对结果进行转换产生新的结果等，一般来说要执行下一个阶段都需要上一个阶段正常完成，这个类也提供了对异常结果的处理接口

2、CompletableFuture的API

2.1 提交任务

在CompletableFuture中提交任务有以下几种方式：

  1. public static CompletableFuture<Void> runAsync(Runnable runnable)
  2. public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)
  3. public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
  4. public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)

这四个方法都是用来提交任务的，不同的是supplyAsync提交的任务有返回值，runAsync提交的任务没有返回值。两个接口都有一个重载的方法，第二个入参为指定的线程池，如果不指定，则默认使用ForkJoinPool.commonPool()线程池。在使用的过程中尽量根据不同的业务来指定不同的线程池，方便对不同线程池进行监控，同时避免业务共用线程池相互影响。

2.2 结果转换

2.2.1 thenApply

  1. public <U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn)
  2. public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn)
  3. public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn, Executor executor)

thenApply这一组函数入参是Function，意思是将上一个CompletableFuture执行结果作为入参，再次进行转换或者计算，重新返回一个新的值。

2.2.2 handle

  1. public <U> CompletableFuture<U> handle(BiFunction<? super T, Throwable, ? extends U> fn)
  2. public <U> CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn)
  3. public <U> CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn, Executor executor)

handle这一组函数入参是BiFunction，该函数式接口有两个入参一个返回值，意思是处理上一个CompletableFuture的处理结果，同时如果有异常，需要手动处理异常。

2.2.3 thenRun

  1. public CompletableFuture<Void> thenRun(Runnable action)
  2. public CompletableFuture<Void> thenRunAsync(Runnable action)
  3. public CompletableFuture<Void> thenRunAsync(Runnable action, Executor executor)

thenRun这一组函数入参是Runnable函数式接口，该接口无需入参和出参，这一组函数是在上一个CompletableFuture任务执行完成后，在执行另外一个接口，不需要上一个任务的结果，也不需要返回值，只需要在上一个任务执行完成后执行即可。

2.2.4 thenAccept

  1. public CompletableFuture<Void> thenAccept(Consumer<? super T> action)
  2. public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action)
  3. public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action, Executor executor)

thenAccept这一组函数的入参是Consumer，该函数式接口有一个入参，没有返回值，所以这一组接口的意思是处理上一个CompletableFuture的处理结果，但是不返回结果。

2.2.5 thenAcceptBoth

  1. public <U> CompletableFuture<Void> thenAcceptBoth(CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action)
  2. public <U> CompletableFuture<Void> thenAcceptBothAsync(CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action)
  3. public <U> CompletableFuture<Void> thenAcceptBothAsync(CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action, Executor executor)

thenAcceptBoth这一组函数入参包括CompletionStage以及BiConsumer，CompletionStage是JDK1.8新增的接口，在JDK中只有一个实现类：CompletableFuture，所以第一个入参就是CompletableFuture，这一组函数是用来接受两个CompletableFuture的返回值，并将其组合到一起。BiConsumer这个函数式接口有两个入参，并且没有返回值，BiConsumer的第一个入参就是调用方CompletableFuture的执行结果，第二个入参就是thenAcceptBoth接口入参的CompletableFuture的执行结果。所以这一组函数意思是将两个CompletableFuture执行结果合并到一起。

2.2.6 thenCombine

  1. public <U,V> CompletableFuture<V> thenCombine(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn)
  2. public <U,V> CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn)
  3. public <U,V> CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn, Executor executor)

thenCombine这一组函数和thenAcceptBoth类似，入参都包含一个CompletionStage，也就是CompletableFuture对象，意思也是组合两个CompletableFuture的执行结果，不同的是thenCombine的第二个入参为BiFunction，该函数式接口有两个入参，同时有一个返回值。所以与thenAcceptBoth不同的是，thenCombine将两个任务结果合并后会返回一个全新的值作为出参。

2.2.7 thenCompose

  1. public <U> CompletableFuture<U> thenCompose(Function<? super T, ? extends CompletionStage<U>> fn)
  2. public <U> CompletableFuture<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn)
  3. public <U> CompletableFuture<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn, Executor executor)

thenCompose这一组函数意思是将调用方的执行结果作为Function函数的入参，同时返回一个新的CompletableFuture对象。

2.3 回调方法

  1. public CompletableFuture<T> whenComplete(BiConsumer<? super T, ? super Throwable> action)
  2. public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action)
  3. public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action, Executor executor)

whenComplete方法意思是当上一个CompletableFuture对象任务执行完成后执行该方法。BiConsumer函数式接口有两个入参没有返回值，这两个入参第一个是CompletableFuture任务的执行结果，第二个是异常信息。表示处理上一个任务的结果，如果有异常，则需要手动处理异常，与handle方法的区别在于，handle方法的BiFunction是有返回值的，而BiConsumer是没有返回值的。

以上方法都有一个带有Async的方法，带有Async的方法表示是异步执行的，会将该任务放到线程池中执行，同时该方法会有一个重载的方法，最后一个参数为Executor，表示异步执行可以指定线程池执行。为了方便进行控制，最好在使用CompletableFuture时手动指定我们的线程池。

2.4 异常处理

  1. public CompletableFuture<T> exceptionally(Function<Throwable, ? extends T> fn)

exceptionally是用来处理异常的，当任务抛出异常后，可以通过exceptionally来进行处理，也可以选择使用handle来进行处理，不过两者有些不同，hand是用来处理上一个任务的结果，如果有异常情况，就处理异常。而exceptionally可以放在CompletableFuture处理的最后，作为兜底逻辑来处理未知异常。

2.5 获取结果

  1. public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs)
  2. public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs)

allOf是需要入参中所有的CompletableFuture任务执行完成，才会进行下一步；

anyOf是入参中任何一个CompletableFuture任务执行完成都可以执行下一步。

  1. public T get() throws InterruptedException, ExecutionException
  2. public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException
  3. public T getNow(T valueIfAbsent)
  4. public T join()

get方法一个是不带超时时间的，一个是带有超时时间的。

getNow方法则是立即返回结果，如果还没有结果，则返回默认值，也就是该方法的入参。

join方法是不带超时时间的等待任务完成。

3、CompletableFuture原理

join方法同样表示获取结果，但是join与get方法有什么区别呢。

  1. public T join() {
  2.     Object r;
  3.     return reportJoin((r = result) == null ? waitingGet(false) : r);
  4. }
  6. public T get() throws InterruptedException, ExecutionException {
  7.     Object r;
  8.     return reportGet((r = result) == null ? waitingGet(true) : r);
  9. }
  11. public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
  12.         Object r;
  13.         long nanos = unit.toNanos(timeout);
  14.         return reportGet((r = result) == null ? timedGet(nanos) : r);
  15. }
  17. public T getNow(T valueIfAbsent) {
  18.         Object r;
  19.         return ((r = result) == null) ? valueIfAbsent : reportJoin(r);
  20. }

以上是CompletableFuture类中两个方法的代码，可以看到两个方法几乎一样。区别在于reportJoin/reportGet，waitingGet方法是一致的，只不过参数不一样，我们在看下reportGet与reportJoin方法。

  1. private static <T> T reportGet(Object r)
  2.         throws InterruptedException, ExecutionException {
  3.         if (r == null) // by convention below, null means interrupted
  4.             throw new InterruptedException();
  5.         if (r instanceof AltResult) {
  6.             Throwable x, cause;
  7.             if ((x = ((AltResult)r).ex) == null)
  8.                 return null;
  9.             if (x instanceof CancellationException)
  10.                 throw (CancellationException)x;
  11.             if ((x instanceof CompletionException) &&
  12.                 (cause = x.getCause()) != null)
  13.                 x = cause;
  14.             throw new ExecutionException(x);
  15.         }
  16.         @SuppressWarnings("unchecked") T t = (T) r;
  17.         return t;
  18.     }
  1. private static <T> T reportJoin(Object r) {
  2.         if (r instanceof AltResult) {
  3.             Throwable x;
  4.             if ((x = ((AltResult)r).ex) == null)
  5.                 return null;
  6.             if (x instanceof CancellationException)
  7.                 throw (CancellationException)x;
  8.             if (x instanceof CompletionException)
  9.                 throw (CompletionException)x;
  10.             throw new CompletionException(x);
  11.         }
  12.         @SuppressWarnings("unchecked") T t = (T) r;
  13.         return t;
  14.     }

可以看到这两个方法很相近，reportGet方法判断了r对象是否为空，并抛出了中断异常，而reportJoin方法没有判断，同时reportJoin抛出的都是运行时异常，所以join方法也是无需手动捕获异常的。

我们在看下waitingGet方法

  1. private Object waitingGet(boolean interruptible) {
  2.         Signaller q = null;
  3.         boolean queued = false;
  4.         int spins = -1;
  5.         Object r;
  6.         while ((r = result) == null) {
  7.             if (spins < 0)
  8.                 spins = SPINS;
  9.             else if (spins > 0) {
  10.                 if (ThreadLocalRandom.nextSecondarySeed() >= 0)
  11.                     --spins;
  12.             }
  13.             else if (q == null)
  14.                 q = new Signaller(interruptible, 0L, 0L);
  15.             else if (!queued)
  16.                 queued = tryPushStack(q);
  17.             else if (interruptible && q.interruptControl < 0) {
  18.                 q.thread = null;
  19.                 cleanStack();
  20.                 return null;
  21.             }
  22.             else if (q.thread != null && result == null) {
  23.                 try {
  24.                     ForkJoinPool.managedBlock(q);
  25.                 } catch (InterruptedException ie) {
  26.                     q.interruptControl = -1;
  27.                 }
  28.             }
  29.         }
  30.         if (q != null) {
  31.             q.thread = null;
  32.             if (q.interruptControl < 0) {
  33.                 if (interruptible)
  34.                     r = null; // report interruption
  35.                 else
  36.                     Thread.currentThread().interrupt();
  37.             }
  38.         }
  39.         postComplete();
  40.         return r;
  41.     }

该waitingGet方法是通过while的方式循环判断是否任务已经完成并产生结果，如果结果为空，则会一直在这里循环，这里需要注意的是在这里初始化了一下spins=-1，当第一次进入while循环的时候，spins是-1，这时会将spins赋值为一个常量，该常量为SPINS。

  1. private static final int SPINS = (Runtime.getRuntime().availableProcessors() > 1 ?
  2.                                       1 << 8 : 0);

这里判断可用CPU数是否大于1，如果大于1，则该常量为 1<< 8，也就是256，否则该常量为0。

第二次进入while循环的时候，spins是256大于0，这里做了减一的操作，下次进入while循环，如果还没有结果，依然是大于0继续做减一的操作，此处用来做短时间的自旋等待结果，只有当spins等于0，后续会进入正常流程判断。

我们在看下timedGet方法的源码

  1. private Object timedGet(long nanos) throws TimeoutException {
  2.         if (Thread.interrupted())
  3.             return null;
  4.         if (nanos <= 0L)
  5.             throw new TimeoutException();
  6.         long d = System.nanoTime() + nanos;
  7.         Signaller q = new Signaller(true, nanos, d == 0L ? 1L : d); // avoid 0
  8.         boolean queued = false;
  9.         Object r;
  10.         // We intentionally don't spin here (as waitingGet does) because
  11.         // the call to nanoTime() above acts much like a spin.
  12.         while ((r = result) == null) {
  13.             if (!queued)
  14.                 queued = tryPushStack(q);
  15.             else if (q.interruptControl < 0 || q.nanos <= 0L) {
  16.                 q.thread = null;
  17.                 cleanStack();
  18.                 if (q.interruptControl < 0)
  19.                     return null;
  20.                 throw new TimeoutException();
  21.             }
  22.             else if (q.thread != null && result == null) {
  23.                 try {
  24.                     ForkJoinPool.managedBlock(q);
  25.                 } catch (InterruptedException ie) {
  26.                     q.interruptControl = -1;
  27.                 }
  28.             }
  29.         }
  30.         if (q.interruptControl < 0)
  31.             r = null;
  32.         q.thread = null;
  33.         postComplete();
  34.         return r;
  35.     }

timedGet方法依然是通过while循环的方式来判断是否已经完成，timedGet方法入参为一个纳秒值，并通过该值计算出一个deadline截止时间，当while循环还未获取到任务结果且已经达到截止时间，则抛出一个TimeoutException异常。

4、CompletableFuture实现多线程任务

这里我们通过CompletableFuture来实现一个多线程处理异步任务的例子。

这里我们创建10个任务提交到我们指定的线程池中执行，并等待这10个任务全部执行完毕。

每个任务的执行流程为第一次先执行加法，第二次执行乘法，如果发生异常则返回默认值，当10个任务执行完成后依次打印每个任务的结果。

  1. public void demo() throws InterruptedException, ExecutionException, TimeoutException {
  2.         // 1、自定义线程池
  3.         ExecutorService executorService = new ThreadPoolExecutor(5, 10,
  4.                 60L, TimeUnit.SECONDS,
  5.                 new LinkedBlockingQueue<>(100));
  7.         // 2、集合保存future对象
  8.         List<CompletableFuture<Integer>> futures = new ArrayList<>(10);
  9.         for (int i = 0; i < 10; i++) {
  10.             int finalI = i;
  11.             CompletableFuture<Integer> future = CompletableFuture
  12.                     // 提交任务到指定线程池
  13.                     .supplyAsync(() -> this.addValue(finalI), executorService)
  14.                     // 第一个任务执行结果在此处进行处理
  15.                     .thenApplyAsync(k -> this.plusValue(finalI, k), executorService)
  16.                     // 任务执行异常时处理异常并返回默认值
  17.                     .exceptionally(e -> this.defaultValue(finalI, e));
  18.             // future对象添加到集合中
  19.             futures.add(future);
  20.         }
  22.         // 3、等待所有任务执行完成，此处最好加超时时间
  23.         CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).get(5, TimeUnit.MINUTES);
  24.         for (CompletableFuture<Integer> future : futures) {
  25.             Integer num = future.get();
  26.             System.out.println("任务执行结果为：" + num);
  27.         }
  28.         System.out.println("任务全部执行完成！");
  29.     }
  31.     private Integer addValue(Integer index) {
  32.         System.out.println("第" + index + "个任务第一次执行");
  33.         if (index == 3) {
  34.             int value = index / 0;
  35.         }
  36.         return index + 3;
  37.     }
  39.     private Integer plusValue(Integer index, Integer num) {
  40.         System.out.println("第" + index + "个任务第二次执行，上次执行结果：" + num);
  41.         return num * 10;
  42.     }
  44.     private Integer defaultValue(Integer index, Throwable e) {
  45.         System.out.println("第" + index + "个任务执行异常！" + e.getMessage());
  46.         e.printStackTrace();
  47.         return 10;
  48.     }

作者：京东物流 丁冬

来源：京东云开发者社区 自猿其说Tech

原文链接：https://www.cnblogs.com/jingdongkeji/p/17587006.html

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

Java热门文章

  • 【IDEA】Application context not configured for this file
  • [IDEA] chapter_reader - idea看小说插件 idea阅读插件 idea摸鱼插件
  • thrift常见异常及原因分析
  • Java 调用 PaddleDetection 模型
  • JVM 详解，大白话带你认识 JVM
  • spring-boot-2.0.3不一样系列之番外篇 - @Configuration、Condition与@Conditional
  • Java Comparator.comparing比较导致空指针异常的解决
  • Java使用EasyExcel进行单元格合并的问题详解
  • 死磕 java集合之ArrayBlockingQueue源码分析
  • Java基础之集合Set详解

    • Java推荐文章

  • 面试场景题：一次关于线程池使用场景的讨论。
  • 前端黑科技：使用 JavaScript 实现网页扫码功能
  • 聊一聊 Netty 数据搬运工 ByteBuf 体系的设计与实现
  • 简单设计一个JAVA并行处理工具类
  • 【杂谈】JPA乐观锁改悲观锁遇到的一些问题与思考
  • 认识netty的基本组件
  • Jenkins如何使用CrumbIssuer防御CSRF攻击
  • 如何平稳地从nacos迁移到r-nacos？
  • 线程池遇到父子任务，有大坑，要注意！
  • Java JVM——12. 垃圾回收理论概述

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号