CompletableFuture，默认依靠fork/join框架启动新的线程实现异步与并发的。
它提供了函数式编程的能力，可以通过回调函数的方式处理返回结果，并且提供了转换和组合CompletableFuture的方法。

0.测试用例

创建一个DeptService，模拟根据Id获取部门的方法getById(Integer id)

  1. public class DeptService {
  2.     public Dept getById(Integer id)  {
  3.         System.out.println("线程：" + Thread.currentThread().getName() + " getById(" + id + ")");
  5.         if (id == 1){
  6.             return new Dept(1, "研发一部");
  7.         } else if (id == 2){
  8.             return new Dept(2, "研发二部");
  9.         } else {
  10.             throw null;
  11.         }
  12.     }
  13. }

创建一个UserService ，模拟getById()和save()这2个方法

  1. public class UserService {
  2.     //根据Id获取User
  3.     public User getById(Integer id) throws Exception {
  4.         System.out.println("线程：" + Thread.currentThread().getName() + " getById(" + id + ")");
  6.         if (id == 1){
  7.             return new User(1, "meizuna", 20);
  8.         } else if (id == 2){
  9.             return new User(2, "misa", 18);
  10.         } else {
  11.             throw new Exception("未能找到人员");
  12.         }
  13.     }
  15.     //保存User
  16.     public User save(User user){
  17.         System.out.println("线程：" + Thread.currentThread().getName() + " save()," + user.toString());
  18.         return user;
  19.     }
  20. }

1. supplyAsync、runAsync

CompletableFuture创建线程有2种方式：supplyAsync（有返回值）和：runAsync（无返回值）

1.1 supplyAsync（有返回值）

supplyAsync有2种，第二个需要多传1个线程池的实现：

  1. public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
  3. public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)
  1. public class Thread01_SupplyAsync {
  2.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  3.         DeptService deptService = new DeptService();
  5.         CompletableFuture<Dept> deptCompletableFuture = CompletableFuture.supplyAsync(() -> {
  6.             Dept dept = deptService.getById(1);
  7.             return dept;
  8.         });
  10.         System.out.println("线程：" + Thread.currentThread().getName() +
  11.                 " 结果：" + deptCompletableFuture.get());
  12.     }
  13. }
  15. //线程：ForkJoinPool.commonPool-worker-1 getById(1)
  16. //线程：main 结果：Dept{id=1, name='研发一部'}

1.2 runAsync（无返回值）

unAsync适用无返回值的情况

  1. public static CompletableFuture<Void> runAsync(Runnable runnable)
  3. public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)

测试代码

  1. public class Thread02_RunAsync {
  2.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  3.         DeptService deptService = new DeptService();
  5.         CompletableFuture<Void> voidCompletableFuture = CompletableFuture.runAsync(() -> {
  6.             deptService.getById(1);
  7.         });
  9.         System.out.println("线程：" + Thread.currentThread().getName() +
  10.                 " 结果：" + voidCompletableFuture.get());
  11.     }
  12. }
  14. //线程：ForkJoinPool.commonPool-worker-1 getById(1)
  15. //线程：main 结果：null

2. thenApply、thenAccept、thenRun

2.1 thenApply 转换结果

thenApply是同步的
thenApplyAsync是异步的

  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)
  1. Function<? super T, ? extends U>
  2. T：上一个任务返回结果的类型
  3. U：当前任务的返回值类型

测试用例

  1. public class Thread03_SupplyAsync_ThenApply {
  2.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  4.         DeptService deptService = new DeptService();
  5.         UserService userService = new UserService();
  7.         User user = new User(1, "meizuna", 20);
  9.         CompletableFuture<User> userCompletableFuture = CompletableFuture.supplyAsync(() -> {
  10.             Dept dept = deptService.getById(1);
  11.             return dept;
  12.         }).thenApplyAsync(dept -> {
  13.           //注意这里用到了上个线程的返回值dept
  14.           user.setDeptId(dept.getId());
  15.           user.setDeptName(dept.getName());
  16.           return userService.save(user);
  17.         });
  19.         System.out.println("线程：" + Thread.currentThread().getName() +
  20.                 " 结果：" + userCompletableFuture.get().toString());
  21.     }
  22. }

2.2 thenAccept 消费结果

thenAccept 同 thenApply 接收上一个任务的返回值作为参数，但是回调方法无返回值

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

