CompletableFuture的使用例子
Posted therhyme
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了CompletableFuture的使用例子相关的知识，希望对你有一定的参考价值。
1. CompletableFuture的介绍
在Java8时被引入，在包java.util.concurrent下，是Java多线程编程中的一个类，扩展了Future中很多功能，CompletableFuture是一个实现了接口Future和CompletionStage的类。
public class CompletableFuture<T> implements Future<T>, CompletionStage<T>
2. Future与CompletableFuture对比
1 Future#get阻塞方法，影响后续代码执行，CompletableFuture可以设置callback的方式处理：CompletableFuture#thenAcceptAsync
2 CompletableFuture可以组合多个CompletableFuture：CompletableFuture#thenCompose、anyof
3 CompletableFuture优雅处理线程异常：CompletableFuture#handle、exceptionally
4 CompletableFuture可以手动设置为完成，即一个线程处理任务的时间过长，可以手动设置为完成，并设置返回值：CompletableFuture#complete
3. CompletableFuture常用方法
3.1. CompletableFuture#runAsync
/** * @see CompletableFuture#runAsync(Runnable) 接收一个Runnable参数 */ @Test public void runAsyncTest() { CompletableFuture<Void> voidCompletableFuture = CompletableFuture.runAsync(() -> { System.out.println(getCurrentThreadName() + "------runSync method"); }); // ForkJoinPool.commonPool-worker-9------runSync method }
3.2. CompletableFuture#supplyAsync
/** * @throws ExecutionException * @throws InterruptedException * @see CompletableFuture#supplyAsync(Supplier) 接受一个Supplier参数 */ @Test public void supplyAsyncTest() throws ExecutionException, InterruptedException { CompletableFuture<String> stringCompletableFuture = CompletableFuture.supplyAsync(() -> "return a string of supplyAsync method"); System.out.println(stringCompletableFuture.get()); // return a string of supplyAsync method }
3.3. CompletableFuture#thenAccept
/** * @see CompletableFuture#thenAccept(Consumer) runAync或者supplyAsync执行完后进行的操作(callback) */ @Test public void thenAcceptTest() { CompletableFuture<Void> voidCompletableFuture = CompletableFuture.runAsync(() -> { System.out.println(getCurrentThreadName() + "------runSync method"); }).thenAccept((consumer) -> { System.out.println(getCurrentThreadName() + "------thenAccept method"); }); System.out.println(getCurrentThreadName() + " End"); // ForkJoinPool.commonPool-worker-9------runSync method // main------thenAccept method // main End }
3.4. CompletableFuture#thenAcceptAsync
/** * @see CompletableFuture#thenAcceptAsync(Consumer) 异步callback */ @Test public void thenAcceptAsyncTest() throws InterruptedException { AtomicReference<Thread> threadAtomicReference = new AtomicReference<>(); CompletableFuture<Void> voidCompletableFuture = CompletableFuture.runAsync(() -> { threadAtomicReference.set(Thread.currentThread()); System.out.println(getCurrentThreadName() + "------runSync method"); }).thenAcceptAsync((consumer) -> { System.out.println(getCurrentThreadName() + "------thenAccept method"); getCurrentThread().notifyAll(); }); TimeUnit.SECONDS.sleep(1); System.out.println(getCurrentThreadName() + " End"); // ForkJoinPool.commonPool-worker-9------runSync method // ForkJoinPool.commonPool-worker-9------thenAccept method // main End }
3.5. CompletableFuture#thenApply
/** * @see CompletableFuture#thenApply(Function) */ @Test public void thenApply() { CompletableFuture<Void> voidCompletableFuture = CompletableFuture.supplyAsync(() -> "supplyAsync method") .thenApply((s) -> s + " -> theApply method") .thenAccept(System.out::println); // supplyAsync method -> theApply method }
3.6. CompletableFuture#thenCompose
/** * @see CompletableFuture#thenCompose(Function) 作为两个CompletableFuture组合使用 这里的Function<? super T, ? * extends CompletionStage<U>> T要转换成CompletionStage对应的子类? extends * CompletionStage<U>,比如另一个CompletableFuture#supplyAsync返回值就是. */ @Test public void thenComposeTest() { CompletableFuture.supplyAsync(() -> getCurrentThreadName() + ": supplyAsync -> ") .thenCompose( (s) -> CompletableFuture.supplyAsync( () -> s + getCurrentThreadName() + "---theApply method -> ")) .thenAccept((f) -> System.out.println(f + getCurrentThreadName() + "---thenAccept")); // ForkJoinPool.commonPool-worker-9: supplyAsync -> ForkJoinPool.commonPool-worker-9---theApply // method -> main---thenAccept }
3.7. CompletableFuture#thenCombine
/** * @see CompletableFuture#thenCombine(CompletionStage, BiFunction) * 等两个CompletableFuture完成后，对它们的返回值进行处理，也是对多个CompletableFuture进行组合 */ @Test public void thenCombineTest() { CompletableFuture.supplyAsync(() -> getCurrentThreadName() + ": the first return value") .thenCombine( CompletableFuture.supplyAsync( () -> getCurrentThreadName() + ": the second return value"), (p1, p2) -> { if (StringUtils.hasText(p1) && StringUtils.hasText(p2)) { return p1 + " " + p2 + " "; } else if (StringUtils.hasText(p1)) { return p1; } else if (StringUtils.hasText(p2)) { return p2; } else { return null; } }) .thenAccept((f) -> System.out.println(f + getCurrentThreadName() + "---thenAccept")); // ForkJoinPool.commonPool-worker-9: the first return value // ForkJoinPool.commonPool-worker-9: the second return value // main---thenAccept }
3.8. CompletableFuture#allOf
/** * 多个CompletableFuture，如果都结束了，就可以获得它们的返回值，进行处理 * * @throws InterruptedException * @see CompletableFuture#allOf(CompletableFuture[]) */ @Test public void allOfTest() throws InterruptedException { CompletableFuture<String> diligent = CompletableFuture.supplyAsync( () -> { try { TimeUnit.MICROSECONDS.sleep(100); return getCurrentThreadName() + ": be a diligent man."; } catch (InterruptedException e) { e.printStackTrace(); return getCurrentThreadName() + ": missing diligent"; } }); CompletableFuture<String> studious = CompletableFuture.supplyAsync( () -> { try { TimeUnit.MICROSECONDS.sleep(100); return getCurrentThreadName() + ": be a studious man."; } catch (InterruptedException e) { e.printStackTrace(); return getCurrentThreadName() + ": missing studious"; } }); CompletableFuture<String> savvy = CompletableFuture.supplyAsync( () -> { try { TimeUnit.MICROSECONDS.sleep(100); return getCurrentThreadName() + ": be a savvy man."; } catch (InterruptedException e) { e.printStackTrace(); return getCurrentThreadName() + ": missing savvy"; } }); CompletableFuture<List<String>> allOf = CompletableFuture.allOf(savvy, diligent, studious) .thenApply( (n) -> { return Stream.of(savvy, diligent, studious) .map( (completableFuture) -> { try { // get every return string return completableFuture.get(); } catch (InterruptedException | ExecutionException e) { e.printStackTrace(); return e.toString(); } }) .collect(Collectors.toList()); }); TimeUnit.SECONDS.sleep(1); // if you want to output every string (and above code of block of "thenApply" also can // forEach(System.out::println)) allOf.thenAccept((stringList) -> stringList.forEach(System.out::println)); // ForkJoinPool.commonPool-worker-11: be a savvy man. // ForkJoinPool.commonPool-worker-9: be a diligent man. // ForkJoinPool.commonPool-worker-2: be a studious man. }
3.9. CompletableFuture#anyOf
/** * anyOf方法，组合多个future，只要有一个结束就完成 * * @see CompletableFuture#anyOf(CompletableFuture[]) */ @Test public void anyOfTest() { CompletableFuture<People> peopleCompletableFuture = CompletableFuture.supplyAsync(People::new); CompletableFuture<String> stringCompletableFuture = CompletableFuture.supplyAsync(() -> "a string"); CompletableFuture<Integer> integerCompletableFuture = CompletableFuture.supplyAsync(() -> 1); CompletableFuture.anyOf( stringCompletableFuture, peopleCompletableFuture, integerCompletableFuture) .thenAccept(System.out::println); }
3.10. CompletableFuture#handle
/** * handle方法无论是否发生异常，都会调用,可以在这里处理异常,另一个处理异常的方式：{@link #exceptionallyTest()} * * @see CompletableFuture#handle(BiFunction) * @throws InterruptedException */ @Test public void handleTest() throws InterruptedException { final String string = ""; CompletableFuture.supplyAsync( () -> { if (!StringUtils.hasText(string)) { throw new NullPointerException("string is null"); } return string; }) .handle( (s, t) -> { if (t != null) { // log.error("handle method", t); log.error("handle method"); } return s; }); // For junit main thread stop after ForkJoinPool thread TimeUnit.SECONDS.sleep(1); System.out.println(getCurrentThreadName() + " stop"); // 18:30:47.837 [ForkJoinPool.commonPool-worker-9] ERROR // com.xy.java.basic.demos.completablefuture.CompletableFutureTest - handle method // main stop }
3.11. Completable#exceptionally
/** * 当发生异常时，进入exceptionally方法，另一个处理异常的方式：{@link #handleTest()} * * @see CompletableFuture#exceptionally(Function) * @throws InterruptedException * @throws ExecutionException */ @Test public void exceptionallyTest() throws InterruptedException, ExecutionException { final String string = ""; CompletableFuture.supplyAsync( () -> { if (!StringUtils.hasText(string)) { throw new NullPointerException("string is null"); } return Optional.ofNullable(string); }) .exceptionally( (t) -> { // log.error("exceptionally method", t); log.error("exceptionally method"); return Optional.empty(); }); // For junit main thread stop after ForkJoinPool thread TimeUnit.SECONDS.sleep(1); System.out.println(getCurrentThreadName() + " stop"); // 18:27:29.451 [ForkJoinPool.commonPool-worker-9] ERROR // com.xy.java.basic.demos.completablefuture.CompletableFutureTest - exceptionally method // main stop }
3.12. CompletableFuture#complete
/** * 手动完成一个耗时很长的Future，并且设置默认值 * * @throws InterruptedException * @throws ExecutionException */ @Test public void completeTest() throws InterruptedException, ExecutionException { CompletableFuture<Boolean> runAsync = CompletableFuture.supplyAsync( () -> { try { TimeUnit.SECONDS.sleep(3); return true; } catch (InterruptedException e) { e.printStackTrace(); return null; } }); TimeUnit.SECONDS.sleep(1); // 手动设置Future为完成状态并设置默认值 runAsync.complete(false); System.out.println(runAsync.get()); // false TimeUnit.SECONDS.sleep(5); }
3.13. 注：Junit测试类中的公共方法
private Thread getCurrentThread() { return Thread.currentThread(); } private String getCurrentThreadName() { return getCurrentThread().getName(); }
4. thenApply与thenCompose的区别
4.1. 用thenApply方法来组合两个CompletableFuture
/** * @see CompletableFuture#thenApply(Function) * 如果想用thenApply方法来组合两个CompletableFuture，看起来会非常不优雅，所以组合多个CompletableFuture推荐使用<b>CompletableFuture#thenCompose</b> */ @Test public void thenApplyNeedReturnCompletionStageTest() { CompletableFuture<Void> voidCompletableFuture = CompletableFuture.supplyAsync(() -> getCurrentThreadName() + "---supplyAsync method -> ") .thenApply( (s) -> CompletableFuture.supplyAsync( () -> s + getCurrentThreadName() + "---theApply method -> ")) .thenAccept( (c) -> c.thenAccept( (f) -> System.out.println(f + getCurrentThreadName() + "---thenAccept"))); // ForkJoinPool.commonPool-worker-9---supplyAsync method -> // ForkJoinPool.commonPool-worker-9---theApply method -> main---thenAccept }
4.2. 用thenCompose方法来组合两个CompletableFuture
/** * @see CompletableFuture#thenCompose(Function) 作为两个CompletableFuture组合使用 这里的Function<? super T, ? * extends CompletionStage<U>> T要转换成CompletionStage对应的子类? extends * CompletionStage<U>,比如另一个CompletableFuture#supplyAsync返回值就是. */ @Test public void thenComposeTest() { CompletableFuture.supplyAsync(() -> getCurrentThreadName() + ": supplyAsync -> ") .thenCompose( (s) -> CompletableFuture.supplyAsync( () -> s + getCurrentThreadName() + "---theApply method -> ")) .thenAccept((f) -> System.out.println(f + getCurrentThreadName() + "---thenAccept")); // ForkJoinPool.commonPool-worker-9: supplyAsync -> ForkJoinPool.commonPool-worker-9---theApply // method -> main---thenAccept }
5. CompletableFuture常用方法总结
1 runAsync接收的Runnable参数，supplyAsync接收的Supplier参数
2 thenAccept与thenAcceptAsync的区别在于：该callback方法是否在当前线程中执行(更具体的例子见前面的代码中的运行结果)
3 thenApply与thenCompose主要区别在于组合多个CompletableFuture
4 其他的方法如上面的代码例子所示
以上是关于CompletableFuture的使用例子的主要内容，如果未能解决你的问题，请参考以下文章
20个使用 Java CompletableFuture的例子