CompletableFuture使用详解（全网看这一篇就行）

CompletableFuture是jdk8的新特性。CompletableFuture实现了CompletionStage接口和Future接口，前者是对后者的一个扩展，增加了异步会点、流式处理、多个Future组合处理的能力，使Java在处理多任务的协同工作时更加顺畅便利。

一、创建异步任务

1. supplyAsync

supplyAsync是创建带有返回值的异步任务。它有如下两个方法，一个是使用默认线程池（ForkJoinPool.commonPool()）的方法，一个是带有自定义线程池的重载方法

  1. // 带返回值异步请求，默认线程池
  2. public static  CompletableFuture supplyAsync(Supplier supplier)
  3.  
  4. // 带返回值的异步请求，可以自定义线程池
  5. public static  CompletableFuture supplyAsync(Supplier supplier, Executor executor)
  6.

测试代码：

  1. public static void main(String[] args) throws ExecutionException, InterruptedException {
  2.         CompletableFuture cf = CompletableFuture.supplyAsync(() -> {
  3.             System.out.println("do something....");
  4.             return "result";
  5.         });
  6.  
  7.         //等待任务执行完成
  8.         System.out.println("结果->" + cf.get());
  9. }
  10.  
  11.  
  12. public static void main(String[] args) throws ExecutionException, InterruptedException {
  13.         // 自定义线程池
  14.         ExecutorService executorService = Executors.newSingleThreadExecutor();
  15.         CompletableFuture cf = CompletableFuture.supplyAsync(() -> {
  16.             System.out.println("do something....");
  17.             return "result";
  18.         }, executorService);
  19.  
  20.         //等待子任务执行完成
  21.         System.out.println("结果->" + cf.get());
  22. }

 测试结果：

 2. runAsync

runAsync是创建没有返回值的异步任务。它有如下两个方法，一个是使用默认线程池（ForkJoinPool.commonPool()）的方法，一个是带有自定义线程池的重载方法

  1. // 不带返回值的异步请求，默认线程池
  2. public static CompletableFuture runAsync(Runnable runnable)
  3.  
  4. // 不带返回值的异步请求，可以自定义线程池
  5. public static CompletableFuture runAsync(Runnable runnable, Executor executor)

测试代码：

  1. public static void main(String[] args) throws ExecutionException, InterruptedException {
  2.         CompletableFuture<Void> cf = CompletableFuture.runAsync(() -> {
  3.             System.out.println("do something....");
  4.         });
  5.  
  6.         //等待任务执行完成
  7.         System.out.println("结果->" + cf.get());
  8. }
  9.  
  10.  
  11. public static void main(String[] args) throws ExecutionException, InterruptedException {
  12.         // 自定义线程池
  13.         ExecutorService executorService = Executors.newSingleThreadExecutor();
  14.         CompletableFuture<Void> cf = CompletableFuture.runAsync(() -> {
  15.             System.out.println("do something....");
  16.         }, executorService);
  17.  
  18.         //等待任务执行完成
  19.         System.out.println("结果->" + cf.get());
  20. }

测试结果：

3.获取任务结果的方法

  1. // 如果完成则返回结果，否则就抛出具体的异常
  2. public T get() throws InterruptedException, ExecutionException 
  3.  
  4. // 最大时间等待返回结果，否则就抛出具体异常
  5. public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException
  6.  
  7. // 完成时返回结果值，否则抛出unchecked异常。为了更好地符合通用函数形式的使用，如果完成此 CompletableFuture所涉及的计算引发异常，则此方法将引发unchecked异常并将底层异常作为其原因
  8. public T join()
  9.  
  10. // 如果完成则返回结果值（或抛出任何遇到的异常），否则返回给定的 valueIfAbsent。
  11. public T getNow(T valueIfAbsent)
  12.  
  13. // 如果任务没有完成，返回的值设置为给定值
  14. public boolean complete(T value)
  15.  
  16. // 如果任务没有完成，就抛出给定异常
  17. public boolean completeExceptionally(Throwable ex) 
  18.

 二、异步回调处理

