创建

工厂方式创建cf对象

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)

没有指定Executor的方法会使用ForkJoinPool.commonPool() 作为它的线程池执行异步代码。如果指定线程池，则使用指定的线程池运行。以下所有的方法都类同。

runAsync

runAsync方法不支持返回值。

  1. import lombok.extern.slf4j.Slf4j;
  3. import java.util.concurrent.CompletableFuture;
  4. import java.util.concurrent.ExecutionException;
  5. import java.util.concurrent.TimeUnit;
  7. @Slf4j
  8. public class CompletableFutureDemo {
  9.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  10.         CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
  11.             try {
  12.                 log.info("run future start");
  13.                 // 模拟耗时操作
  14.                 TimeUnit.SECONDS.sleep(1);
  15.             } catch (InterruptedException e) {
  16.                 e.printStackTrace();
  17.             }
  18.             log.info("run future end");
  19.         });
  20.         future.get();
  21.         log.info("run main end");
  22.     }
  23. }

结果

  1. 09:44:43.101 [ForkJoinPool.commonPool-worker-1] INFO CompletableFutureDemo - run future start
  2. 09:44:44.109 [ForkJoinPool.commonPool-worker-1] INFO CompletableFutureDemo - run future end
  3. 09:44:44.109 [main] INFO cn.cubg.callcenter.commons.cron.CompletableFutureDemo - run main end

supplyAsync

supplyAsync可以支持返回值。

  1. import lombok.extern.slf4j.Slf4j;
  3. import java.util.concurrent.CompletableFuture;
  4. import java.util.concurrent.ExecutionException;
  5. import java.util.concurrent.TimeUnit;
  7. @Slf4j
  8. public class CompletableFutureDemo {
  9.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  10.         long startTime = System.currentTimeMillis();
  11.         CompletableFuture<Long> future = CompletableFuture.supplyAsync(() -> {
  12.             try {
  13.                 log.info("run future start");
  14.                 TimeUnit.SECONDS.sleep(1);
  15.             } catch (InterruptedException e) {
  16.             }
  17.             log.info("run future end");
  18.             return System.currentTimeMillis();
  19.         });
  21.         long endTime = future.get();
  22.         log.info("run main time {}", endTime - startTime);
  23.     }
  24. }

计算结果完成时的回调方法

CompletableFuture获取结果的方式有如下4个方法：
1：get 阻塞获取结果，实现Future的get接口，显式抛出异常
2：getNow(T valueIfAbsent) 获取执行结果，如果当前任务未执行完成，则返回valueIfAbsent
3: join 执行完成后返回执行结果，或者抛出unchecked异常
4: T get(long timeout, TimeUnit unit) 在有限时间内获取数据
以下是CompletableFuture的创建对象以及api的使用

当CompletableFuture的计算结果完成，或者抛出异常的时候，可以执行特定的Action。主要是下面的方法：

  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)
  4. public CompletableFuture<T> exceptionally(Function<Throwable,? extends T> fn)

可以看到Action的类型是BiConsumer<? super T,? super Throwable>它可以处理正常的计算结果，或者异常情况。
whenComplete 和 whenCompleteAsync 的区别： whenComplete：是执行当前任务的线程执行继续执行 whenComplete 的任务。 whenCompleteAsync：是执行把 whenCompleteAsync 这个任务继续提交给线程池来进行执行。

whenComplete

  1. @Slf4j
  2. public class CompletableFutureDemo {
  3.     public static void main(String[] args)  {
  4.         CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
  5.             try {
  6.                 TimeUnit.SECONDS.sleep(1);
  7.             } catch (InterruptedException e) {
  8.             }
  9.             if(new Random().nextInt()%2>=0) {
  10.                 log.info("run random func");
  11.                 int i = 12/0;
  12.             }
  13.             System.out.println("run end ...");
  14.         });
  15.         future.whenComplete((unused, throwable) -> System.out.println("执行完成！"));
  16.         future.exceptionally(t -> {
  17.             System.out.println("执行失败！"+t.getMessage());
  18.             return null;
  19.         });
  20.         try {
  21.             TimeUnit.SECONDS.sleep(2);
  22.         } catch (InterruptedException e) {
  23.             e.printStackTrace();
  24.         }
  25.     }
  26. }

thenApply 方法

当一个线程依赖另一个线程时，可以使用 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)

