基础篇(异步编程不会(我教你啊!CompletableFuture(JDK1.8)))

前言 以前需要异步执行一个任务时,一般是用Thread或者线程池Executor去创建。如果需要返回值,则是调用Executor.submit获取Future。但是多个线程存在依赖组合,我们又能怎么办?可使用同步组件CountDownLatch、CyclicBarrier等;其实有简单的方法,就是用CompeletableFuture

  • 线程任务的创建
  • 线程任务的串行执行
  • 线程任务的并行执行
  • 处理任务结果和异常
  • 多任务的简单组合
  • 取消执行线程任务
  • 任务结果的获取和完成与否判断
关注公众号,一起交流,微信搜一搜: 潜行前行 基础篇(异步编程不会(我教你啊!CompletableFuture(JDK1.8)))
文章图片

github地址,感谢star 1 创建异步线程任务 根据supplier创建CompletableFuture任务
//使用内置线程ForkJoinPool.commonPool(),根据supplier构建执行任务 public static CompletableFuture supplyAsync(Supplier supplier) //指定自定义线程,根据supplier构建执行任务 public static CompletableFuture supplyAsync(Supplier supplier, Executor executor)

根据runnable创建CompletableFuture任务
//使用内置线程ForkJoinPool.commonPool(),根据runnable构建执行任务 public static CompletableFuture runAsync(Runnable runnable) //指定自定义线程,根据runnable构建执行任务 public static CompletableFuture runAsync(Runnable runnable, Executor executor)

  • 使用示例
    ExecutorService executor = Executors.newSingleThreadExecutor(); CompletableFuture rFuture = CompletableFuture .runAsync(() -> System.out.println("hello siting"), executor); //supplyAsync的使用 CompletableFuture future = CompletableFuture .supplyAsync(() -> { System.out.print("hello "); return "siting"; }, executor); //阻塞等待,runAsync 的future 无返回值,输出null System.out.println(rFuture.join()); //阻塞等待 String name = future.join(); System.out.println(name); executor.shutdown(); // 线程池需要关闭 --------输出结果-------- hello siting null hello siting

    常量值作为CompletableFuture返回
    //有时候是需要构建一个常量的CompletableFuture public static CompletableFuture completedFuture(U value)

2 线程串行执行 基础篇(异步编程不会(我教你啊!CompletableFuture(JDK1.8)))
文章图片

任务完成则运行action,不关心上一个任务的结果,无返回值
public CompletableFuture thenRun(Runnable action) public CompletableFuture thenRunAsync(Runnable action) //action用指定线程池执行 public CompletableFuture thenRunAsync(Runnable action, Executor executor)

  • 使用示例
    CompletableFuture future = CompletableFuture .supplyAsync(() -> "hello siting", executor) .thenRunAsync(() -> System.out.println("OK"), executor); executor.shutdown(); --------输出结果-------- OK

任务完成则运行action,依赖上一个任务的结果,无返回值
public CompletableFuture thenAccept(Consumer action) public CompletableFuture thenAcceptAsync(Consumer action) //action用指定线程池执行 public CompletableFuture thenAcceptAsync(Consumer action, Executor executor)

  • 使用示例
    ExecutorService executor = Executors.newSingleThreadExecutor(); CompletableFuture future = CompletableFuture .supplyAsync(() -> "hello siting", executor) .thenAcceptAsync(System.out::println, executor); executor.shutdown(); --------输出结果-------- hello siting

任务完成则运行fn,依赖上一个任务的结果,有返回值
public CompletableFuture thenApply(Function fn) public CompletableFuture thenApplyAsync(Function fn) //fn用指定线程池执行 public CompletableFuture thenApplyAsync(Function fn, Executor executor)

  • 【基础篇(异步编程不会(我教你啊!CompletableFuture(JDK1.8)))】使用示例
    ExecutorService executor = Executors.newSingleThreadExecutor(); CompletableFuture future = CompletableFuture .supplyAsync(() -> "hello world", executor) .thenApplyAsync(data -> { System.out.println(data); return "OK"; }, executor); System.out.println(future.join()); executor.shutdown(); --------输出结果-------- hello world OK

    thenCompose - 任务完成则运行fn,依赖上一个任务的结果,有返回值
  • 类似thenApply(区别是thenCompose的返回值是CompletionStage,thenApply则是返回 U),提供该方法为了和其他CompletableFuture任务更好地配套组合使用
    public CompletableFuture thenCompose(Function> fn) public CompletableFuture thenComposeAsync(Function> fn) public CompletableFuture thenComposeAsync(Function> fn, Executor executor)

  • 使用示例
    //第一个异步任务,常量任务 CompletableFuture f = CompletableFuture.completedFuture("OK"); //第二个异步任务 ExecutorService executor = Executors.newSingleThreadExecutor(); CompletableFuture future = CompletableFuture .supplyAsync(() -> "hello world", executor) .thenComposeAsync(data -> { System.out.println(data); return f; //使用第一个任务作为返回 }, executor); System.out.println(future.join()); executor.shutdown(); --------输出结果-------- hello world OK