测试用例

  1. public class Thread04_SupplyAsync_ThenAccept {
  2.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  3.         DeptService deptService = new DeptService();
  5.         CompletableFuture<Void> voidCompletableFuture = CompletableFuture.supplyAsync(() -> {
  6.             Dept dept = deptService.getById(1);
  7.             return dept;
  8.         })
  9.                 .thenAcceptAsync(dept -> {
  10.                     //注意这里用到了上个线程的返回值dept
  11.                     System.out.println("线程：" + Thread.currentThread().getName() +
  12.                             "假设把dept作为日志记录发给Kafka: " + dept.toString());
  13.                     //thenAccept是没有返回值的
  14.                 });
  15.         System.out.println("线程：" + Thread.currentThread().getName() +
  16.                 " 结果：" + voidCompletableFuture.get());
  17.     }
  18. }

2.3 thenRun 任务完成后触发的回调

thenRun 是上一个任务完成后触发的回调，没有入参和返回值

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

测试用例

  1. public class Thread05_SupplyAsync_ThenRun {
  2.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  3.         DeptService deptService = new DeptService();
  5.         CompletableFuture<Void> voidCompletableFuture = CompletableFuture.supplyAsync(() -> {
  6.             Dept dept = deptService.getById(1);
  7.             return dept;
  8.         })
  9.                 .thenRun(() -> {//注意没有入参
  10.                     System.out.println("线程：" + Thread.currentThread().getName() + " do something");
  11.                     //thenRun注意没有入参，也没有返回值
  12.                 });
  14.         System.out.println("线程：" + Thread.currentThread().getName() +
  15.                 " 结果：" + voidCompletableFuture.get());
  16.     }
  17. }

3. exceptionally

在用CompletableFuture编写多线程时，如果需要处理异常，可以用exceptionally，它的作用相当于catch
exceptionally的特点：
当出现异常时，会触发回调方法exceptionally
exceptionally中可指定默认返回结果，如果出现异常，则返回默认的返回结果

测试用例

  1. public class Thread01_Exceptionally {
  2.     public static void main(String[] args) throws InterruptedException, ExecutionException {
  4.         CompletableFuture<Double> future = CompletableFuture.supplyAsync(() -> {
  5.             if (Math.random() < 0.5) {
  6.                 throw new RuntimeException("抛出异常");
  7.             }
  8.             System.out.println("正常结束");
  9.             return 1.1;
  10.         })
  11.             .thenApply(result -> {
  12.             System.out.println("thenApply接收到的参数 = " + result);
  13.             return result;
  14.         })
  15.             .exceptionally(new Function<Throwable, Double>() {
  17.             @Override
  18.             public Double apply(Throwable throwable) {
  19.                 System.out.println("异常：" + throwable.getMessage());
  20.                 return 0.0;
  21.             }
  22.         });
  24.         System.out.println("最终返回的结果 = " + future.get());
  26.     }
  27. }

4. whenComplete

当CompletableFuture的任务不论是正常完成还是出现异常都会调用whenComplete
正常完成：whenComplete返回结果和上级任务一致，异常为null；
出现异常：whenComplete返回结果为null，异常为上级任务的异常；
即调用get()时，正常完成时就获取到结果，出现异常时就会抛出异常，需要处理该异常

  1. public class Thread02_WhenComplete {
  2.     public static void main(String[] args) throws InterruptedException, ExecutionException {
  3.         CompletableFuture<Double> future = CompletableFuture.supplyAsync(() -> {
  4.             if (Math.random() < 0.5) {
  5.                 throw new RuntimeException("出错了");
  6.             }
  7.             System.out.println("正常结束");
  8.             return 0.11;
  9.         }).whenComplete(new BiConsumer<Double, Throwable>() {
  10.             @Override
  11.             public void accept(Double aDouble, Throwable throwable) {
  13.                 if (aDouble == null) {
  14.                     System.out.println("whenComplete aDouble is null");
  15.                 } else {
  16.                     System.out.println("whenComplete aDouble is " + aDouble);
  17.                 }
  19.                 if (throwable == null) {
  20.                     System.out.println("whenComplete throwable is null");
  21.                 } else {
  22.                     System.out.println("whenComplete throwable is " + throwable.getMessage());
  23.                 }
  24.             }
  25.         });
  27.         System.out.println("最终返回的结果 = " + future.get());
  28.     }
  29. }