Function<? super T,? extends U> T：上一个任务返回结果的类型 U：当前任务的返回值类型

  1. @Slf4j
  2. public class CompletableFutureDemo {
  3.     public static void main(String[] args)  {
  4.         CompletableFuture<Long> future = CompletableFuture.supplyAsync(() -> {
  5.             long result = new Random().nextInt(10)%2;
  6.             if (result==0){
  7.                 log.info("result1={}",result/0);
  8.             }else {
  9.                 log.info("result1={}",result);
  10.             }
  12.             return result;
  13.         }).thenApply(t -> {
  14.             long result = t*5;
  15.             log.info("result2={}",result);
  16.             return result;
  17.         });
  19.         long result = 0;
  20.         try {
  21.             result = future.get();
  22.         } catch (Exception e) {
  23.           log.error(e.getMessage());
  24.         }
  25.        log.info("result ={}",result);
  26.     }
  27. }

handle 方法

handle 是执行任务完成时对结果的处理。 handle 方法和 thenApply 方法处理方式基本一样。不同的是 handle 是在任务完成后再执行，还可以处理异常的任务。thenApply 只可以执行正常的任务，任务出现异常则不执行 thenApply 方法。

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

eg:

  1. @Slf4j
  2. public class CompletableFutureDemo {
  3.     public static void main(String[] args) {
  4.         CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
  5.             Integer result = new Random().nextInt(100) % 2;
  6.             if (result == 0) {
  7.                 log.info("result1={}", result / 0);
  8.             } else {
  9.                 log.info("result1={}", result);
  10.             }
  11.             return result;
  12.         }).handle((param, throwable) -> {
  13.             int result = -1;
  14.             if (throwable == null) {
  15.                 result = param * 2;
  16.             } else {
  17.                 log.info(throwable.getMessage());
  18.             }
  19.             return result;
  20.         });
  22.         long result = 0;
  23.         try {
  24.             result = future.get();
  25.         } catch (Exception e) {
  26.             log.error(e.getMessage());
  27.         }
  28.         log.info("result ={}", result);
  29.     }
  30. }

thenAccept 消费处理结果

接收任务的处理结果，并消费处理，无返回结果。

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

eg:

  1. public static void thenAccept() throws Exception{
  2.     CompletableFuture<Void> future = CompletableFuture.supplyAsync(new Supplier<Integer>() {
  3.         @Override
  4.         public Integer get() {
  5.             return new Random().nextInt(10);
  6.         }
  7.     }).thenAccept(integer -> {
  8.         System.out.println(integer);
  9.     });
  10.     future.get();
  11. }

从示例代码中可以看出，该方法只是消费执行完成的任务，并可以根据上面的任务返回的结果进行处理。并没有后续的输错操作。

thenRun 方法

跟 thenAccept 方法不一样的是，不关心任务的处理结果。只要上面的任务执行完成，就开始执行 thenAccept 。

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

eg

  1. public static void thenRun() throws Exception{
  2.     CompletableFuture<Void> future = CompletableFuture.supplyAsync(new Supplier<Integer>() {
  3.         @Override
  4.         public Integer get() {
  5.             return new Random().nextInt(10);
  6.         }
  7.     }).thenRun(() -> {
  8.         System.out.println("thenRun ...");
  9.     });
  10.     future.get();
  11. }

该方法同 thenAccept 方法类似。不同的是上个任务处理完成后，并不会把计算的结果传给 thenRun 方法。只是处理玩任务后，执行 thenAccept 的后续操作。

thenCombine 合并任务

thenCombine 会把 两个 CompletionStage 的任务都执行完成后，把两个任务的结果一块交给 thenCombine 来处理。

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

eg:

  1. @Slf4j
  2. public class CompletableFutureDemo {
  3.     public static void main(String[] args) {
  4.         log.info("start");
  5.         CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> {
  6.             try {
  7.                 TimeUnit.SECONDS.sleep(1);
  8.             } catch (InterruptedException e) {
  9.                 e.printStackTrace();
  10.             }
  11.             return "hello";
  12.         });
  16.         CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> {
  17.             try {
  18.                 TimeUnit.SECONDS.sleep(2);
  19.             } catch (InterruptedException e) {
  20.                 e.printStackTrace();
  21.             }
  22.             return "world";
  23.         });
  24.         CompletableFuture<String> result = f1.thenCombine(f2, (r1, r2) -> r1+" "+r2);
  26.         String r = null;
  27.         try {
  28.             r=  result.get();
  29.         } catch (Exception e) {
  30.             e.printStackTrace();
  31.         }
  32.         log.info("result ={}",r);
  33.     }
  34. }

thenAcceptBoth

当两个CompletionStage都执行完成后，把结果一块交给thenAcceptBoth来进行消耗

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

