java8 CompletableFuture 异步执行
CompletableFuture类实现了CompletionStage和Future接口。Future是Java 5添加的类,用来描述一个异步计算的结果,但是获取一个结果时方法较少,要么通过轮询isDone,确认完成后,调用get()获取值,要么调用get()设置一个超时时间。但是这个get()方法会阻塞住调用线程,这种阻塞的方式显然和我们的异步编程的初衷相违背。 为了解决这个问题,JDK吸收了guava的设计思想,加入了Future的诸多扩展功能形成了CompletableFuture。 CompletionStage是一个接口,从命名上看得知是一个完成的阶段,它里面的方法也标明是在某个运行阶段得到了结果之后要做的事情 。
1.进行变换
public <U> CompletionStage<U> thenApply(Function<? super T,? extends U> fn); public <U> CompletionStage<U> thenApplyAsync(Function<? super T,? extends U> fn); public <U> CompletionStage<U> thenApplyAsync(Function<? super T,? extends U> fn,Executor executor);
首先说明一下已Async结尾的方法都是可以异步执行的,如果指定了线程池,会在指定的线程池中执行,如果没有指定,默认会在ForkJoinPool.commonPool()中执行,下文中将会有好多类似的,都不详细解释了。关键的入参只有一个Function,它是函数式接口,所以使用Lambda表示起来会更加优雅。它的入参是上一个阶段计算后的结果,返回值是经过转化后结果。
例如.
@Test public void thenApply() { String result = CompletableFuture.supplyAsync(() -> "hello").thenApply(s -> s + " world").join(); System.out.println(result); }
输出:
hello world
接下来看一下获取执行结果的几个方法.
V get();
V get(long timeout,Timeout unit);
T getNow(T defaultValue);
T join();
上面两个方法是Future中的实现方式,get()会堵塞当前的线程,这就造成了一个问题,如果执行线程迟迟没有返回数据,get()会一直等待下去,因此,第二个get()方法可以设置等待的时间。
getNow()方法比较有意思,表示当有了返回结果时会返回结果,如果异步线程抛了异常会返回自己设置的默认值。
thenCompose 和 thenApply 差不多只是thenCompose 返回一个新的CompletableFuture,而 thenApply 返回还是那一个 thenApply 。相同点是,在异步操作完成的时候对异步操作的结果进行一些操作。
CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> { return 100; }); CompletableFuture<String> f = future.thenCompose( i -> { return CompletableFuture.supplyAsync(() -> { return (i * 10) + ""; }); }); System.out.println(f.get()); //1000
2.进行消耗
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);
thenAccept是针对结果进行消耗,因为他的入参是Consumer,有入参无返回值。
例如:
@Test public void thenAccept(){ CompletableFuture.supplyAsync(() -> "hello").thenAccept(s -> System.out.println(s+" world")); }
输出:
hello world
3.对上一步的计算结果不关心,执行下一个操作。
public CompletionStage<Void> thenRun(Runnable action); public CompletionStage<Void> thenRunAsync(Runnable action); public CompletionStage<Void> thenRunAsync(Runnable action,Executor executor);
thenRun它的入参是一个Runnable的实例,表示当得到上一步的结果时的操作。
例如:
@Test public void thenRun(){ CompletableFuture.supplyAsync(() -> { try { Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); } return "hello"; }).thenRun(() -> System.out.println("hello world")); while (true){} }
输出:
hello world
4.结合两个CompletionStage的结果,进行转化后返回
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);
它需要原来的处理返回值,并且other代表的CompletionStage也要返回值之后,利用这两个返回值,进行转换后返回指定类型的值。
例如:
@Test public void thenCombine() { String result = CompletableFuture.supplyAsync(() -> { try { Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); } return "hello"; }).thenCombine(CompletableFuture.supplyAsync(() -> { try { Thread.sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } return "world"; }), (s1, s2) -> s1 + " " + s2).join(); System.out.println(result); }
输出:
hello world
5.结合两个CompletionStage的结果,进行消耗
public <U> CompletionStage<Void> thenAcceptBoth(CompletionStage<? extends U> other,BiConsumer<? super T, ? super U> action); public <U> CompletionStage<Void> thenAcceptBothAsync(CompletionStage<? extends U> other,BiConsumer<? super T, ? super U> action); public <U> CompletionStage<Void> thenAcceptBothAsync(CompletionStage<? extends U> other,BiConsumer<? super T, ? super U> action, Executor executor);
它需要原来的处理返回值,并且other代表的CompletionStage也要返回值之后,利用这两个返回值,进行消耗。
例如:
@Test public void thenAcceptBoth() { CompletableFuture.supplyAsync(() -> { try { Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); } return "hello"; }).thenAcceptBoth(CompletableFuture.supplyAsync(() -> { try { Thread.sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } return "world"; }), (s1, s2) -> System.out.println(s1 + " " + s2)); while (true){} }
输出:
hello world
6.在两个CompletionStage都运行完执行。
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);
不关心这两个CompletionStage的结果,只关心这两个CompletionStage执行完毕,之后在进行操作(Runnable)。
例如:
@Test public void runAfterBoth(){ CompletableFuture.supplyAsync(() -> { try { Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); } return "s1"; }).runAfterBothAsync(CompletableFuture.supplyAsync(() -> { try { Thread.sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } return "s2"; }), () -> System.out.println("hello world")); while (true){} }
输出:
hello world
7.两个CompletionStage,谁计算的快,我就用那个CompletionStage的结果进行下一步的转化操作。
public <U> CompletionStage<U> applyToEither(CompletionStage<? extends T> other,Function<? super T, U> fn); public <U> CompletionStage<U> applyToEitherAsync(CompletionStage<? extends T> other,Function<? super T, U> fn); public <U> CompletionStage<U> applyToEitherAsync(CompletionStage<? extends T> other,Function<? super T, U> fn,Executor executor);
我们现实开发场景中,总会碰到有两种渠道完成同一个事情,所以就可以调用这个方法,找一个最快的结果进行处理。
例如:
@Test public void applyToEither() { String result = CompletableFuture.supplyAsync(() -> { try { Thread.sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } return "s1"; }).applyToEither(CompletableFuture.supplyAsync(() -> { try { Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); } return "hello world"; }), s -> s).join(); System.out.println(result); }
输出:
hello world
8.两个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);
例如:
@Test public void acceptEither() { CompletableFuture.supplyAsync(() -> { try { Thread.sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } return "s1"; }).acceptEither(CompletableFuture.supplyAsync(() -> { try { Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); } return "hello world"; }), System.out::println); while (true){} }
输出:
hello world
9.两个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);
例如:
@Test public void runAfterEither() { CompletableFuture.supplyAsync(() -> { try { Thread.sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } return "s1"; }).runAfterEither(CompletableFuture.supplyAsync(() -> { try { Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); } return "s2"; }), () -> System.out.println("hello world")); while (true) { } }
输出:
hello world
10.当运行时出现了异常,可以通过exceptionally进行补偿。
public CompletionStage<T> exceptionally(Function<Throwable, ? extends T> fn);
例如:
@Test public void exceptionally() { String result = CompletableFuture.supplyAsync(() -> { try { Thread.sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } if (1 == 1) { throw new RuntimeException("测试一下异常情况"); } return "s1"; }).exceptionally(e -> { System.out.println(e.getMessage()); return "hello world"; }).join(); System.out.println(result); }
输出:
java.lang.RuntimeException: 测试一下异常情况 hello world
11.当运行完成时,对结果的记录。这里的完成时有两种情况,一种是正常执行,返回值。另外一种是遇到异常抛出造成程序的中断。这里为什么要说成记录,因为这几个方法都会返回CompletableFuture,当Action执行完毕后它的结果返回原始的CompletableFuture的计算结果或者返回异常。所以不会对结果产生任何的作用。
public CompletionStage<T> whenComplete(BiConsumer<? super T, ? super Throwable> action); public CompletionStage<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action); public CompletionStage<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action,Executor executor);
例如:
@Test public void whenComplete() { String result = CompletableFuture.supplyAsync(() -> { try { Thread.sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } if (1 == 1) { throw new RuntimeException("测试一下异常情况"); } return "s1"; }).whenComplete((s, t) -> { System.out.println(s); System.out.println(t.getMessage()); }).exceptionally(e -> { System.out.println(e.getMessage()); return "hello world"; }).join(); System.out.println(result); }
输出:
null java.lang.RuntimeException: 测试一下异常情况 java.lang.RuntimeException: 测试一下异常情况 hello world
这里也可以看出,如果使用了exceptionally,就会对最终的结果产生影响,它没有口子返回如果没有异常时的正确的值,这也就引出下面我们要介绍的handle。
12.运行完成时,对结果的处理。这里的完成时有两种情况,一种是正常执行,返回值。另外一种是遇到异常抛出造成程序的中断。
public <U> CompletionStage<U> handle(BiFunction<? super T, Throwable, ? extends U> fn); public <U> CompletionStage<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn); public <U> CompletionStage<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn,Executor executor);
例如:
出现异常时
@Test public void handle() { String result = CompletableFuture.supplyAsync(() -> { try { Thread.sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } //出现异常 if (1 == 1) { throw new RuntimeException("测试一下异常情况"); } return "s1"; }).handle((s, t) -> { // 不为null 则有异常 if (t != null) { return "hello world"; } return s; }).join(); System.out.println(result); }
输出:
hello world
未出现异常时
@Test public void handle() { String result = CompletableFuture.supplyAsync(() -> { // 抛出异常就没有异常了 try { Thread.sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } return "s1"; }).handle((s, t) -> { if (t != null) { return "hello world"; } return s; }).join(); System.out.println(result); }
结果为:
s1
13.allOf 参数是多个CompletableFuture对象 我们在处理业务时,有时会有多任务异步处理,同步返回结果的情况,在java中,我们可以使用CompletableFuture的allOf方法来实现多实例的同时返回
在这里我们可以将对各future实例添加到allOf方法中,然后通过future的get()方法获取future的状态。如果allOf里面的所有线程为执行完毕,主线程会阻塞,直到allOf里面的所有线程都执行,线程就会被唤醒。
CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> { try { Thread.sleep(10); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("future1 finished!"); return "future1 finished!"; }); CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> { System.out.println("future2 finished!"); return "future2 finished!"; }); CompletableFuture<Void> combindFuture = CompletableFuture.allOf(future1, future2); try { //get() 获取future 的状态 combindFuture.get(); System.out.println("状态:"+combindFuture); } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.printStackTrace(); } System.out.println("future1: " + future1.isDone() + " future2: " + future2.isDone());
14:.handle 该方法可以返回一个新的CompletableFuture的返回类型。
CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> { System.out.println("hello world"); return "result"; }); CompletableFuture<Integer> f211 = f1.handle((r, e) -> { try { System.out.println("handle"+f1.get()); } catch (InterruptedException e1) { e1.printStackTrace(); } catch (ExecutionException e1) { e1.printStackTrace(); } return 2; }); System.out.println("新的CompletableFuture:"+f211);
15.join 和 get 区别是 join不用抛异常 get 得抛异常(还是get 好)
join: public static void thenApplyAsyncExample() { CompletableFuture cf = CompletableFuture.completedFuture("message").thenApplyAsync(s -> { return s.toUpperCase(); }); //join 和 get 区别是 join不用抛异常 get 得抛异常(还是get 好) String str = (String) cf.join(); System.out.println(str); } get: public static void thenApplyAsyncExample() { CompletableFuture cf = CompletableFuture.completedFuture("message").thenApplyAsync(s -> { return s.toUpperCase(); }); //join 和 get 区别是 join不用抛异常 get 得抛异常(还是get 好) try { String o = (String) cf.get(); } catch (InterruptedException e) { } catch (ExecutionException e) { e.printStackTrace(); } System.out.println(o); }
16.completeExceptionally如果执行caculatePrice方法出现异常,会导致调用CompletableFuture的get方法永远等待下去。异常被限制在计算商品价格的线程范围,最后会杀死该线程,所以调用Future的get会一直等待。
当然,你可以使用带超时参数的get的重载方法来避免无限期等待,最终抛出一个TimeoutException,但这样你并不知道执行任务的线程到底出现了什么问题。
为了解决这个问题,你可以使用CompletableFuture的completeExceptionally将异常抛出
public Future<Double> getPriceAsync(String product) { CompletableFuture<Double> future = new CompletableFuture<>(); new Thread(){ @Override public void run() { double price = caculatePrice(product); try { // 这里对0做除法并不会抛出异常,在不在线程中使用会抛出异常 //System.out.println(price / 0); // 数组越界会抛出异常 int[] arr = {}; System.out.println(arr[1]); future.complete(price); } catch (Exception e) { future.completeExceptionally(e); } } }.start(); return future; }
比如,上面我们认为模拟了一个数组下标越界的异常。调用future的get方法的客户端会立即得到一个异常,get方法就不会无限期等待了。另外,调用get方法时,始终建议使用带超时参数的重载方法,避免无限等待。
另外,注意:CompletableFuture的所有异步方法都使用了同样的错误管理机制,你不用再花大力气去处理异常了。
上面就是CompletionStage接口中方法的使用实例,CompletableFuture同样也同样实现了Future,所以也同样可以使用get进行阻塞获取值,总的来说,CompletableFuture使用起来还是比较爽的,看起来也比较优雅一点。如有不足请多多指教。
- Java 程序优化 缩短程序执行时间 同步查询改异步查询 CompletableFuture supplyAsync
- Java中使用Future获取线程异步执行结果的使用
- 一心多用多线程-future-java线程中的异步执行
- java Future 异步程执行获取结果
- JAVA 并发编程-返回执行结果(Callable和Future)(九)
- Java 如何在Main函数中,执行完异步任务后才退出主线程
- java中异步计算之Future
- java异步计算Future的使用
- Java CompletableFuture
- 20个Java CompletableFuture的例子
- Java实现限定方法执行时间-java 1.5以上的Future类可以执行超时处理
- java异步计算Future的使用(转)
- java异步执行业务逻辑
- Java并发编程实践:Callable异步回调Future、FutureTask用法
- Java程序执行超时――Future接口介绍
- Java 异步线程FutureTask的使用和SwingWorker
- Java程序执行超时——Future接口介绍
- Jsp 中JavaScript 和 Java代码的异步执行特点
- java利用线程池(ExecutorService)配合Callable和Future实现执行方法超时的阻断
- java中 synchronized 的使用,确保异步执行某一段代码。