4.1 whenComplete + exceptionally

  1. public class Thread03_WhenComplete_Exceptionally {
  3.     public static void main(String[] args) throws InterruptedException, ExecutionException {
  5.         CompletableFuture<Double> future = CompletableFuture.supplyAsync(() -> {
  7.             if (Math.random() < 0.5) {
  8.                 throw new RuntimeException("出错了");
  9.             }
  10.             System.out.println("正常结束");
  11.             return 0.11;
  13.         }).whenComplete(new BiConsumer<Double, Throwable>() {
  14.             @Override
  15.             public void accept(Double aDouble, Throwable throwable) {
  17.                 if (aDouble == null){
  18.                     System.out.println("whenComplete aDouble is null");
  19.                 } else {
  20.                     System.out.println("whenComplete aDouble is " + aDouble);
  21.                 }
  23.                 if (throwable == null){
  24.                     System.out.println("whenComplete throwable is null");
  25.                 } else {
  26.                     System.out.println("whenComplete throwable is " + throwable.getMessage());
  27.                 }
  28.             }
  29.         }).exceptionally(new Function<Throwable, Double>() {
  30.             @Override
  31.             public Double apply(Throwable throwable) {
  32.                 System.out.println("exceptionally中异常：" + throwable.getMessage());
  33.                 return 0.0;
  34.             }
  35.         });
  37.         System.out.println("最终返回的结果 = " + future.get());
  38.     }
  39. }

5. handle()

不论正常返回还是出异常都会进入handle，类似whenComplete

handle()一般接收new BiFunction()
T：就是任务传入的对象类型
Throwable：就是任务传入的异常
R：就是handle自己返回的对象类型

5.1 handle和thenApply的区别

thenApply：任务出现异常就不会进入thenApply
handle：任务出现异常也会进入handle，可对异常处理

5.2 handle和whenComplete的区别

handle对传入值进行转换，并产生自己的返回结果，T -> R
whenComplete的返回值和上级任务传入的结果一致，不能对其转换

测试用例

  1. public class Thread04_Handle {
  2.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  3.         DeptService deptService = new DeptService();
  4.         UserService userService = new UserService();
  6.         CompletableFuture<User> future = CompletableFuture
  7.                 .supplyAsync(() -> {
  8.                             //int a = 1 / 0;//如果出现异常，那么thenApply则不执行
  9.                             return deptService.getById(1);
  10.                         }
  11.                 )
  12.                 .handle(new BiFunction<Dept, Throwable, User>() {
  13.                             @Override
  14.                             public User apply(Dept dept, Throwable throwable) {
  15.                                 if (throwable != null){
  16.                                     System.out.println(throwable.getMessage());
  17.                                     return null;
  18.                                 } else {
  19.                                     User user = new User(1, "winter", 32, dept.getId(), dept.getName());
  20.                                     return userService.save(user);
  21.                                 }
  22.                             }
  23.                         }
  24.                 );
  26.         System.out.println("线程：" + Thread.currentThread().getName() +
  27.                 " 结果：" + future.get());
  28.     }
  29. }
  31. //正常
  32. //线程：ForkJoinPool.commonPool-worker-1 Dept.getById(1)
  33. //线程：main User.save(),User(id=1, name=winter, age=32, deptId=1, deptName=研发一部)
  34. //线程：main 结果：User(id=1, name=winter, age=32, deptId=1, deptName=研发一部)
  36. //异常
  37. //java.lang.ArithmeticException: / by zero
  38. //线程：main 结果：null

6. thenCompose、thenCombine

6.1 thenCompose

thenCompose 可以用于组合多个CompletableFuture，将前一个任务的返回结果作为下一个任务的参数，它们之间存在着先后顺序。
thenCompose方法会在某个任务执行完成后，将该任务的执行结果作为方法入参然后执行指定的方法，该方法会返回一个新的CompletableFuture实例。

  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) ;

6.2 thenApply和thenCompose的区别

thenApply转换的是泛型中的类型，是同一个CompletableFuture，相当于将CompletableFuture 转换成CompletableFuture
thenCompose用来连接两个CompletableFuture，是生成一个新的CompletableFuture
测试用例

  1. public class Thread06_SupplyAsync_ThenCompose {
  2.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  4.         DeptService deptService = new DeptService();
  5.         UserService userService = new UserService();
  7.         User user = new User(1, "meizuna", 20);
  9.         CompletableFuture<User> userCompletableFuture = CompletableFuture.supplyAsync(() -> {
  11.             Dept dept = deptService.getById(1);
  12.             return dept;
  13.         })
  14.                 .thenCompose(dept -> CompletableFuture.supplyAsync(() -> {
  15.                     //注意这里用到了上个线程的返回值dept
  16.                     user.setDeptId(dept.getId());
  17.                     user.setDeptName(dept.getName());
  19.                     return userService.save(user);
  20.                 }));
  22.         System.out.println("线程：" + Thread.currentThread().getName() +
  23.                 " 结果：" + userCompletableFuture.get().toString());
  24.     }
  25. }

6.3 thenCombine

