Posted on 一月 17, 2019
CompletableFuture 详解
一、Future模式
Java 1.5开始,提供了Callable和Future,通过它们可以在任务执行完毕之后得到任务执行结果。
Future接口可以构建异步应用,是多线程开发中常见的设计模式。
当我们需要调用一个函数方法时。如果这个函数执行很慢,那么我们就要进行等待。但有时候,我们可能并不急着要结果。
因此,我们可以让被调用者立即返回,让他在后台慢慢处理这个请求。对于调用者来说,则可以先处理一些其他任务,在真正需要数据的场合再去尝试获取需要的数据。
用法如下:
public static void main(String[] args) { ExecutorService executor = Executors.newCachedThreadPool(); Future<String> future=executor.submit(new Callable<String>() { @Override public String call() throws Exception { String result=new Random().nextLong()+""; Thread.sleep(2000); return result; } }); try { System.out.println(future.get()); } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.printStackTrace(); } }
Future以及相关使用方法提供了异步执行任务的能力,但对于结果的获取却是不方便,只能通过阻塞或轮询的方式得到任务结果。阻塞的方式与我们理解的异步编程其实是相违背的,而轮询又会耗无谓的CPU资源。而且还不能及时得到计算结果.
Future 接口的局限性
了解了Future的使用,这里就要谈谈Future的局限性。Future很难直接表述多个Future 结果之间的依赖性,开发中,我们经常需要达成以下目的:
将两个异步计算合并为一个(这两个异步计算之间相互独立,同时第二个又依赖于第一个的结果)
等待 Future 集合中的所有任务都完成。
仅等待 Future 集合中最快结束的任务完成,并返回它的结果。
CompletableFuture的作用
CompletableFuture是Java8的一个新加的类,它在原来的Future类上,结合Java8的函数式编程,扩展了一系列强大的功能.
CompletableFuture 的方法比较多;
Either
Either 表示的是两个CompletableFuture,当其中任意一个CompletableFuture计算完成的时候就会执行。
方法名 |
描述 |
acceptEither(CompletionStage<? extends T> other, Consumer<? super T> action) |
当任意一个CompletableFuture完成的时候,action这个消费者就会被执行。 |
acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action) |
当任意一个CompletableFuture完成的时候,action这个消费者就会被执行。使用ForkJoinPool |
acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action, Executor executor) |
当任意一个CompletableFuture完成的时候,action这个消费者就会被执行。使用指定的线程池 |
例子:
public static void main(String[] args) { Random random = new Random(); CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> { try { Thread.sleep(random.nextInt(1000)); } catch (InterruptedException e) { e.printStackTrace(); } return "from future1"; }); CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> { try { Thread.sleep(random.nextInt(1000)); } catch (InterruptedException e) { e.printStackTrace(); } return "from future2"; }); CompletableFuture<Void> future = future1.acceptEither(future2, str -> System.out.println("The future is " + str)); try { future.get(); } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.printStackTrace(); } }
执行结果:The future is from future1 或者 The future is from future2。
因为future1和future2,执行的顺序是随机的。
applyToEither 方法:
applyToEither 跟 acceptEither 类似。
方法名 |
描述 |
applyToEither(CompletionStage<? extends T> other, Function<? super T,U> fn) |
当任意一个CompletableFuture完成的时候,fn会被执行,它的返回值会当作新的CompletableFuture<U>的计算结果。 |
applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T,U> fn) |
当任意一个CompletableFuture完成的时候,fn会被执行,它的返回值会当作新的CompletableFuture<U>的计算结果。使用ForkJoinPool |
applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T,U> fn, Executor executor) |
当任意一个CompletableFuture完成的时候,fn会被执行,它的返回值会当作新的CompletableFuture<U>的计算结果。使用指定的线程池 |
例子省略。。。。。
allOf 方法
方法名 |
描述 |
allOf(CompletableFuture<?>… cfs) |
在所有Future对象完成后结束,并返回一个future。 |
allOf()方法所返回的CompletableFuture,并不能组合前面多个CompletableFuture的计算结果。于是我们借助Java 8的Stream来组合多个future的结果。
CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> "tony");
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> "cafei");
CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> "aaron");
CompletableFuture.allOf(future1, future2, future3) .thenApply(v -> Stream.of(future1, future2, future3) .map(CompletableFuture::join) .collect(Collectors.joining(" "))) .thenAccept(System.out::print);
|
执行结果:
tony cafei aaron
3.7.2 anyOf
方法名 |
描述 |
anyOf(CompletableFuture<?>… cfs) |
在任何一个Future对象结束后结束,并返回一个future。 |
Random rand = new Random(); CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> { try { Thread.sleep(rand.nextInt(1000)); } catch (InterruptedException e) { e.printStackTrace(); } return "from future1"; }); CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> { try { Thread.sleep(rand.nextInt(1000)); } catch (InterruptedException e) { e.printStackTrace(); } return "from future2"; }); CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> { try { Thread.sleep(rand.nextInt(1000)); } catch (InterruptedException e) { e.printStackTrace(); } return "from future3"; }); CompletableFuture<Object> future = CompletableFuture.anyOf(future1,future2,future3); try { System.out.println(future.get()); } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.printStackTrace(); }
使用anyOf()时,只要某一个future完成,就结束了。所以执行结果可能是"from future1"、"from future2"、"from future3"中的任意一个。
anyOf 和 acceptEither、applyToEither的区别在于,后两者只能使用在两个future中,而anyOf可以使用在多个future中。
异步方法:
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) { return asyncSupplyStage(asyncPool, supplier); } public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier,Executor executor) { return asyncSupplyStage(screenExecutor(executor), supplier); } public static CompletableFuture<Void> runAsync(Runnable runnable) { return asyncRunStage(asyncPool, runnable); } public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor) { return asyncRunStage(screenExecutor(executor), runnable); }
supplyAsync 返回Future有结果,而runAsync返回CompletableFuture<Void>。
Executor参数可以手动指定线程池,否则默认ForkJoinPool.commonPool()系统级公共线程池。
CompletableFuture<Void> cf = CompletableFuture.runAsync(() -> { |
在这段代码中,runAsync 是异步执行的 ,通过 Thread.currentThread().isDaemon() 打印的结果就可以知道是Daemon线程异步执行的。
同步执行方法:
public <U> CompletionStage<U> thenApply(Function<? super T,? extends U> fn); public CompletableFuture<Void> thenAccept(Consumer<? super T> action); public <U> CompletableFuture<Void> thenAcceptBoth(CompletionStage<? extends U> other, BiConsumer<? super T,? super U> action); public CompletableFuture<Void> thenRun(Runnable action); CompletableFuture<String> cf = CompletableFuture.completedFuture("message").thenApply(s -> { System.out.println(Thread.currentThread().isDaemon() + "_"+ s); try { TimeUnit.SECONDS.sleep(3); } catch (InterruptedException e) { e.printStackTrace(); } return s.toUpperCase(); }); System.out.println("done=" + cf.isDone()); TimeUnit.SECONDS.sleep(4); System.out.println("done=" + cf.isDone()); // 一直等待成功,然后返回结果 System.out.println(cf.join());
thenApply
当前阶段正常完成以后执行,而且当前阶段的执行的结果会作为下一阶段的输入参数。thenApplyAsync默认是异步执行的。这里所谓的异步指的是不在当前线程内执行。
thenApply相当于回调函数(callback)
thenAccept与thenRun
可以看到,thenAccept和thenRun都是无返回值的。如果说thenApply是不停的输入输出的进行生产,那么thenAccept和thenRun就是在进行消耗。它们是整个计算的最后两个阶段。
同样是执行指定的动作,同样是消耗,二者也有区别:
thenAccept接收上一阶段的输出作为本阶段的输入,同步执行;
thenRun根本不关心前一阶段的输出,根本不不关心前一阶段的计算结果,因为它不需要输入参数
CompletableFuture异常处理
exceptionally
方法名 |
描述 |
exceptionally(Function<Throwable,? extends T> fn) |
只有当CompletableFuture抛出异常的时候,才会触发这个exceptionally的计算,调用function计算值。 |
CompletableFuture.supplyAsync(() -> "hello world") .thenApply(s -> { s = null; int length = s.length(); return length; }).thenAccept(i -> System.out.println(i)) .exceptionally(t -> { System.out.println("Unexpected error:" + t); return null; });
执行结果:
Unexpected error:java.util.concurrent.CompletionException: java.lang.NullPointerException
对上面的代码稍微做了一下修改,修复了空指针的异常。
CompletableFuture.supplyAsync(() -> "hello world") .thenApply(s -> { // s = null; int length = s.length(); return length; }).thenAccept(i -> System.out.println(i)) .exceptionally(t -> { System.out.println("Unexpected error:" + t); return null; });
执行结果:
11
whenComplete
whenComplete 在上一篇文章其实已经介绍过了,在这里跟exceptionally的作用差不多,可以捕获任意阶段的异常。如果没有异常的话,就执行action。
CompletableFuture.supplyAsync(() -> "hello world") .thenApply(s -> { s = null; int length = s.length(); return length; }).thenAccept(i -> System.out.println(i)) .whenComplete((result, throwable) -> { if (throwable != null) { System.out.println("Unexpected error:"+throwable); } else { System.out.println(result); } });
执行结果:
Unexpected error:java.util.concurrent.CompletionException: java.lang.NullPointerException
跟whenComplete相似的方法是handle,handle的用法在上一篇文章中也已经介绍过。
方法名 |
描述 |
thenCompose(Function<? super T, ? extends CompletionStage<U>> fn) |
在异步操作完成的时候对异步操作的结果进行一些操作,并且仍然返回CompletableFuture类型。 |
thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn) |
在异步操作完成的时候对异步操作的结果进行一些操作,并且仍然返回CompletableFuture类型。使用ForkJoinPool。 |
thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn,Executor executor) |
在异步操作完成的时候对异步操作的结果进行一些操作,并且仍然返回CompletableFuture类型。使用指定的线程池。 |
thenCompose可以用于组合多个CompletableFuture,将前一个结果作为下一个计算的参数,它们之间存在着先后顺序。
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello") .thenCompose(s -> CompletableFuture.supplyAsync(() -> s + " World")); try { System.out.println(future.get()); } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.printStackTrace(); }
执行结果:
Hello World
下面的例子展示了多次调用thenCompose()
CompletableFuture<Double> future = CompletableFuture.supplyAsync(() -> "100") .thenCompose(s -> CompletableFuture.supplyAsync(() -> s + "100")) .thenCompose(s -> CompletableFuture.supplyAsync(() -> Double.parseDouble(s))); try { System.out.println(future.get()); } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.printStackTrace(); }
执行结果:
100100.0
组合
方法名 |
描述 |
thenCombine(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn) |
当两个CompletableFuture都正常完成后,执行提供的fn,用它来组合另外一个CompletableFuture的结果。 |
thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn) |
当两个CompletableFuture都正常完成后,执行提供的fn,用它来组合另外一个CompletableFuture的结果。使用ForkJoinPool。 |
thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn, Executor executor) |
当两个CompletableFuture都正常完成后,执行提供的fn,用它来组合另外一个CompletableFuture的结果。使用指定的线程池。 |
现在有CompletableFuture<T>、CompletableFuture<U>和一个函数(T,U)->V,thenCompose就是将CompletableFuture<T>和CompletableFuture<U>变为CompletableFuture<V>。
CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> "100"); CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> 100); CompletableFuture<Double> future = future1.thenCombine(future2, (s, i) -> Double.parseDouble(s + i)); try { System.out.println(future.get()); } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.printStackTrace(); }
执行结果:
100100.0
使用thenCombine()之后future1、future2之间是并行执行的,最后再将结果汇总。这一点跟thenCompose()不同。
thenAcceptBoth跟thenCombine类似,但是返回CompletableFuture<Void>类型。
方法名 |
描述 |
thenAcceptBoth(CompletionStage<? extends U> other, BiConsumer<? super T,? super U> action) |
当两个CompletableFuture都正常完成后,执行提供的action,用它来组合另外一个CompletableFuture的结果。 |
thenAcceptBothAsync(CompletionStage<? extends U> other, BiConsumer<? super T,? super U> action) |
当两个CompletableFuture都正常完成后,执行提供的action,用它来组合另外一个CompletableFuture的结果。使用ForkJoinPool。 |
thenAcceptBothAsync(CompletionStage<? extends U> other, BiConsumer<? super T,? super U> action, Executor executor) |
当两个CompletableFuture都正常完成后,执行提供的action,用它来组合另外一个CompletableFuture的结果。使用指定的线程池。 |
CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> "100"); CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> 100); CompletableFuture<Void> future = future1.thenAcceptBoth(future2, (s, i) -> System.out.println(Double.parseDouble(s + i))); try { future.get(); } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.printStackTrace(); }
执行结果:
100100.0
3.5 计算结果完成时的处理
当CompletableFuture完成计算结果后,我们可能需要对结果进行一些处理。
3.5.1 执行特定的Action
方法名 |
描述 |
whenComplete(BiConsumer<? super T,? super Throwable> action) |
当CompletableFuture完成计算结果时对结果进行处理,或者当CompletableFuture产生异常的时候对异常进行处理。 |
whenCompleteAsync(BiConsumer<? super T,? super Throwable> action) |
当CompletableFuture完成计算结果时对结果进行处理,或者当CompletableFuture产生异常的时候对异常进行处理。使用ForkJoinPool。 |
whenCompleteAsync(BiConsumer<? super T,? super Throwable> action, Executor executor) |
当CompletableFuture完成计算结果时对结果进行处理,或者当CompletableFuture产生异常的时候对异常进行处理。使用指定的线程池。 |
CompletableFuture.supplyAsync(() -> "Hello")
.thenApply(s->s+" World")
.thenApply(s->s+ "\nThis is CompletableFuture demo")
.thenApply(String::toLowerCase)
.whenComplete((result, throwable) -> System.out.println(result));
执行结果:
hello world
this is completablefuture demo
3.5.2 执行完Action可以做转换
方法名 |
描述 |
handle(BiFunction<? super T, Throwable, ? extends U> fn) |
当CompletableFuture完成计算结果或者抛出异常的时候,执行提供的fn |
handleAsync(BiFunction<? super T, Throwable, ? extends U> fn) |
当CompletableFuture完成计算结果或者抛出异常的时候,执行提供的fn,使用ForkJoinPool。 |
handleAsync(BiFunction<? super T, Throwable, ? extends U> fn, Executor executor) |
当CompletableFuture完成计算结果或者抛出异常的时候,执行提供的fn,使用指定的线程池。 |
CompletableFuture<Double> future = CompletableFuture.supplyAsync(() -> "100")
.thenApply(s->s+"100")
.handle((s, t) -> s != null ? Double.parseDouble(s) : 0);
try {
System.out.println(future.get());
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
执行结果:
100100.0
在这里,handle()的参数是BiFunction,apply()方法返回R,相当于转换的操作。
@FunctionalInterface
public interface BiFunction<T, U, R> {
/**
* Applies this function to the given arguments.
*
* @param t the first function argument
* @param u the second function argument
* @return the function result
*/
R apply(T t, U u);
/**
* Returns a composed function that first applies this function to
* its input, and then applies the {@code after} function to the result.
* If evaluation of either function throws an exception, it is relayed to
* the caller of the composed function.
*
* @param <V> the type of output of the {@code after} function, and of the
* composed function
* @param after the function to apply after this function is applied
* @return a composed function that first applies this function and then
* applies the {@code after} function
* @throws NullPointerException if after is null
*/
default <V> BiFunction<T, U, V> andThen(Function<? super R, ? extends V> after) {
Objects.requireNonNull(after);
return (T t, U u) -> after.apply(apply(t, u));
}
}
而whenComplete()的参数是BiConsumer,accept()方法返回void。
@FunctionalInterface
public interface BiConsumer<T, U> {
/**
* Performs this operation on the given arguments.
*
* @param t the first input argument
* @param u the second input argument
*/
void accept(T t, U u);
/**
* Returns a composed {@code BiConsumer} that performs, in sequence, this
* operation followed by the {@code after} operation. If performing either
* operation throws an exception, it is relayed to the caller of the
* composed operation. If performing this operation throws an exception,
* the {@code after} operation will not be performed.
*
* @param after the operation to perform after this operation
* @return a composed {@code BiConsumer} that performs in sequence this
* operation followed by the {@code after} operation
* @throws NullPointerException if {@code after} is null
*/
default BiConsumer<T, U> andThen(BiConsumer<? super T, ? super U> after) {
Objects.requireNonNull(after);
return (l, r) -> {
accept(l, r);
after.accept(l, r);
};
}
}
所以,handle()相当于whenComplete()+转换。
3.5.3 纯消费(执行Action)
方法名 |
描述 |
thenAccept(Consumer<? super T> action) |
当CompletableFuture完成计算结果,只对结果执行Action,而不返回新的计算值 |
thenAcceptAsync(Consumer<? super T> action) |
当CompletableFuture完成计算结果,只对结果执行Action,而不返回新的计算值,使用ForkJoinPool。 |
thenAcceptAsync(Consumer<? super T> action, Executor executor) |
当CompletableFuture完成计算结果,只对结果执行Action,而不返回新的计算值 |
thenAccept()是只会对计算结果进行消费而不会返回任何结果的方法。
CompletableFuture.supplyAsync(() -> "Hello")
.thenApply(s->s+" World")
.thenApply(s->s+ "\nThis is CompletableFuture demo")
.thenApply(String::toLowerCase)
.thenAccept(System.out::print);
执行结果:
hello world
this is completablefuture demo