Java Future与CompletableFuture的用法、区别

╰半橙微兮° 2022-09-08 15:53 288阅读 0赞

Future的用法  
多线程场景时，一般是实现runnable接口，覆写run方法，返回值是void类型，因此这种情况下不需要线程的返回结果。   
如果需要线程的返回结果，就需要用callable接口来代替了。   
callable用法和runnable一样，只不过覆写的是call方法，该方法有一个泛型返回值类型，可以根据需要指定。   
那么何时到Future呢？当你启动callable线程时，就可以声明一个Future对象，用于接收返回结果。  
Futrue可以监视目标线程调用call的情况，当你调用Future的get()方法以获得结果时，调用方的线程就被阻塞，直到目标线程的call方法结束并返回结果。  
Future接口，一般都是取回Callable执行的状态用的。其中的主要方法：

cancel，取消Callable的执行，当Callable还没有完成时  
get，获得Callable的返回值  
isCanceled，判断是否取消了  
isDone，判断是否完成  
举个栗子。  
四个刚需（线程）去买房摇号，future获取摇号结果。摇号结果未出，就一直阻塞。

public class FutureTest \{ 
  
    /\*\* 
     \* 买房摇号 
     \*/ 
    public static class Yaohao implements Callable<Integer> \{ 
        /\*\* 
         \* 返回摇号结果 
         \* @return 0：中签   1：没中 
         \* @throws Exception 
         \*/ 
        @Override 
        public Integer call() throws Exception \{ 
            Random random = new Random(); 
            //模拟摇号，10天内出结果 
            TimeUnit.SECONDS.sleep(random.nextInt(10)); 
            int result = random.nextInt(2); 
            System.out.println("     "+Thread.currentThread().getName()+" is done!"); 
            return result; 
        \} 
    \} 
  
    public static void main(String\[\] args) throws InterruptedException, ExecutionException \{ 
  
        Yaohao gangxu1 = new Yaohao(); 
        Yaohao gangxu2 = new Yaohao(); 
        ExecutorService es = Executors.newCachedThreadPool(); 
        Future<Integer> result1 = es.submit(gangxu1); 
        Future<Integer> result2 = es.submit(gangxu2); 
        es.shutdown(); 
  
        System.out.println("刚需1，摇号结果："+(result1.get()==1?"中签":"没中")); 
        System.out.println("刚需2，摇号结果："+(result2.get()==1?"中签":"没中")); 
    \} 
  
\} 
CompletableFuture的用法 
 创建异步操作，runAsync（不支持返回值） 和 supplyAsync方法（支持返回值） 
计算结果完成时的回调方法 
   whenComplete：执行完当前任务的线程，继续执行 whenComplete 的任务。 
   whenCompleteAsync： 执行完当前任务的线程，把whenCompleteAsync 的任务继续提交给线程池来执行。 
   exceptionally：当前任务出现异常时，执行exceptionally中的回调方法。 
thenApply 方法，当一个线程依赖另一个线程时，可以使用 thenApply 方法来把这两个线程串行化。 
handle 方法 
handle 是执行任务完成时对结果的处理。 
handle 方法和 thenApply 方法处理方式基本一样。不同的是 handle 是在任务完成后再执行，还可以处理异常的任务。thenApply 只可以执行正常的任务，任务出现异常则不执行 thenApply 方法。 
thenAccept 消费处理结果，接收任务的处理结果，并消费处理，无返回结果。 
thenRun 方法，跟 thenAccept 方法不一样的是，不关心任务的处理结果。只要上面的任务执行完成，就开始执行 thenAccept 。 
thenCombine 合并任务，thenCombine 会把 两个 CompletionStage 的任务都执行完成后，把两个任务的结果一块交给 thenCombine 来处理。 
thenCompose 方法，thenCompose 方法允许你对两个 CompletionStage 进行流水线操作，第一个操作完成时，将其结果作为参数传递给第二个操作。 
案例：

## 1、 runAsync 和 supplyAsync方法 ##

CompletableFuture 提供了四个静态方法来创建一个异步操作。

public static CompletableFuture<Void> runAsync(Runnable runnable)
 public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)
 public static CompletableFuture supplyAsync(Supplier supplier)
 public static CompletableFuture supplyAsync(Supplier supplier, Executor executor)

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

*  runAsync方法不支持返回值。
 *  supplyAsync可以支持返回值。

## 示例 ##

//无返回值
 public static void runAsync() throws Exception {
 CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
 try {
 TimeUnit.SECONDS.sleep(1);
 } catch (InterruptedException e) {
 }
 System.out.println("run end ...");
 });
 
 future.get();
 }
 
 //有返回值
 public static void supplyAsync() throws Exception { 
 CompletableFuture<Long> future = CompletableFuture.supplyAsync(() -> {
 try {
 TimeUnit.SECONDS.sleep(1);
 } catch (InterruptedException e) {
 }
 System.out.println("run end ...");
 return System.currentTimeMillis();
 });
 
 long time = future.get();
 System.out.println("time = "+time);
 }

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

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

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)

可以看到Action的类型是BiConsumer<? super T,? super Throwable>它可以处理正常的计算结果，或者异常情况。