1.thenApply和thenApplyAsync

 thenApply 表示某个任务执行完成后执行的动作，即回调方法，会将该任务的执行结果即方法返回值作为入参传递到回调方法中，带有返回值。

测试代码：

  1. public static void main(String[] args) throws ExecutionException, InterruptedException {
  2.         CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
  3.             System.out.println(Thread.currentThread() + " cf1 do something....");
  4.             return 1;
  5.         });
  6.  
  7.         CompletableFuture cf2 = cf1.thenApplyAsync((result) -> {
  8.             System.out.println(Thread.currentThread() + " cf2 do something....");
  9.             result += 2;
  10.             return result;
  11.         });
  12.         //等待任务1执行完成
  13.         System.out.println("cf1结果->" + cf1.get());
  14.         //等待任务2执行完成
  15.         System.out.println("cf2结果->" + cf2.get());
  16. }
  17.  
  18. public static void main(String[] args) throws ExecutionException, InterruptedException {
  19.         CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
  20.             System.out.println(Thread.currentThread() + " cf1 do something....");
  21.             return 1;
  22.         });
  23.  
  24.         CompletableFuture cf2 = cf1.thenApply((result) -> {
  25.             System.out.println(Thread.currentThread() + " cf2 do something....");
  26.             result += 2;
  27.             return result;
  28.         });
  29.         //等待任务1执行完成
  30.         System.out.println("cf1结果->" + cf1.get());
  31.         //等待任务2执行完成
  32.         System.out.println("cf2结果->" + cf2.get());
  33. }

测试结果：

   

从上面代码和测试结果我们发现thenApply和thenApplyAsync区别在于，使用thenApply方法时子任务与父任务使用的是同一个线程，而thenApplyAsync在子任务中是另起一个线程执行任务，并且thenApplyAsync可以自定义线程池，默认的使用ForkJoinPool.commonPool()线程池。

2.thenAccept和thenAcceptAsync

 thenAccep表示某个任务执行完成后执行的动作，即回调方法，会将该任务的执行结果即方法返回值作为入参传递到回调方法中，无返回值。

测试代码

  1. public static void main(String[] args) throws ExecutionException, InterruptedException {
  2.         CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
  3.             System.out.println(Thread.currentThread() + " cf1 do something....");
  4.             return 1;
  5.         });
  6.  
  7.         CompletableFuture cf2 = cf1.thenAccept((result) -> {
  8.             System.out.println(Thread.currentThread() + " cf2 do something....");
  9.         });
  10.  
  11.         //等待任务1执行完成
  12.         System.out.println("cf1结果->" + cf1.get());
  13.         //等待任务2执行完成
  14.         System.out.println("cf2结果->" + cf2.get());
  15. }
  16.  
  17.  
  18. public static void main(String[] args) throws ExecutionException, InterruptedException {
  19.         CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
  20.             System.out.println(Thread.currentThread() + " cf1 do something....");
  21.             return 1;
  22.         });
  23.  
  24.         CompletableFuture cf2 = cf1.thenAcceptAsync((result) -> {
  25.             System.out.println(Thread.currentThread() + " cf2 do something....");
  26.         });
  27.  
  28.         //等待任务1执行完成
  29.         System.out.println("cf1结果->" + cf1.get());
  30.         //等待任务2执行完成
  31.         System.out.println("cf2结果->" + cf2.get());
  32. }

测试结果：

 从上面代码和测试结果我们发现thenAccep和thenAccepAsync区别在于，使用thenAccep方法时子任务与父任务使用的是同一个线程，而thenAccepAsync在子任务中可能是另起一个线程执行任务，并且thenAccepAsync可以自定义线程池，默认的使用ForkJoinPool.commonPool()线程池。

2.thenRun和thenRunAsync

 thenRun表示某个任务执行完成后执行的动作，即回调方法，无入参，无返回值。

