CompletableFuture异步任务详解

原创
Endwas
2021-10-31
Java基础
1935
0

在工作中，常常会调用多个服务或者方法去获取不同的数据，如果传统做法就是串行一个个获取，然后封装返回。我们可以尝试使用CompletableFuture，将多个操作交给异步线程执行，然后主线程等待最长任务完成，将所有结果一并返回即可。

Future局限性

当我们得到包含结果的Future时，我们可以使用get方法等待线程完成并获取返回值，但我们都知道future.get()是阻塞的方法，会一直等到线程执行完毕拿到返回值。我们可以看到FutureTask中的get方法，就是循环代码直到线程执行完成返回。

  /**
  * Awaits completion or aborts on interrupt or timeout.
  *
  * @param timed true if use timed waits
  * @param nanos time to wait, if timed
  * @return state upon completion
  */
  private int awaitDone(boolean timed, long nanos) throws InterruptedException {
	  final long deadline = timed ? System.nanoTime() + nanos : 0L;
	  WaitNode q = null;
	  boolean queued = false;
	  for (;;) {
	  //循环 省略代码
	  ...
	  }
  ｝

我们考虑一种场景，如果我们执行完该异步任务1需要拿到返回值，然后使用该返回值执行其他异步调用2，那么我们就需要在主线程阻塞等待异步任务1完成，然后交由执行异步任务2，然后继续阻塞等待任务2的返回。这样不仅阻塞主线程，而且性能差。

CompletableFuture介绍

什么是CompletableFuture：CompletableFuture结合了Future的优点，提供了非常强大的Future的扩展功能，可以帮助我们简化异步编程的复杂性，提供了函数式编程的能力，可以通过回调的方式处理计算结果，并且提供了转换和组合CompletableFuture的方法。

函数式编程：使用Functional Interface作为参数，可使用lambda表达式建议实现，编程便捷，之前有讲解过。

常用的辅助方法

  1. isCompletedExceptionally：该CompletableFuture是否异常结束。
// 包括取消cancel、显示调用completeExceptionally、中断。
public boolean isCompletedExceptionally() {
	Object r;
	return ((r = result) instanceof AltResult) && r != NIL;
}
  1. isCancelled：该CompletableFuture是否在正常执行完成前被取消。
 public boolean isCancelled() {
	 Object r;
	 // 判断异常是否是CancellationException
	 return ((r = result) instanceof AltResult) &&
            (((AltResult)r).ex instanceof CancellationException);
 }
  1. isDone：该CompletableFuture是否已经执行结束包括产生异常和取消。
 public boolean isDone() {
 	return result != null;
 }
  1. get：阻塞获取CompletableFuture结果
 public T get() throws InterruptedException, ExecutionException {
	 Object r;
	 return reportGet((r = result) == null ? waitingGet(true) : r);
 }
  1. join：阻塞获取CompletableFuture结果
 public T join() {
        Object r;
        return reportJoin((r = result) == null ? waitingGet(false) : r);
 }

join() 与get() 区别在于join() 返回计算的结果或者抛出一个unchecked异常(CompletionException)，而get() 返回一个具体的异常.

CompletableFuture构建

CompletableFuture有无参数的构造方法，这个时候创建的是未完成的CompletableFuture。使用get会一直阻塞主线程。

所以我们一般使用静态方法来创建实例。

// 无入参有返回值
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier);
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor);
// 无入参无返回值，简单的执行
public static CompletableFuture<Void> runAsync(Runnable runnable);
public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor);

我们注意到每种方法都有一个重构的方法。Executor参数可以手动指定线程池，否则默认ForkJoinPool.commonPool()系统级公共线程池。

【注意】：默认的commonPool的这些线程都是守护线程。我们在编程的时候需要谨慎使用守护线程，如果将我们普通的用户线程设置成守护线程，当我们的程序主线程结束，JVM中不存在其余用户线程，那么CompletableFuture的守护线程会直接退出，造成任务无法完成的问题！！

CompletableFuture常用Api

我们后续讲解的api一般会有三种相似的方法。我就只演示第三种。

  1. xxx(method);
  2. xxxAsync(method);
  3. xxxAsync(method, executor) 三种的区别就是第一种同步执行由主线程执行，第二种异步交由默认线程池，第三种异步交由创建的线程池。

1.构建CompletableFuture实例

现在就将CompletableFuture来试着使用。先从基本的构建开始。

/**
* 1.run + runAsync + supply + supplyAsync
*/
/** 1.异步运行交由默认线程池（forkjoinpool）无入参无返回值run
*   2.异步运行交由创建线程池（threadPoolExecutor）无入参无返回值run
*/
CompletableFuture<Void> run = CompletableFuture.runAsync(() ->
        System.out.println("completablefuture runs asynchronously"));