whenComplete 和 whenCompleteAsync 的区别：  
whenComplete：是执行当前任务的线程执行继续执行 whenComplete 的任务。  
whenCompleteAsync：是执行把 whenCompleteAsync 这个任务继续提交给线程池来进行执行。

## 示例 ##

public static void whenComplete() throws Exception {
 CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
 try {
 TimeUnit.SECONDS.sleep(1);
 } catch (InterruptedException e) {
 }
 if(new Random().nextInt()%2>=0) {
 int i = 12/0;
 }
 System.out.println("run end ...");
 });
 
 future.whenComplete(new BiConsumer<Void, Throwable>() {
 @Override
 public void accept(Void t, Throwable action) {
 System.out.println("执行完成！");
 }
 
 });
 future.exceptionally(new Function<Throwable, Void>() {
 @Override
 public Void apply(Throwable t) {
 System.out.println("执行失败！"+t.getMessage());
 return null;
 }
 });
 
 TimeUnit.SECONDS.sleep(2);
 }

## 3、 thenApply 方法 ##

当一个线程依赖另一个线程时，可以使用 thenApply 方法来把这两个线程串行化。

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)

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

## 示例 ##

private static void thenApply() throws Exception {
 CompletableFuture<Long> future = CompletableFuture.supplyAsync(new Supplier<Long>() {
 @Override
 public Long get() {
 long result = new Random().nextInt(100);
 System.out.println("result1="+result);
 return result;
 }
 }).thenApply(new Function<Long, Long>() {
 @Override
 public Long apply(Long t) {
 long result = t*5;
 System.out.println("result2="+result);
 return result;
 }
 });
 
 long result = future.get();
 System.out.println(result);
 }

第二个任务依赖第一个任务的结果。

## 4、 handle 方法 ##

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

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

## 示例 ##

public static void handle() throws Exception{
 CompletableFuture<Integer> future = CompletableFuture.supplyAsync(new Supplier<Integer>() {
 
 @Override
 public Integer get() {
 int i= 10/0;
 return new Random().nextInt(10);
 }
 }).handle(new BiFunction<Integer, Throwable, Integer>() {
 @Override
 public Integer apply(Integer param, Throwable throwable) {
 int result = -1;
 if(throwable==null){
 result = param * 2;
 }else{
 System.out.println(throwable.getMessage());
 }
 return result;
 }
 });
 System.out.println(future.get());
 }

从示例中可以看出，在 handle 中可以根据任务是否有异常来进行做相应的后续处理操作。而 thenApply 方法，如果上个任务出现错误，则不会执行 thenApply 方法。

## 5、 thenAccept 消费处理结果 ##

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

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

## 示例 ##

public static void thenAccept() throws Exception{
 CompletableFuture<Void> future = CompletableFuture.supplyAsync(new Supplier<Integer>() {
 @Override
 public Integer get() {
 return new Random().nextInt(10);
 }
 }).thenAccept(integer -> {
 System.out.println(integer);
 });
 future.get();
 }

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

## 6、thenRun 方法 ##

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

public CompletionStage<Void> thenRun(Runnable action);
 public CompletionStage<Void> thenRunAsync(Runnable action);
 public CompletionStage<Void> thenRunAsync(Runnable action,Executor executor);

## 示例 ##

public static void thenRun() throws Exception{
 CompletableFuture<Void> future = CompletableFuture.supplyAsync(new Supplier<Integer>() {
 @Override
 public Integer get() {
 return new Random().nextInt(10);
 }
 }).thenRun(() -> {
 System.out.println("thenRun ...");
 });
 future.get();
 }

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

## 7、thenCombine 合并任务 ##

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

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

## 示例 ##

private static void thenCombine() throws Exception {
 CompletableFuture<String> future1 = CompletableFuture.supplyAsync(new Supplier<String>() {
 @Override
 public String get() {
 return "hello";
 }
 });
 CompletableFuture<String> future2 = CompletableFuture.supplyAsync(new Supplier<String>() {
 @Override
 public String get() {
 return "hello";
 }
 });
 CompletableFuture<String> result = future1.thenCombine(future2, new BiFunction<String, String, String>() {
 @Override
 public String apply(String t, String u) {
 return t+" "+u;
 }
 });
 System.out.println(result.get());
 }

## 8、thenAcceptBoth ##

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

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

## 示例 ##

private static void thenAcceptBoth() throws Exception {
 CompletableFuture<Integer> f1 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
 @Override
 public Integer get() {
 int t = new Random().nextInt(3);
 try {
 TimeUnit.SECONDS.sleep(t);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 System.out.println("f1="+t);
 return t;
 }
 });
 
 CompletableFuture<Integer> f2 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
 @Override
 public Integer get() {
 int t = new Random().nextInt(3);
 try {
 TimeUnit.SECONDS.sleep(t);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 System.out.println("f2="+t);
 return t;
 }
 });
 f1.thenAcceptBoth(f2, new BiConsumer<Integer, Integer>() {
 @Override
 public void accept(Integer t, Integer u) {
 System.out.println("f1="+t+";f2="+u+";");
 }
 });
 }