3 线程并行执行 基础篇(异步编程不会(我教你啊!CompletableFuture(JDK1.8)))
文章图片

两个CompletableFuture并行执行完,然后执行action,不依赖上两个任务的结果,无返回值
public CompletableFuture runAfterBoth(CompletionStage other, Runnable action) public CompletableFuture runAfterBothAsync(CompletionStage other, Runnable action) public CompletableFuture runAfterBothAsync(CompletionStage other, Runnable action, Executor executor)

  • 使用示例
    //第一个异步任务,常量任务 CompletableFuture first = CompletableFuture.completedFuture("hello world"); ExecutorService executor = Executors.newSingleThreadExecutor(); CompletableFuture future = CompletableFuture //第二个异步任务 .supplyAsync(() -> "hello siting", executor) // () -> System.out.println("OK") 是第三个任务 .runAfterBothAsync(first, () -> System.out.println("OK"), executor); executor.shutdown(); --------输出结果-------- OK

两个CompletableFuture并行执行完,然后执行action,依赖上两个任务的结果,无返回值
//调用方任务和other并行完成后执行action,action再依赖消费两个任务的结果,无返回值 public CompletableFuture thenAcceptBoth(CompletionStage other, BiConsumer action) //两个任务异步完成,fn再依赖消费两个任务的结果,无返回值,使用默认线程池 public CompletableFuture thenAcceptBothAsync(CompletionStage other, BiConsumer action) //两个任务异步完成,fn(用指定线程池执行)再依赖消费两个任务的结果,无返回值 public CompletableFuture thenAcceptBothAsync(CompletionStage other, BiConsumer action, Executor executor)

  • 使用示例
    //第一个异步任务,常量任务 CompletableFuture first = CompletableFuture.completedFuture("hello world"); ExecutorService executor = Executors.newSingleThreadExecutor(); CompletableFuture future = CompletableFuture //第二个异步任务 .supplyAsync(() -> "hello siting", executor) // (w, s) -> System.out.println(s) 是第三个任务 .thenAcceptBothAsync(first, (s, w) -> System.out.println(s), executor); executor.shutdown(); --------输出结果-------- hello siting

两个CompletableFuture并行执行完,然后执行fn,依赖上两个任务的结果,有返回值
//调用方任务和other并行完成后,执行fn,fn再依赖消费两个任务的结果,有返回值 public CompletableFuture thenCombine(CompletionStage other, BiFunction fn) //两个任务异步完成,fn再依赖消费两个任务的结果,有返回值,使用默认线程池 public CompletableFuture thenCombineAsync(CompletionStage other, BiFunction fn) //两个任务异步完成,fn(用指定线程池执行)再依赖消费两个任务的结果,有返回值 public CompletableFuture thenCombineAsync(CompletionStage other, BiFunction fn, Executor executor)

  • 使用示例
    //第一个异步任务,常量任务 CompletableFuture first = CompletableFuture.completedFuture("hello world"); ExecutorService executor = Executors.newSingleThreadExecutor(); CompletableFuture future = CompletableFuture //第二个异步任务 .supplyAsync(() -> "hello siting", executor) // (w, s) -> System.out.println(s) 是第三个任务 .thenCombineAsync(first, (s, w) -> { System.out.println(s); return "OK"; }, executor); System.out.println(future.join()); executor.shutdown(); --------输出结果-------- hello siting OK

4 线程并行执行,谁先执行完则谁触发下一任务(二者选其最快) 基础篇(异步编程不会(我教你啊!CompletableFuture(JDK1.8)))
文章图片

上一个任务或者other任务完成, 运行action,不依赖前一任务的结果,无返回值
public CompletableFuture runAfterEither(CompletionStage other, Runnable action) public CompletableFuture runAfterEitherAsync(CompletionStage other, Runnable action) //action用指定线程池执行 public CompletableFuture runAfterEitherAsync(CompletionStage other, Runnable action, Executor executor)

  • 使用示例
    //第一个异步任务,休眠1秒,保证最晚执行晚 CompletableFuture first = CompletableFuture.supplyAsync(()->{ try{ Thread.sleep(1000); }catch (Exception e){} System.out.println("hello world"); return "hello world"; }); ExecutorService executor = Executors.newSingleThreadExecutor(); CompletableFuture future = CompletableFuture //第二个异步任务 .supplyAsync(() ->{ System.out.println("hello siting"); return "hello siting"; } , executor) //() ->System.out.println("OK") 是第三个任务 .runAfterEitherAsync(first, () ->System.out.println("OK") , executor); executor.shutdown(); --------输出结果-------- hello siting OK