测试代码：

  1. public static void main(String[] args) throws ExecutionException, InterruptedException {
  2.         CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
  3.             System.out.println(Thread.currentThread() + " cf1 do something....");
  4.             return 1;
  5.         });
  6.  
  7.         CompletableFuture cf2 = cf1.thenRun(() -> {
  8.             System.out.println(Thread.currentThread() + " cf2 do something....");
  9.         });
  10.  
  11.         //等待任务1执行完成
  12.         System.out.println("cf1结果->" + cf1.get());
  13.         //等待任务2执行完成
  14.         System.out.println("cf2结果->" + cf2.get());
  15. }
  16.  
  17. public static void main(String[] args) throws ExecutionException, InterruptedException {
  18.         CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
  19.             System.out.println(Thread.currentThread() + " cf1 do something....");
  20.             return 1;
  21.         });
  22.  
  23.         CompletableFuture cf2 = cf1.thenRunAsync(() -> {
  24.             System.out.println(Thread.currentThread() + " cf2 do something....");
  25.         });
  26.  
  27.         //等待任务1执行完成
  28.         System.out.println("cf1结果->" + cf1.get());
  29.         //等待任务2执行完成
  30.         System.out.println("cf2结果->" + cf2.get());
  31. }

 测试结果：

从上面代码和测试结果我们发现thenRun和thenRunAsync区别在于，使用thenRun方法时子任务与父任务使用的是同一个线程，而thenRunAsync在子任务中可能是另起一个线程执行任务，并且thenRunAsync可以自定义线程池，默认的使用ForkJoinPool.commonPool()线程池。

3.whenComplete和whenCompleteAsync

 whenComplete是当某个任务执行完成后执行的回调方法，会将执行结果或者执行期间抛出的异常传递给回调方法，如果是正常执行则异常为null，回调方法对应的CompletableFuture的result和该任务一致，如果该任务正常执行，则get方法返回执行结果，如果是执行异常，则get方法抛出异常。

测试代码：

  1.  public static void main(String[] args) throws ExecutionException, InterruptedException {
  2.         CompletableFuture<Integer> cf1 = CompletableFuture.supplyAsync(() -> {
  3.             System.out.println(Thread.currentThread() + " cf1 do something....");
  4.             int a = 1/0;
  5.             return 1;
  6.         });
  7.  
  8.         CompletableFuture<Integer> cf2 = cf1.whenComplete((result, e) -> {
  9.             System.out.println("上个任务结果：" + result);
  10.             System.out.println("上个任务抛出异常：" + e);
  11.             System.out.println(Thread.currentThread() + " cf2 do something....");
  12.         });
  13.  
  14. //        //等待任务1执行完成
  15. //        System.out.println("cf1结果->" + cf1.get());
  16. //        //等待任务2执行完成
  17.         System.out.println("cf2结果->" + cf2.get());
  18.     }

测试结果：

 

 whenCompleteAsync和whenComplete区别也是whenCompleteAsync可能会另起一个线程执行任务，并且thenRunAsync可以自定义线程池，默认的使用ForkJoinPool.commonPool()线程池。

4.handle和handleAsync

 跟whenComplete基本一致，区别在于handle的回调方法有返回值。

测试代码：

  1. public static void main(String[] args) throws ExecutionException, InterruptedException {
  2.         CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
  3.             System.out.println(Thread.currentThread() + " cf1 do something....");
  4.             // int a = 1/0;
  5.             return 1;
  6.         });
  7.  
  8.         CompletableFuture cf2 = cf1.handle((result, e) -> {
  9.             System.out.println(Thread.currentThread() + " cf2 do something....");
  10.             System.out.println("上个任务结果：" + result);
  11.             System.out.println("上个任务抛出异常：" + e);
  12.             return result+2;
  13.         });
  14.  
  15.         //等待任务2执行完成
  16.         System.out.println("cf2结果->" + cf2.get());
  17. }

测试结果 ：

三、多任务组合处理 

1.thenCombine、thenAcceptBoth 和runAfterBoth

这三个方法都是将两个CompletableFuture组合起来处理，只有两个任务都正常完成时，才进行下阶段任务。