## 9、applyToEither 方法 ##

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

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

## 示例 ##

private static void applyToEither() throws Exception {
 CompletableFuture<Integer> f1 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
 @Override
 public Integer get() {
 int t = new Random().nextInt(3);
 try {
 TimeUnit.SECONDS.sleep(t);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 System.out.println("f1="+t);
 return t;
 }
 });
 CompletableFuture<Integer> f2 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
 @Override
 public Integer get() {
 int t = new Random().nextInt(3);
 try {
 TimeUnit.SECONDS.sleep(t);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 System.out.println("f2="+t);
 return t;
 }
 });
 
 CompletableFuture<Integer> result = f1.applyToEither(f2, new Function<Integer, Integer>() {
 @Override
 public Integer apply(Integer t) {
 System.out.println(t);
 return t * 2;
 }
 });
 
 System.out.println(result.get());
 }

## 10、acceptEither 方法 ##

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

public CompletionStage<Void> acceptEither(CompletionStage<? extends T> other,Consumer<? super T> action);
 public CompletionStage<Void> acceptEitherAsync(CompletionStage<? extends T> other,Consumer<? super T> action);
 public CompletionStage<Void> acceptEitherAsync(CompletionStage<? extends T> other,Consumer<? super T> action,Executor executor);

## 示例 ##

private static void acceptEither() throws Exception {
 CompletableFuture<Integer> f1 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
 @Override
 public Integer get() {
 int t = new Random().nextInt(3);
 try {
 TimeUnit.SECONDS.sleep(t);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 System.out.println("f1="+t);
 return t;
 }
 });
 
 CompletableFuture<Integer> f2 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
 @Override
 public Integer get() {
 int t = new Random().nextInt(3);
 try {
 TimeUnit.SECONDS.sleep(t);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 System.out.println("f2="+t);
 return t;
 }
 });
 f1.acceptEither(f2, new Consumer<Integer>() {
 @Override
 public void accept(Integer t) {
 System.out.println(t);
 }
 });
 }

## 11、runAfterEither 方法 ##

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

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

## 示例 ##

private static void runAfterEither() throws Exception {
 CompletableFuture<Integer> f1 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
 @Override
 public Integer get() {
 int t = new Random().nextInt(3);
 try {
 TimeUnit.SECONDS.sleep(t);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 System.out.println("f1="+t);
 return t;
 }
 });
 
 CompletableFuture<Integer> f2 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
 @Override
 public Integer get() {
 int t = new Random().nextInt(3);
 try {
 TimeUnit.SECONDS.sleep(t);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 System.out.println("f2="+t);
 return t;
 }
 });
 f1.runAfterEither(f2, new Runnable() {
 
 @Override
 public void run() {
 System.out.println("上面有一个已经完成了。");
 }
 });
 }

## 12、runAfterBoth ##

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

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

## 示例 ##

private static void runAfterBoth() throws Exception {
 CompletableFuture<Integer> f1 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
 @Override
 public Integer get() {
 int t = new Random().nextInt(3);
 try {
 TimeUnit.SECONDS.sleep(t);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 System.out.println("f1="+t);
 return t;
 }
 });
 
 CompletableFuture<Integer> f2 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
 @Override
 public Integer get() {
 int t = new Random().nextInt(3);
 try {
 TimeUnit.SECONDS.sleep(t);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 System.out.println("f2="+t);
 return t;
 }
 });
 f1.runAfterBoth(f2, new Runnable() {
 
 @Override
 public void run() {
 System.out.println("上面两个任务都执行完成了。");
 }
 });
 }

## 13、thenCompose 方法 ##

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

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

## 示例 ##

private static void thenCompose() throws Exception {
 CompletableFuture<Integer> f = CompletableFuture.supplyAsync(new Supplier<Integer>() {
 @Override
 public Integer get() {
 int t = new Random().nextInt(3);
 System.out.println("t1="+t);
 return t;
 }
 }).thenCompose(new Function<Integer, CompletionStage<Integer>>() {
 @Override
 public CompletionStage<Integer> apply(Integer param) {
 return CompletableFuture.supplyAsync(new Supplier<Integer>() {
 @Override
 public Integer get() {
 int t = param *2;
 System.out.println("t2="+t);
 return t;
 }
 });
 }
 
 });
 System.out.println("thenCompose result : "+f.get());
 }

Future和CompletableFuture的区别  
Future在Java5就引入了。

优点：一定程度上让一个线程池内的任务异步执行  
缺点：传统回调最大的问题就是不能将控制流分离到不同的事件处理器中。例如主线程等待各个异步执行的线程返回的结果来做下一步操作，则必须阻塞在future.get()的地方等待结果返回，变成了同步。

CompletableFuture在Java8引入。

实现了Future和CompletionStage接口，保留了Future的优点，并且弥补了其不足。即异步的任务完成后，需要用其结果继续操作时，无需等待。可以直接通过thenAccept、thenApply、thenCompose等方式将前面异步处理的结果交给另外一个异步事件处理线程来处理。  
可见，这种方式才是我们需要的异步处理。一个控制流的多个异步事件处理能无缝的连接在一起。