thenCombine会在两个任务都执行完成后，把两个任务的结果合并
注意：
两个任务中只要有一个执行异常，则将该异常信息作为指定任务的执行结果
两个任务是并行执行的，它们之间并没有先后依赖顺序
测试用例

  1. public class Thread10_ThenCombine {
  3.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  4.         DeptService deptService = new DeptService();
  5.         UserService userService = new UserService();
  7.         //第1个任务：获取id=1的部门
  8.         CompletableFuture<Dept> deptFuture = CompletableFuture
  9.                 .supplyAsync(() -> {
  10.                             return deptService.getById(1);
  11.                         }
  12.                 );
  14.         //第2个任务：获取id=1的人员
  15.         CompletableFuture<User> userFuture = CompletableFuture
  16.                 .supplyAsync(() -> {
  17.                     try {
  18.                         //int a = 1 / 0;//出了异常就报错
  19.                         return userService.getById(1);
  20.                     } catch (Exception e) {
  21.                         e.printStackTrace();
  22.                     }
  23.                     return null;
  24.                 });
  26.         //将上面2个任务的返回结果dept和user合并，返回新的user
  27.         CompletableFuture<User> resultFuture = deptFuture
  28.                 .thenCombine(userFuture,
  29.                         new BiFunction<Dept, User, User>() {
  30.                             @Override
  31.                             public User apply(Dept dept, User user) {
  32.                                 user.setDeptId(dept.getId());
  33.                                 user.setDeptName(dept.getName());
  34.                                 return userService.save(user);
  35.                             }
  36.                         }
  37.                 );
  39.         System.out.println("线程：" + Thread.currentThread().getName() + " 结果：" + resultFuture.get());
  40.     }
  41. }

7. allOf

场景：当有一批任务交给线程池执行，我们需要获取所有线程的返回结果

  • Future的get()时阻塞的，如果循环get()每一个线程的结果，一个线程会卡住后面所有线程
  • CompletionService的take().get()虽然不会因为某个线程阻塞后面的线程，但是功能不丰富
  • CompletableFuture提供的功能丰富，使用简单，代码优雅
  • Java9中CompletableFuture还添加了completeOnTimeout、orTimeout，方便对超时任务的处理

测试用例

  • 一串数字1, 2, 3, 4, 5, 6, 7, 8, 9, 10
  • 开启线程，执行乘以2的计算
  • 其中任务2会抛出异常
  • 要获取所有线程的返回结果和异常结果

这里用handle方法来处理线程的结果

  1. public class Thread17_AllOf {
  2.     public static void main(String[] args) {
  3.        System.out.println("==========begin==========");
  4.         //记录开始时间
  5.         Long start = System.currentTimeMillis();
  6.         //定长10线程池
  7.         ExecutorService executor = Executors.newFixedThreadPool(3);
  8.         //任务
  9.         final List<Integer> taskList = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
  11.         List<String> resultList = new ArrayList<>();
  12.         Map<String, String> errorList = new HashMap<>();
  14.         Stream<CompletableFuture<String>> completableFutureStream = taskList.stream()
  15.               .map(num -> CompletableFuture
  16.                 .supplyAsync(() -> getDouble(num), executor)
  17.                 .handle((s, throwable) -> {
  18.                   if (throwable == null) {
  19.                     System.out.println("任务" + num + "完成! result=" + s + ", " + new Date());
  20.                     resultList.add(s.toString());
  21.                   } else {
  22.                     System.out.println("任务" + num + "异常! e=" + throwable + ", " + new Date());
  23.                     errorList.put(num.toString(), throwable.getMessage());
  24.                   }
  25.                   return "";
  26.                  })
  27.               );
  29.         CompletableFuture[] completableFutures = completableFutureStream.toArray(CompletableFuture[]::new);
  31.         CompletableFuture.allOf(completableFutures)
  32.           .whenComplete((v, th) -> {
  33.             System.out.println("所有任务执行完成触发\n resultList=" + resultList + "\n errorList=" + errorList + "\n耗时=" + (System.currentTimeMillis() - start));
  34.           }).join();
  36.    System.out.println("==========end==========");
  38.     //根据数字判断线程休眠的时间
  39.     public static Integer getDouble(Integer i) {
  40.         try {
  41.             if (i == 1) {
  42.                 //任务1耗时3秒
  43.                 Thread.sleep(3000);
  44.             } else if (i == 2) {
  45.                 //任务2耗时1秒,还出错
  46.                 Thread.sleep(1000);
  47.                 throw new RuntimeException("出异常了");
  48.             } else {
  49.                 //其它任务耗时1秒
  50.                 Thread.sleep(1000);
  51.             }
  52.         } catch (InterruptedException e) {
  53.             e.printStackTrace();
  54.         }
  55.         return 2 * i;
  56.     }
  57. }