区别：thenCombine会将两个任务的执行结果作为所提供函数的参数，且该方法有返回值；thenAcceptBoth同样将两个任务的执行结果作为方法入参，但是无返回值；runAfterBoth没有入参，也没有返回值。注意两个任务中只要有一个执行异常，则将该异常信息作为指定任务的执行结果。

测试代码：

  1. public static void main(String[] args) throws ExecutionException, InterruptedException {
  2.         CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
  3.             System.out.println(Thread.currentThread() + " cf1 do something....");
  4.             return 1;
  5.         });
  6.  
  7.         CompletableFuture cf2 = CompletableFuture.supplyAsync(() -> {
  8.             System.out.println(Thread.currentThread() + " cf2 do something....");
  9.             return 2;
  10.         });
  11.  
  12.         CompletableFuture cf3 = cf1.thenCombine(cf2, (a, b) -> {
  13.             System.out.println(Thread.currentThread() + " cf3 do something....");
  14.             return a + b;
  15.         });
  16.  
  17.         System.out.println("cf3结果->" + cf3.get());
  18. }
  19.  
  20.  public static void main(String[] args) throws ExecutionException, InterruptedException {
  21.         CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
  22.             System.out.println(Thread.currentThread() + " cf1 do something....");
  23.             return 1;
  24.         });
  25.  
  26.         CompletableFuture cf2 = CompletableFuture.supplyAsync(() -> {
  27.             System.out.println(Thread.currentThread() + " cf2 do something....");
  28.             return 2;
  29.         });
  30.         
  31.         CompletableFuture cf3 = cf1.thenAcceptBoth(cf2, (a, b) -> {
  32.             System.out.println(Thread.currentThread() + " cf3 do something....");
  33.             System.out.println(a + b);
  34.         });
  35.  
  36.         System.out.println("cf3结果->" + cf3.get());
  37. }
  38.  
  39. public static void main(String[] args) throws ExecutionException, InterruptedException {
  40.         CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
  41.             System.out.println(Thread.currentThread() + " cf1 do something....");
  42.             return 1;
  43.         });
  44.  
  45.         CompletableFuture cf2 = CompletableFuture.supplyAsync(() -> {
  46.             System.out.println(Thread.currentThread() + " cf2 do something....");
  47.             return 2;
  48.         });
  49.  
  50.         CompletableFuture cf3 = cf1.runAfterBoth(cf2, () -> {
  51.             System.out.println(Thread.currentThread() + " cf3 do something....");
  52.         });
  53.  
  54.         System.out.println("cf3结果->" + cf3.get());
  55. }

测试结果：

 

 2.applyToEither、acceptEither和runAfterEither

这三个方法和上面一样也是将两个CompletableFuture组合起来处理，当有一个任务正常完成时，就会进行下阶段任务。

区别：applyToEither会将已经完成任务的执行结果作为所提供函数的参数，且该方法有返回值；acceptEither同样将已经完成任务的执行结果作为方法入参，但是无返回值；runAfterEither没有入参，也没有返回值。