CompletableFuture<Void> runCustomize = CompletableFuture.runAsync(() ->
        System.out.println("completablefuture runs asynchronously in customize threadPool"
        ), threadPoolExecutor);


//================下面是supply===================
/** 1.异步运行交由默认线程池（forkjoinpool）无入参有返回值supply
*   2.异步运行交由自己创建的线程池 无入参有返回值supply
*/
CompletableFuture<String> supply = CompletableFuture.supplyAsync(() -> {
    System.out.println("completablefuture supplys asynchronously");
    return "success";
});

CompletableFuture<String> supplyCustomize = CompletableFuture.supplyAsync(() -> {
    System.out.println("completablefuture supplys asynchronously in customize threadPool");
    return "success";
}, threadPoolExecutor);

result

completablefuture runs asynchronously completablefuture runs asynchronously in customize threadPool completablefuture supplys asynchronously completablefuture supplys asynchronously in customize threadPool

前两种的CompletableFuture是没有值的，所以当后续链式调用想使用的时候入参是null的。

2.whenComplete

考虑当我们在CompletableFuture执行结束的时候，希望能够得到执行结果、或者异常，然后对结果或者异常做进一步处理。那么我们就需要使用到whenComplete。

  /**
  * 入参是BiConsumer，第一个参数是上一步结果、第二个是上一步执行的异常
  */
 CompletableFuture<String> supply = CompletableFuture.supplyAsync(() -> {
          int a = 1/0;
          System.out.println("completablefuture supplys asynchronously");
          return "success";
 });
 CompletableFuture<String> complete = supply.whenCompleteAsync((result, throwable) -> {
     System.out.println("whenComplete: " + result + " throws " + throwable);
 });

result：

whenComplete: null throws java.util.concurrent.CompletionException: java.lang.ArithmeticException: / by zero

3.handle

handle和whenComplete入参是一样的，但是它可以在执行完成后返回执行结果，而whenComplete只能处理无法返回。

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

example:

/**
* 入参是BiFunction 第一个参数是上一步结果、第二个是上一步执行的异常 
* 返回值可以是任何类型
*/
CompletableFuture<String> handleAsyncCustomize = supply.handleAsync((a, throwable) -> {
            System.out.println("handleAsyncCustomize: " + a + " throws " + throwable);
            return "handleAsync success";
}, threadPoolExecutor);

System.out.println(handleAsyncCustomize.get());

result：

handleAsyncCustomize: success throws null handleAsync success

4.thenApply

thenApply又和handle很类似，有入参也有返回值的，但是他只有一个入参，无法处理上一异步任务异常的情况。如果发生异常get的时候会报错。

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

/**
* Function 入参是第一个异步任务执行结果、出参是返回值
* 若异步任务1异常 则2无法执行 get会报错
*/
 CompletableFuture<String> supplyCustomize = CompletableFuture.supplyAsync(() -> {
     System.out.println("completablefuture supplys asynchronously in customize threadPool");
     int a = 1/0;
     return "success";
 }, threadPoolExecutor);

 CompletableFuture<String> applyAsyncCustomize = supplyCustomize.thenApplyAsync(a -> {
     System.out.println("applyAsyncCustomize " + a);
     return "success";
 }, threadPoolExecutor);
 System.out.println(applyAsyncCustomize.get());

result:

Exception in thread "main" java.util.concurrent.ExecutionException: java.lang.ArithmeticException: / by zero
	at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357)
	at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1908)
	at CompletableFutureTask.main.main(main.java:48)
Caused by: java.lang.ArithmeticException: / by zero
	at CompletableFutureTask.main.lambda$main$3(main.java:40)
	at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1604)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)

5.thenAccept

thenAccept是有入参无返回值，如果继续链式调用那么下一个异步任务将会得到null值。

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

example:

/**
*	Consumer第一个入参是上一步返回结果，若没返回结果则为null
*	无返回值
*/
CompletableFuture<String> supplyCustomize = CompletableFuture.supplyAsync(() -> {
   System.out.println("completablefuture supplys asynchronously in customize threadPool");
    return "success";
}, threadPoolExecutor);

CompletableFuture<Void> applyAsyncCustomize = supplyCustomize.thenAcceptAsync(a ->{
    System.out.println("accept " + a);
}, threadPoolExecutor);
System.out.println(applyAsyncCustomize.get());

result：

completablefuture supplys asynchronously in customize threadPool accept success null

6.thenCompose

和thenCombine有所不同，thenCombine是组合两个CompletableFuture返回结果进行异步处理，而thenCompose则是根据第一个返回结果，封装成新的CompletableFuture返回

example：