eg

  1. @Slf4j
  2. public class CompletableFutureDemo {
  3.     public static void main(String[] args) {
  4.         log.info("start");
  5.         CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> {
  6.             try {
  7.                 TimeUnit.SECONDS.sleep(1);
  8.             } catch (InterruptedException e) {
  9.                 e.printStackTrace();
  10.             }
  11.             return "hello";
  12.         });
  14.         CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> {
  15.             try {
  16.                 TimeUnit.SECONDS.sleep(2);
  17.             } catch (InterruptedException e) {
  18.                 e.printStackTrace();
  19.             }
  20.             return " world";
  21.         });
  22.         final String[] r = {""};
  23.        CompletableFuture f = f1.thenAcceptBoth(f2, new BiConsumer<String, String>() {
  24.             @Override
  25.             public void accept(String s1, String s2) {
  26.                 r[0] = s1+s2;
  27.             }
  28.         });
  31.         try {
  32.            f.get();
  33.         } catch (Exception e) {
  34.             e.printStackTrace();
  35.         }
  36.         log.info("result ={}", r[0]);
  37.     }
  38. }

acceptEither取计算速度最快的结果

针对两个CompletionStage，将计算最快的那个CompletionStage的结果用来作为下一步的消耗。 此方法接受Consumer只对结果进行消耗.

  1. public CompletionStage<Void> acceptEither(CompletionStage<? extends T> other,Consumer<? super T> action); 
  2. public CompletionStage<Void> acceptEitherAsync(CompletionStage<? extends T> other,Consumer<? super T> action); 
  3. public CompletionStage<Void> acceptEitherAsync(CompletionStage<? extends T> other,Consumer<? super T>

eg

  1. @Slf4j
  2. public class CompletableFutureDemo {
  3.     public static void main(String[] args) {
  4.         log.info("start");
  5.         CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> {
  6.             try {
  7.                 int t = new Random().nextInt(4);
  8.                 TimeUnit.SECONDS.sleep(t);
  9.             } catch (InterruptedException e) {
  10.                 e.printStackTrace();
  11.             }
  12.             return "hello";
  13.         });
  14.         CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> {
  15.             try {
  16.                 int t = new Random().nextInt(4);
  17.                 TimeUnit.SECONDS.sleep(t);
  18.             } catch (InterruptedException e) {
  19.                 e.printStackTrace();
  20.             }
  21.             return "world";
  22.         });
  24.         CompletableFuture<Void> r = f1.acceptEither(f2, result -> {
  25.             log.info("result={}", result);
  26.         });
  27.         try {
  28.             r.get();
  29.         } catch (Exception e) {
  30.             e.printStackTrace();
  31.         }
  32.     }
  33. }

applyToEither计算最快的cf的结果转换

两个CompletionStage，谁执行返回的结果快，就用那个CompletionStage的结果进行下一步的转化操作。

  1. public <U> CompletionStage<U> applyToEither(CompletionStage<? extends T> other,Function<? super T, U> fn);
  2. public <U> CompletionStage<U> applyToEitherAsync(CompletionStage<? extends T> other,Function<? super T, U> fn);
  3. public <U> CompletionStage<U> applyToEitherAsync(CompletionStage<? extends T> other,Function<? super T, U> fn,Executor executor);

eg

  1. @Slf4j
  2. public class CompletableFutureDemo {
  3.     public static void main(String[] args) {
  4.         log.info("start");
  5.         CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> {
  6.             try {
  7.                 TimeUnit.SECONDS.sleep(1);
  8.             } catch (InterruptedException e) {
  9.                 e.printStackTrace();
  10.             }
  11.             return "hello";
  12.         });
  14.         CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> {
  15.             try {
  16.                 TimeUnit.SECONDS.sleep(2);
  17.             } catch (InterruptedException e) {
  18.                 e.printStackTrace();
  19.             }
  20.             return " world";
  21.         });
  22.         final String[] r = {""};
  23.         CompletableFuture f = f1.acceptEither(f2, s -> r[0] = s);
  25.         try {
  26.             f.get();
  27.         } catch (Exception e) {
  28.             e.printStackTrace();
  29.         }
  30.         log.info("result ={}", r[0]);
  31.     }
  32. }

runAfterEither 方法

两个CompletionStage，任何一个完成了都会执行下一步的操作（Runnable）

  1. public CompletionStage<Void> runAfterEither(CompletionStage<?> other,Runnable action);
  2. public CompletionStage<Void> runAfterEitherAsync(CompletionStage<?> other,Runnable action);
  3. public CompletionStage<Void> runAfterEitherAsync(CompletionStage<?> other,Runnable action,Executor executor);

eg

  1. @Slf4j
  2. public class CompletableFutureDemo {
  3.     public static void main(String[] args) {
  4.         log.info("start");
  5.         CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> {
  6.             try {
  7.                 TimeUnit.SECONDS.sleep(1);
  8.             } catch (InterruptedException e) {
  9.                 e.printStackTrace();
  10.             }
  11.             return "hello";
  12.         });
  14.         CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> {
  15.             try {
  16.                 TimeUnit.SECONDS.sleep(2);
  17.             } catch (InterruptedException e) {
  18.                 e.printStackTrace();
  19.             }
  20.             return " world";
  21.         });
  23.         CompletableFuture f = f1.runAfterEither(f2, new Runnable() {
  24.             @Override
  25.             public void run() {
  26.                 log.info("有一个执行完成,");
  27.             }
  28.         });
  30.         try {
  31.             f.get();
  32.         } catch (Exception e) {
  33.             e.printStackTrace();
  34.         }
  35.     }
  36. }