测试代码：

  1. public static void main(String[] args) throws ExecutionException, InterruptedException {
  2.         CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
  3.             try {
  4.                 System.out.println(Thread.currentThread() + " cf1 do something....");
  5.                 Thread.sleep(2000);
  6.             } catch (InterruptedException e) {
  7.                 e.printStackTrace();
  8.             }
  9.             return "cf1 任务完成";
  10.         });
  11.  
  12.         CompletableFuture cf2 = CompletableFuture.supplyAsync(() -> {
  13.             try {
  14.                 System.out.println(Thread.currentThread() + " cf2 do something....");
  15.                 Thread.sleep(5000);
  16.             } catch (InterruptedException e) {
  17.                 e.printStackTrace();
  18.             }
  19.             return "cf2 任务完成";
  20.         });
  21.  
  22.         CompletableFuture cf3 = cf1.applyToEither(cf2, (result) -> {
  23.             System.out.println("接收到" + result);
  24.             System.out.println(Thread.currentThread() + " cf3 do something....");
  25.             return "cf3 任务完成";
  26.         });
  27.  
  28.         System.out.println("cf3结果->" + cf3.get());
  29. }
  30.  
  31.  
  32. public static void main(String[] args) throws ExecutionException, InterruptedException {
  33.         CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
  34.             try {
  35.                 System.out.println(Thread.currentThread() + " cf1 do something....");
  36.                 Thread.sleep(2000);
  37.             } catch (InterruptedException e) {
  38.                 e.printStackTrace();
  39.             }
  40.             return "cf1 任务完成";
  41.         });
  42.  
  43.         CompletableFuture cf2 = CompletableFuture.supplyAsync(() -> {
  44.             try {
  45.                 System.out.println(Thread.currentThread() + " cf2 do something....");
  46.                 Thread.sleep(5000);
  47.             } catch (InterruptedException e) {
  48.                 e.printStackTrace();
  49.             }
  50.             return "cf2 任务完成";
  51.         });
  52.  
  53.         CompletableFuture cf3 = cf1.acceptEither(cf2, (result) -> {
  54.             System.out.println("接收到" + result);
  55.             System.out.println(Thread.currentThread() + " cf3 do something....");
  56.         });
  57.  
  58.         System.out.println("cf3结果->" + cf3.get());
  59. }
  60.  
  61. public static void main(String[] args) throws ExecutionException, InterruptedException {
  62.         CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
  63.             try {
  64.                 System.out.println(Thread.currentThread() + " cf1 do something....");
  65.                 Thread.sleep(2000);
  66.             } catch (InterruptedException e) {
  67.                 e.printStackTrace();
  68.             }
  69.             System.out.println("cf1 任务完成");
  70.             return "cf1 任务完成";
  71.         });
  72.  
  73.         CompletableFuture cf2 = CompletableFuture.supplyAsync(() -> {
  74.             try {
  75.                 System.out.println(Thread.currentThread() + " cf2 do something....");
  76.                 Thread.sleep(5000);
  77.             } catch (InterruptedException e) {
  78.                 e.printStackTrace();
  79.             }
  80.             System.out.println("cf2 任务完成");
  81.             return "cf2 任务完成";
  82.         });
  83.  
  84.         CompletableFuture cf3 = cf1.runAfterEither(cf2, () -> {
  85.             System.out.println(Thread.currentThread() + " cf3 do something....");
  86.             System.out.println("cf3 任务完成");
  87.         });
  88.  
  89.         System.out.println("cf3结果->" + cf3.get());
  90. }

测试结果： 

从上面可以看出cf1任务完成需要2秒，cf2任务完成需要5秒，使用applyToEither组合两个任务时，只要有其中一个任务完成时，就会执行cf3任务，显然cf1任务先完成了并且将自己任务的结果传值给了cf3任务，cf3任务中打印了接收到cf1任务完成，接着完成自己的任务，并返回cf3任务完成；acceptEither和runAfterEither类似，acceptEither会将cf1任务的结果作为cf3任务的入参，但cf3任务完成时并无返回值；runAfterEither不会将cf1任务的结果作为cf3任务的入参，它是没有任务入参，执行完自己的任务后也并无返回值。

3.allOf / anyOf 

allOf：CompletableFuture是多个任务都执行完成后才会执行，只有有一个任务执行异常，则返回的CompletableFuture执行get方法时会抛出异常，如果都是正常执行，则get返回null。

anyOf ：CompletableFuture是多个任务只要有一个任务执行完成，则返回的CompletableFuture执行get方法时会抛出异常，如果都是正常执行，则get返回执行完成任务的结果。