CompletableFuture<String> supply = CompletableFuture.supplyAsync(() -> {
     System.out.println("completablefuture supplys asynchronously");
     return "success";
 });
 CompletableFuture<String> future = supply.thenComposeAsync(a -> {
     String name = a + " or fail";
     return CompletableFuture.supplyAsync(() -> {
         return name;
     });
 });
 System.out.println(future.get());

result：

completablefuture supplys asynchronously in customize threadPool success or fail

7.exceptionally

当发生异常时候的处理，注意异常后返回值类型需要和发生异常的CF返回值类型一致，相当于一种服务降级的思想。

example：

/**
*	发生异常时候，返回0
*/
CompletableFuture<Integer> supplyCustomize = CompletableFuture.supplyAsync(() -> {
     int a = 1/0;
     System.out.println("completablefuture supplys asynchronously in customize threadPool");
     return 2;
 }, threadPoolExecutor).exceptionally(a->{
     System.out.println(a);
     return 0;
 });

 System.out.println(supplyCustomize.get());

result：

completablefuture runs asynchronously in customize threadPool java.util.concurrent.CompletionException: java.lang.ArithmeticException: / by zero 0

CompletableFuture组合式Api-任务全完成

1.thenCombine

thenCombine是组合式任务，上面的CompletableFuture使用是链式完成，当完成第一个时候，根据第一个执行结果进行下一步异步调用，而组合式异步，可以做到两个异步任务完全独立，只有当他们都完成时候才会继续执行。

example：

/**
*	BiFunction入参，两个CF的返回结果作为入参，有返回值
*/
CompletableFuture<String> supply = CompletableFuture.supplyAsync(() -> {
	System.out.println("completablefuture supplys asynchronously");
    return "success";
});
// 异步运行交由默认线程池（forkjoinpool）无入参有返回值supply
CompletableFuture<String> supplyCustomize = CompletableFuture.supplyAsync(() -> {
    System.out.println("completablefuture supplys asynchronously in customize threadPool");
    return "success";
}, threadPoolExecutor);

CompletableFuture<String> future = supply.thenCombine(supplyCustomize, (a, b) -> {
    return "thenCombine result: " + a + "-" + b;
});
System.out.println(future.get());

result：

completablefuture supplys asynchronously completablefuture supplys asynchronously in customize threadPool thenCombine result: success-success

2.thenAcceptBoth

thenAcceptBoth和thenCombine区别就是没有返回值，将两个CF返回值进行处理，没有返回值

example：

/**
*	接受CF、BiConsumer，无返回值
*/
CompletableFuture<String> supply = CompletableFuture.supplyAsync(() -> {
    System.out.println("completablefuture supplys asynchronously");
    return "success";
});
// 异步运行交由默认线程池（forkjoinpool）无入参有返回值supply
CompletableFuture<String> supplyCustomize = CompletableFuture.supplyAsync(() -> {
    System.out.println("completablefuture supplys asynchronously in customize threadPool");
    return "success";
}, threadPoolExecutor);

CompletableFuture<Void> bothAsync = supplyCustomize.thenAcceptBothAsync(supply, (a, b) -> {
    System.out.println("thenAcceptBoth result " + a + "-" + b);
}, threadPoolExecutor);

result：

completablefuture supplys asynchronously completablefuture supplys asynchronously in customize threadPool thenAcceptBoth result success-success null

3.runAfterBoth

无入参无返回值，只会在前面两者都运行完成后才会执行runAfterBoth的方法，我们可以在任一CF模拟长时间运行进行测试。

example：

CompletableFuture<String> supply = CompletableFuture.supplyAsync(() -> {
     System.out.println("completablefuture supplys asynchronously");
     return "success";
 });
 // 异步运行交由默认线程池（forkjoinpool）无入参有返回值supply
 CompletableFuture<String> supplyCustomize = CompletableFuture.supplyAsync(() -> {
     System.out.println("completablefuture supplys asynchronously in customize threadPool");
     return "success";
 }, threadPoolExecutor);
 CompletableFuture<Void> bothAsync = supply.runAfterBothAsync(supplyCustomize, () -> {
     System.out.println("run After Both Async");
 });

result：

completablefuture supplys asynchronously completablefuture supplys asynchronously in customize threadPool run After Both Async

CompletableFuture组合式Api-任一任务完成

1.acceptEither

和thenAcceptBoth对应，两个CompletableFuture任一执行完成，就会继续下一步异步任务

example：