runAfterBoth

两个CompletionStage，都完成了计算才会执行下一步的操作（Runnable）

  1. public CompletionStage<Void> runAfterBoth(CompletionStage<?> other,Runnable action);
  2. public CompletionStage<Void> runAfterBothAsync(CompletionStage<?> other,Runnable action);
  3. public CompletionStage<Void> runAfterBothAsync(CompletionStage<?> other,Runnable action,Executor executor);

eg:

  1. @Slf4j
  2. public class CompletableFutureDemo {
  3.     public static void main(String[] args) {
  4.         log.info("start");
  5.         CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> {
  6.             try {
  7.                 TimeUnit.SECONDS.sleep(1);
  8.             } catch (InterruptedException e) {
  9.                 e.printStackTrace();
  10.             }
  11.             return "hello";
  12.         });
  14.         CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> {
  15.             try {
  16.                 TimeUnit.SECONDS.sleep(2);
  17.             } catch (InterruptedException e) {
  18.                 e.printStackTrace();
  19.             }
  20.             return " world";
  21.         });
  23.         CompletableFuture f = f1.runAfterBoth(f2, new Runnable() {
  24.             @Override
  25.             public void run() {
  26.                 log.info("有一个执行完成,");
  27.             }
  28.         });
  30.         try {
  31.             f.get();
  32.         } catch (Exception e) {
  33.             e.printStackTrace();
  34.         }
  35.     }
  36. }

thenCompose 方法

thenCompose 方法允许你对两个 CompletionStage 进行流水线操作，第一个操作完成时，将其结果作为参数传递给第二个操作。

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

eg:

  1. @Slf4j
  2. public class CompletableFutureDemo {
  3.     public static void main(String[] args) {
  4.         log.info("start");
  5.         CompletableFuture<String> f = CompletableFuture.supplyAsync(() -> {
  6.             try {
  7.                 TimeUnit.SECONDS.sleep(1);
  8.             } catch (InterruptedException e) {
  9.                 e.printStackTrace();
  10.             }
  11.             return "hello";
  12.         }).thenCompose(new Function<String, CompletionStage<String>>() {
  13.             @Override
  14.             public CompletionStage<String> apply(String s) {
  16.                return CompletableFuture.supplyAsync(new Supplier<String>() {
  17.                    @Override
  18.                    public String get() {
  19.                        try {
  20.                            TimeUnit.SECONDS.sleep(2);
  21.                        } catch (InterruptedException e) {
  22.                            e.printStackTrace();
  23.                        }
  24.                        return s+" world";
  25.                    }
  26.                });
  27.             }
  28.         });
  30.         try {
  31.             String r= f.get();
  32.             log.info("result={}",r);
  33.         } catch (Exception e) {
  34.             e.printStackTrace();
  35.         }
  36.     }
  37. }

anyOf

方法名 描述
anyOf(CompletableFuture<?>… cfs) 在任何一个Future对象结束后结束，并返回一个future。 
  1. @Slf4j
  2. public class CompletableFutureDemo {
  3.     public static void main(String[] args) {
  5.         Random rand = new Random();
  6.         CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> {
  7.             try {
  8.                 Thread.sleep(rand.nextInt(2000));
  9.             } catch (InterruptedException e) {
  10.                 e.printStackTrace();
  11.             }
  12.             return "F1";
  13.         });
  14.         CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> {
  15.             try {
  16.                 Thread.sleep(rand.nextInt(2000));
  17.             } catch (InterruptedException e) {
  18.                 e.printStackTrace();
  19.             }
  20.             return "F2";
  21.         });
  22.         CompletableFuture<String> f3 = CompletableFuture.supplyAsync(() -> {
  23.             try {
  24.                 Thread.sleep(rand.nextInt(2000));
  25.             } catch (InterruptedException e) {
  26.                 e.printStackTrace();
  27.             }
  28.             return "F3";
  29.         });
  31.         CompletableFuture<Object> future = CompletableFuture.anyOf(f1, f2, f3);
  32.         try {
  33.             System.out.println(future.get());
  34.         } catch (InterruptedException e) {
  35.             e.printStackTrace();
  36.         } catch (ExecutionException e) {
  37.             e.printStackTrace();
  38.         }
  39.     }
  40. }

参考

https://www.jianshu.com/p/807e6822292a