测试代码：

  1. public static void main(String[] args) throws ExecutionException, InterruptedException {
  2.         CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
  3.             try {
  4.                 System.out.println(Thread.currentThread() + " cf1 do something....");
  5.                 Thread.sleep(2000);
  6.             } catch (InterruptedException e) {
  7.                 e.printStackTrace();
  8.             }
  9.             System.out.println("cf1 任务完成");
  10.             return "cf1 任务完成";
  11.         });
  12.  
  13.         CompletableFuture cf2 = CompletableFuture.supplyAsync(() -> {
  14.             try {
  15.                 System.out.println(Thread.currentThread() + " cf2 do something....");
  16.                 int a = 1/0;
  17.                 Thread.sleep(5000);
  18.             } catch (InterruptedException e) {
  19.                 e.printStackTrace();
  20.             }
  21.             System.out.println("cf2 任务完成");
  22.             return "cf2 任务完成";
  23.         });
  24.  
  25.         CompletableFuture cf3 = CompletableFuture.supplyAsync(() -> {
  26.             try {
  27.                 System.out.println(Thread.currentThread() + " cf2 do something....");
  28.                 Thread.sleep(3000);
  29.             } catch (InterruptedException e) {
  30.                 e.printStackTrace();
  31.             }
  32.             System.out.println("cf3 任务完成");
  33.             return "cf3 任务完成";
  34.         });
  35.  
  36.         CompletableFuture cfAll = CompletableFuture.allOf(cf1, cf2, cf3);
  37.         System.out.println("cfAll结果->" + cfAll.get());
  38. }
  39.  
  40.  
  41. public static void main(String[] args) throws ExecutionException, InterruptedException {
  42.         CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
  43.             try {
  44.                 System.out.println(Thread.currentThread() + " cf1 do something....");
  45.                 Thread.sleep(2000);
  46.             } catch (InterruptedException e) {
  47.                 e.printStackTrace();
  48.             }
  49.             System.out.println("cf1 任务完成");
  50.             return "cf1 任务完成";
  51.         });
  52.  
  53.         CompletableFuture cf2 = CompletableFuture.supplyAsync(() -> {
  54.             try {
  55.                 System.out.println(Thread.currentThread() + " cf2 do something....");
  56.                 Thread.sleep(5000);
  57.             } catch (InterruptedException e) {
  58.                 e.printStackTrace();
  59.             }
  60.             System.out.println("cf2 任务完成");
  61.             return "cf2 任务完成";
  62.         });
  63.  
  64.         CompletableFuture cf3 = CompletableFuture.supplyAsync(() -> {
  65.             try {
  66.                 System.out.println(Thread.currentThread() + " cf2 do something....");
  67.                 Thread.sleep(3000);
  68.             } catch (InterruptedException e) {
  69.                 e.printStackTrace();
  70.             }
  71.             System.out.println("cf3 任务完成");
  72.             return "cf3 任务完成";
  73.         });
  74.  
  75.         CompletableFuture cfAll = CompletableFuture.anyOf(cf1, cf2, cf3);
  76.         System.out.println("cfAll结果->" + cfAll.get());
  77. }
    78.  

    测试结果：
     

     

     
    
                
    
        
        
    文章知识点与官方知识档案匹配，可进一步学习相关知识
    Java技能树Java异步任务Future与CompletableFuture140178 人正在系统学习中
    
    
                
                
    
                    注：本文转载自blog.csdn.net的代码搬运工阿新的文章"https://blog.csdn.net/zsx_xiaoxin/article/details/123898171"。版权归原作者所有，此博客不拥有其著作权，亦不承担相应法律责任。如有侵权，请联系我们删除。
                
    
                
    
                    
    
                        
    
                            
    
                                
                                
                                
                                
                                    
                                
                                
                                
                                
                                
                                
                                
                                
                                
                                
                                
    
                                    复制链接
                                    
                                
    
                                
    
                                    
                                    
    
                                        
    
                                            
    
                                        
    
                                    
    
                                
    
                            
    
                        
    
                    
    
                
    

                
    
                    
    




                        
                    
    
                
    
                
    
                
            
    


            
            
    
                
    相关推荐
    
                
    
                    
    
                    
    
                
    
            
    
            
    
                
    
                    
    
                        
    
							
								发表评论
                        
    
                        
                    
    
                    
    
                        
    
                            
    
								
                            
    
                            
    
                                
    
                                    
    
                                        
                                    
    
                                    
    
                                        
                                    
    
                                
    
                            
    
                            
    
                                
                            
    
                        
    
                        
    评论记录：
    
                        
                        
                        
                        
                        
                        
    未查询到任何数据！