/**
*	任务一耗时长，所以当任务二完成就会执行acceptEitherAsync
*	参数 Consumer 入参为执行完成任务返回值
*/
CompletableFuture<String> supply = CompletableFuture.supplyAsync(() -> {
    try {
        TimeUnit.SECONDS.sleep(2);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    System.out.println("completablefuture supplys asynchronously");
    return "success1";
CompletableFuture<String> supplyCustomize = CompletableFuture.supplyAsync(() -> {
    System.out.println("completablefuture supplys asynchronously in customize threadPool");
    return "success2";
}, threadPoolExecutor);
CompletableFuture<Void> future = supply.acceptEitherAsync(supplyCustomize, a -> {
    System.out.println("acceptEither result " + a);
});

result：

completablefuture supplys asynchronously in customize threadPool acceptEither result success2 completablefuture supplys asynchronously

2.applyToEither

对应thenCombine，两个CompletableFuture任一执行完成，就会继续下一步异步任务

example：

/**
*	参数为Function，有入参也有返回值
*/
CompletableFuture<String> supply = CompletableFuture.supplyAsync(() -> {
    try {
        TimeUnit.SECONDS.sleep(2);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    System.out.println("completablefuture supplys asynchronously");
    return "success1";
});
// 异步运行交由默认线程池（forkjoinpool）无入参有返回值supply
CompletableFuture<String> supplyCustomize = CompletableFuture.supplyAsync(() -> {
    System.out.println("completablefuture supplys asynchronously in customize threadPool");
    return "success2";
}, threadPoolExecutor);
CompletableFuture<String> future = supply.applyToEither(supplyCustomize, a -> {
    System.out.println("acceptEither result " + a);
    return "appleTo " + a;
});

3.runAfterEither

对应runAfterBoth，两个CompletableFuture任一执行完成，就会继续下一步异步任务

/**
*	runnable 无入参无返回值
*/
CompletableFuture<String> supply = CompletableFuture.supplyAsync(() -> {
    try {
        TimeUnit.SECONDS.sleep(2);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    System.out.println("completablefuture supplys asynchronously");
    return "success1";
});
// 异步运行交由默认线程池（forkjoinpool）无入参有返回值supply
CompletableFuture<String> supplyCustomize = CompletableFuture.supplyAsync(() -> {
    System.out.println("completablefuture supplys asynchronously in customize threadPool");
    return "success2";
}, threadPoolExecutor);
CompletableFuture<Void> future = supply.runAfterEitherAsync(supplyCustomize, () -> {
    System.out.println("runAfterEither result ");
});

CompletableFuture之allOf|anyOf

1.allOf

所有完成get方法才能获得返回值，返回值是null，每个任务返回值需要调用每个CompletableFuture。

example：

/**
*	当所有任务完成后，返回null，具体每个任务返回值，需要调用具体
*	异步任务获取
*/
CompletableFuture<String> supply = CompletableFuture.supplyAsync(() -> {
    try {
        TimeUnit.SECONDS.sleep(2);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    System.out.println("completablefuture supplys asynchronously");
    return "success1";
});
CompletableFuture<String> supplyCustomize = CompletableFuture.supplyAsync(() -> {
    System.out.println("completablefuture supplys asynchronously in customize threadPool");
    return "success2";
}, threadPoolExecutor);
CompletableFuture<Void> allOf = CompletableFuture.allOf(supply, supplyCustomize);
allOf.get();
System.out.println(supplyCustomize.get() + "=>" + supply.get());

result：

completablefuture supplys asynchronously in customize threadPool
completablefuture supplys asynchronously
success2==success1

2.anyOf

/**
* 任一先执行完毕的结果将会先返回
*/
CompletableFuture<String> supply = CompletableFuture.supplyAsync(() -> {
try {
    TimeUnit.SECONDS.sleep(2);
} catch (InterruptedException e) {
    e.printStackTrace();
}
	System.out.println("completablefuture supplys asynchronously");
	return "success1";
});
// 异步运行交由默认线程池（forkjoinpool）无入参有返回值supply
CompletableFuture<Integer> supplyCustomize = CompletableFuture.supplyAsync(() -> {
	System.out.println("completablefuture supplys asynchronously in customize threadPool");
	return 2;
}, threadPoolExecutor);
CompletableFuture<Object> anyOf = CompletableFuture.anyOf(supply, supplyCustomize);
System.out.println(anyOf.get());

result：先执行完所以先返回2

completablefuture supplys asynchronously in customize threadPool 2 completablefuture supplys asynchronously

总结

CompletableFuture丰富的异步调用方法，可以帮助我们避免使用主线程来将多个异步任务连结，提高程序性能，加快响应速度。同时优秀的异常处理机制，在异步调用出错过程中也能完美解决。

end
  • 作者：Endwas（联系作者）
  • 发表时间：2021-10-31 01:18
  • 版权声明：自由转载-非商用-非衍生-保持署名（创意共享3.0许可证）
  • 转载声明：如果是转博主转载的文章，请附上原文链接
  • 公众号转载：请在文末添加作者名字和博客地址
    •

    评论