上一个任务或者other任务完成, 运行action,依赖最先完成任务的结果,无返回值
public CompletableFuture acceptEither(CompletionStage other, Consumer action) public CompletableFuture acceptEitherAsync(CompletionStage other, Consumer action, Executor executor) //action用指定线程池执行 public CompletableFuture acceptEitherAsync(CompletionStage other, Consumer action, Executor executor)

  • 使用示例
    //第一个异步任务,休眠1秒,保证最晚执行晚 CompletableFuture first = CompletableFuture.supplyAsync(()->{ try{ Thread.sleep(1000); }catch (Exception e){} return "hello world"; }); ExecutorService executor = Executors.newSingleThreadExecutor(); CompletableFuture future = CompletableFuture //第二个异步任务 .supplyAsync(() -> "hello siting", executor) // data ->System.out.println(data) 是第三个任务 .acceptEitherAsync(first, data ->System.out.println(data) , executor); executor.shutdown(); --------输出结果-------- hello siting

上一个任务或者other任务完成, 运行fn,依赖最先完成任务的结果,有返回值
public CompletableFuture applyToEither(CompletionStage other, Function fn) public CompletableFuture applyToEitherAsync(CompletionStage other, Function fn) //fn用指定线程池执行 public CompletableFuture applyToEitherAsync(CompletionStage other, Function fn, Executor executor)

  • 使用示例
    //第一个异步任务,休眠1秒,保证最晚执行晚 CompletableFuture first = CompletableFuture.supplyAsync(()->{ try{ Thread.sleep(1000); }catch (Exception e){} return "hello world"; }); ExecutorService executor = Executors.newSingleThreadExecutor(); CompletableFuture future = CompletableFuture //第二个异步任务 .supplyAsync(() -> "hello siting", executor) // data ->System.out.println(data) 是第三个任务 .applyToEitherAsync(first, data ->{ System.out.println(data); return "OK"; } , executor); System.out.println(future); executor.shutdown(); --------输出结果-------- hello siting OK

5 处理任务结果或者异常 exceptionally-处理异常 基础篇(异步编程不会(我教你啊!CompletableFuture(JDK1.8)))
文章图片

public CompletableFuture exceptionally(Function fn)

  • 如果之前的处理环节有异常问题,则会触发exceptionally的调用相当于 try...catch
  • 使用示例
    CompletableFuture first = CompletableFuture .supplyAsync(() -> { if (true) { throw new RuntimeException("main error!"); } return "hello world"; }) .thenApply(data -> 1) .exceptionally(e -> { e.printStackTrace(); // 异常捕捉处理,前面两个处理环节的日常都能捕获 return 0; });

handle-任务完成或者异常时运行fn,返回值为fn的返回
  • 相比exceptionally而言,即可处理上一环节的异常也可以处理其正常返回值
    public CompletableFuture handle(BiFunction fn) public CompletableFuture handleAsync(BiFunction fn) public CompletableFuture handleAsync(BiFunction fn, Executor executor)

  • 使用示例
    CompletableFuture first = CompletableFuture .supplyAsync(() -> { if (true) { throw new RuntimeException("main error!"); } return "hello world"; }) .thenApply(data -> 1) .handleAsync((data,e) -> { e.printStackTrace(); // 异常捕捉处理 return data; }); System.out.println(first.join()); --------输出结果-------- java.util.concurrent.CompletionException: java.lang.RuntimeException: main error! ... 5 more null

whenComplete-任务完成或者异常时运行action,有返回值
  • whenComplete与handle的区别在于,它不参与返回结果的处理,把它当成监听器即可
  • 即使异常被处理,在CompletableFuture外层,异常也会再次复现
  • 使用whenCompleteAsync时,返回结果则需要考虑多线程操作问题,毕竟会出现两个线程同时操作一个结果
    public CompletableFuture whenComplete(BiConsumer action) public CompletableFuture whenCompleteAsync(BiConsumer action) public CompletableFuture whenCompleteAsync(BiConsumer action, Executor executor)

  • 使用示例
    CompletableFuture first = CompletableFuture .supplyAsync(() -> { if (true) {throw new RuntimeException("main error!"); } return "hello world"; }) .thenApply(data -> new AtomicBoolean(false)) .whenCompleteAsync((data,e) -> { //异常捕捉处理, 但是异常还是会在外层复现 System.out.println(e.getMessage()); }); first.join(); --------输出结果-------- java.lang.RuntimeException: main error! Exception in thread "main" java.util.concurrent.CompletionException: java.lang.RuntimeException: main error! ... 5 more

6 多个任务的简单组合
public static CompletableFuture allOf(CompletableFuture... cfs) public static CompletableFuture anyOf(CompletableFuture... cfs)
基础篇(异步编程不会(我教你啊!CompletableFuture(JDK1.8)))
文章图片

基础篇(异步编程不会(我教你啊!CompletableFuture(JDK1.8)))
文章图片