CompletableFuture异步回调

CompletableFuture异步回调

    • CompletableFuture
      • Future模式
      • CompletableFuture详解
        • 1.CompletableFuture的UML类关系
        • 2.CompletionStage接口
        • 3.使用runAsync和supplyAcync创建子任务
        • 4.设置子任务回调钩子
        • 5.调用handle()方法统一处理异常和结果
        • 6.线程池的使用
      • 异步任务的串行执行
        • thenApply()方法
        • thenRun()方法
        • thenAccept()方法
        • thenCompose()方法
        • 四个任务串行方法的区别
      • 异步任务的合并执行
        • thenCombine()方法
        • runAfterBoth()方法
        • thenAcceptBoth()方法
        • allOf()等待所有的任务结束
      • 泡茶喝实例

CompletableFuture

Future模式

  Future模式是高并发设计与开发过程中常见的设计模式,它的核心思想是异步调用。对于Future模式来说,它不是立即返回我们所需要的数据,但是它会返回一个契约(或异步任务),将来我们可以凭借这个契约(或异步任务)获取需要的结果。

  在进行传统的RPC(远程调用)时,同步调用RPC是一段耗时的过程。当客户端发出RPC请求后,服务端完成请求处理需要很长的一段时间才会返回,这个过程中客户端一直在等待,直到数据返回后,再进行其他任务的处理。现有一个Client同步对三个Server分别进行一次RPC调用,具体如下图:

image-20231004124526298

  假设一次远程调用的时间为500毫秒,则一个Client同步对三个Server分别进行一次RPC调用的总时间需要耗费1500毫秒。可以使用Future模式对其进行改造,将同步的RPC调用改为异步并发的RPC调用,一个Client异步并发对三个Server分别进行一次RPC调用,如下图:

image-20231004124912843

  假设一次远程调用的时间为500毫秒,则一个Client异步并发对三个Server分别进行一次RPC调用的总时间只要耗费500毫秒。使用Future模式异步并发地进行RPC调用,客户端在得到一个RPC地返回结果前并不急于获取该结果,而是充分利用等待时间去执行其他地耗时操作(如其他RPC调用),这就是Future模式地核心所在。

  Java的Future实现类并没有支持异步回调,仍然需要主动获取耗时任务的结果,而Java8的CompletableFuture组件实现了异步回调模式。

  在Java中,Future只是一个泛型接口,位于java.util.concurrent包下,其中定义了5个方法,主要包括如下几个功能:

  • 取消异步执行中的任务
  • 判断任务是否被取消
  • 判断异步任务是否执行完成
  • 获取异步任务完成后的执行结果
public interface Future<V> {//取消异步执行boolean cancel(boolean mayInterruptIfRunning);//获取异步任务的取消状态boolean isCancelled();//判断异步任务是否执行完成boolean isDone();//获取异步任务完成后的执行结果V get() throws InterruptedException, ExecutionException;//设置时限,获取异步任务完成后的执行结果V get(long timeout, TimeUnit unit)throws InterruptedException, ExecutionException, TimeoutException;
}

  总体来说,Future是一个对异步任务进行交互、操作的接口。但是Future仅仅是一个接口,通过它没有办法直接对异步任务操作,JDK提供了一个默认的实现类–FutureTask。

CompletableFuture详解

  CompletableFuture是JDK1.8引入的实现类,该类实现了FutureCompletionStage两个接口。该类的实例作为一个异步任务,可以在自己异步执行完成之后触发一些其他的异步任务,从而达到异步回调的效果。

1.CompletableFuture的UML类关系

image-20231004180648069

  这里的CompletionStage代表异步计算过程中的某一个阶段,一个阶段完成以后可能会进入另一个阶段。一个阶段可以理解为一个子任务,每一个子任务会包装一个Java函数式接口实例,表示该子任务所要执行的操作。

2.CompletionStage接口

  CompletionStage代表某个同步或异步计算的一个阶段,或者一系列异步任务中的一个子任务(或者阶段性任务)。

  每个CompletionStage子任务所包装的可以是一个FunctionConsumer或者Runnable函数式接口实例。这三个常用的函数式接口的特点如下:

  (1)Function

  Function接口的特点是:有输入、有输出。包装了Function实例的CompletionStage子任务需要一个输入参数,并会产生一个输出结果到下一步。

  (2)Runnable

  Runnable接口的特点是:无输入、无输出。包装了Runnable实例的CompletionStage子任务既不需要任何输入参数,又不会产生任何输出。

  (3)Consumer

  Consumer接口的特点是:有输入、无输出。包装了Consumer实例的CompletinStage子任务需要一个输入参数,但不会产生任何输出。

  多个CompletionStage构成了一条任务流水线,一个环节执行完成了可以将结果移交给下一个环节(子任务)。多个CompletionStage子任务之间可以使用链式调用,下面是一个例子:

oneStage.thenApply(x->square(x)).thenAccept(y->System.out.println(y)).thenRun(()->System.out.println())

对上例的解释如下:

oneStage是一个CompletionStage子任务,这是一个前提。

x->square(x)是一个Function类型的Lambda表达式,被thenApply()方法包装成了一个CompletionStage子任务,该子任务需要接收一个参数x,然后输出一个结果----x的平方值。

y->System.out.println(y)是一个Consumer类型的Lambda表达式,被thenAccept()方法包装成了一个CompletionStage子任务,该子任务需要消耗上一个子任务的输出值,但是此子任务并没有输出。

()->System.out.println()是一个Runnable类型的Lambda表达式,被thenRun()方法包装成了一个CompletionStage子任务,既不消耗上一个子任务的输出,又不产生结果。

  CompletionStage代表异步计算过程中的某一个阶段,一个阶段完成以后可能会触发另一个阶段。虽然一个子任务可以触发其他子任务,但是并不能保证后续子任务的执行顺序。

3.使用runAsync和supplyAcync创建子任务

  CompletionStage子任务的创建是通过CompletebleFuture完成的。CompletableFuture类提供了非常强大的Future的扩展功能来帮助我们简化异步编程的复杂性,提供了函数式编程的能力来帮我们通过回调的方式处理计算结果,也提供了转换和组合CompletionStage()的方法。

  CompletableFuture定义了一组方法用于创建CompletionStage子任务(或者阶段性任务),基础的方法如下:

  	//子任务包装一个Supplier实例,并调用ForkJoinPool.commonPool()线程来执行public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) {return asyncSupplyStage(asyncPool, supplier);}//子任务包装一个Supplier实例,并使用指定的executor线程池来执行public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier,Executor executor) {return asyncSupplyStage(screenExecutor(executor), supplier);}//子任务包装一个Runnable实例,并调用ForkJoinPool.commonPool()线程来执行public static CompletableFuture<Void> runAsync(Runnable runnable) {return asyncRunStage(asyncPool, runnable);}//子任务包装一个Runnable实例,并使用指定的executor线程池来执行public static CompletableFuture<Void> runAsync(Runnable runnable,Executor executor) {return asyncRunStage(screenExecutor(executor), runnable);}

  在使用CompletableFuture创建CompletionStage子任务时,如果没有指定Executor线程池,在默认情况下CompletionStage会使用公共的ForkJoinPool线程池。

  两个创建CompletionStage子任务的示例如下:

public class CompletableFutureDemo {//创建一个无消耗值(无输入值)、无返回值的异步子任务public static void runAsyncDemo() throws Exception {CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {try {Thread.sleep(1000);System.out.println("run end...");} catch (InterruptedException e) {throw new RuntimeException(e);}});//等待异步任务执行完成,限时等待2秒future.get(2, TimeUnit.SECONDS);}//创建一个无消耗值(无输入值)、有返回值的异步子任务public static void supplyAsyncDemo() throws Exception {CompletableFuture<Long> future = CompletableFuture.supplyAsync(() -> {long start = System.currentTimeMillis();try {Thread.sleep(1000);} catch (InterruptedException e) {throw new RuntimeException(e);}System.out.println("run end...");return System.currentTimeMillis() - start;});//等待异步任务执行完成,限时等待2秒long time = future.get(2, TimeUnit.SECONDS);System.out.println("异步执行耗时(秒):" + time / 1000);}public static void main(String[] args) throws Exception {runAsyncDemo();supplyAsyncDemo();}
}
4.设置子任务回调钩子

  可以为CompletionStage子任务设置特定的回调钩子,当计算结果完成或者抛出异常的时候,执行这些特定的回调钩子。

  设置子任务回调钩子的主要函数如下:

//设置子任务完成时的回调钩子
public CompletableFuture<T> whenComplete(BiConsumer<? super T, ? super Throwable> action) {return uniWhenCompleteStage(null, action);
}
//设置子任务完成时的回调钩子,可能不在同一线程执行
public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action) {return uniWhenCompleteStage(asyncPool, action);
}
//设置子任务完成时的回调钩子,提交给线程池executor执行
public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action, Executor executor) {return uniWhenCompleteStage(screenExecutor(executor), action);
}
//设置异常处理的回调钩子
public CompletableFuture<T> exceptionally(Function<Throwable, ? extends T> fn) {return uniExceptionallyStage(fn);
}

下面是一个CompletionStage子任务设置完成钩子和异常钩子的简单示例:

public class CompletableFutureDemo1 {public static void whenCompleteDemo() throws Exception {//创建异步任务CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {try {//模拟执行一秒Thread.sleep(1000);} catch (InterruptedException e) {throw new RuntimeException(e);}System.out.println(Thread.currentThread().getName()+":抛出异常");throw new RuntimeException(Thread.currentThread().getName()+":发生异常");});//设置异步任务执行完成后的回调钩子future.whenComplete(new BiConsumer<Void, Throwable>() {@Overridepublic void accept(Void unused, Throwable throwable) {System.out.println(Thread.currentThread().getName()+":执行完成!");}});//设置异步任务发生异常后的回调钩子future.exceptionally(new Function<Throwable, Void>() {@Overridepublic Void apply(Throwable throwable) {System.out.println(Thread.currentThread().getName()+":执行失败!" + throwable.getMessage());return null;}});//获取异步任务的结果future.get();}public static void main(String[] args) throws Exception {whenCompleteDemo();}
}

image-20231004193257898

  调用cancel()方法取消CompletableFuture时,任务被视为异常完成,completeExceptionally()方法所设置的异常回调钩子也会被执行。

  如果没有设置异常回调钩子,发生内部异常时会有两种情况发生:

  (1)在调用get()和get(long,TimeUnit)方法启动任务时,如果遇到内部异常,get()方法就会抛出ExecutionException(执行异常)。

  (2)在调用join()和getNow(T)启动任务时(大多数情况下都是如此),如果遇到内部异常,join()和getNow(T)方法就会抛出CompletionException。

5.调用handle()方法统一处理异常和结果

  除了分别通过whenComplete、exceptionally设置完成钩子、异常钩子之外,还可以调用handle()方法统一处理结果和异常。

  handle方法有三个重载版本如下:

//在执行任务的同一个线程中处理异常和结果
public <U> CompletableFuture<U> handle(BiFunction<? super T, Throwable, ? extends U> fn) {return uniHandleStage(null, fn);
}
//可能不在执行任务的同一个线程中处理异常和结果
public <U> CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn) {return uniHandleStage(asyncPool, fn);
}
//在指定线程池executor中处理异常和结果
public <U> CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn, Executor executor) {return uniHandleStage(screenExecutor(executor), fn);
}

  handle()方法的示例代码如下:

public class CompletableDemo2 {public static void handleDemo() throws Exception {CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {try {//模拟执行1秒Thread.sleep(1000);} catch (InterruptedException e) {throw new RuntimeException(e);}System.out.println(Thread.currentThread().getName()+":抛出异常");throw new RuntimeException(Thread.currentThread().getName()+":发生异常");});//统一处理异常和结果future.handle(new BiFunction<Void, Throwable, Void>() {@Overridepublic Void apply(Void unused, Throwable throwable) {if (throwable == null) {System.out.println(Thread.currentThread().getName()+":没有发生异常!");} else {System.out.println(Thread.currentThread().getName()+":sorry,发生了异常!");}return null;}});future.get();}public static void main(String[] args) throws Exception {handleDemo();}
}

image-20231004193956636

6.线程池的使用

  默认情况下,通过静态方法runAsync()、supplyAsync()创建的CompletableFuture任务会使用公共的ForkJoinPool线程池,默认的线程数是CPU的核数。

  问题是,如果所有CompletableFuture共享一个线程池,那么一旦有任务执行一些很慢的IO操作,就会导致线程池中的所有线程都阻塞在IO操作上,造成线程饥饿,进而影响整个系统的性能。所以,强烈建议根据不同的业务类型创建不同的线程池,以避免相互干扰

public class CompletableDemo3 {public static void threadPoolDemo() throws Exception {//混合线程池ExecutorService pool = Executors.newFixedThreadPool(10);CompletableFuture<Long> future = CompletableFuture.supplyAsync(() -> {System.out.println(Thread.currentThread().getName()+":run begin...");long start = System.currentTimeMillis();try {Thread.sleep(1000);//模拟执行1秒} catch (InterruptedException e) {throw new RuntimeException(e);}System.out.println(Thread.currentThread().getName()+":run end...");return System.currentTimeMillis() - start;}, pool);//等待异步任务执行完成,限时等待2秒Long time = future.get(2, TimeUnit.SECONDS);System.out.println(Thread.currentThread().getName()+":异步执行耗时(秒):" + time / 1000);}public static void main(String[] args) throws Exception {threadPoolDemo();}
}

image-20231004194419567

异步任务的串行执行

  如果两个异步任务需要串行(一个任务依赖另一个任务)执行,可以通过CompletionStage接口的thenApply()thenAccept()thenRun()thenCompose()四个方法来实现。

thenApply()方法
//后一个任务与前一个任务在同一个线程中执行
public <U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn) 
//后一个任务与前一个任务不在同一个线程中执行
public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn) 
//后一个任务在指定的executor线程池中执行
public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn, Executor executor)

   thenApply的三个重载版本有一个共同的参数fn,该参数表示要串行执行的第二个异步任务,它的类型为Function。fn的类型声明涉及两个泛型参数,具体如下:

  • 泛型参数T:上一个任务所返回结果的类型。
  • 泛型参数U:当前任务的返回值类型。

调用thenApply分两步计算(10+10)*2:

//调用thenApply分两步计算(10+10)*2
public class ThenApplyDemo {public static void main(String[] args) throws ExecutionException, InterruptedException {CompletableFuture<Long> future = CompletableFuture.supplyAsync(new Supplier<Long>() {@Overridepublic Long get() {long firstStep = 10L + 10L;System.out.println(Thread.currentThread().getName()+":firstStep out is " + firstStep);return firstStep;}}).thenApplyAsync(new Function<Long, Long>() {@Overridepublic Long apply(Long firstStepOut) {long secondStep = firstStepOut * 2;System.out.println(Thread.currentThread().getName()+":secondStep out is " + secondStep);return secondStep;}});Long result = future.get();System.out.println(Thread.currentThread().getName()+":out is " + result);}
}

image-20231004195110503

thenRun()方法

  thenRun()方法与thenApply()方法不同的是,不关心任务的处理结果。只要前一个任务执行完成,就开始执行后一个串行任务。

//后一个任务与前一个任务在同一个线程中执行
public CompletableFuture<Void> thenRun(Runnable action)
//后一个任务与前一个任务不在同一个线程中执行
public CompletableFuture<Void> thenRunAsync(Runnable action)
//后一个任务在指定的executor线程池中执行
public CompletableFuture<Void> thenRunAsync(Runnable action,Executor executor) 

  从方法的声明可以看出,thenRun()方法同thenApply()方法类似,不同的是前一个任务处理完成后,thenRun()并不会把计算的结果传给后一个任务,而且后一个任务也没有结果输出。

thenRun系列方法中的action参数是Runnable类型的,所以thenRun()既不能接收参数也不支持返回值。

thenAccept()方法

  调用thenAccept()方法时后一个任务可以接收(或消费)前一个任务的处理结果,但是后一个任务没有结果输出。

//后一个任务与前一个任务在同一个线程中执行
public CompletableFuture<Void> thenAccept(Consumer<? super T> action)
//后一个任务与前一个任务不在同一个线程中执行
public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action) 
//后一个任务在指定的executor线程池中执行
public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action,Executor executor)
thenCompose()方法

  thenCompose方法在功能上与thenApply()、thenAccept()、thenRun()一样,可以对两个任务进行串行的调度操作,第一个任务操作完成时,将它的结果作为参数传递给第二个任务。

public <U> CompletableFuture<U> thenCompose(Function<? super T, ? extends CompletionStage<U>> fn) 
public <U> CompletableFuture<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn) public <U> CompletableFuture<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn,Executor executor) 

  thenCompose()方法要求第二个任务的返回值是一个CompletionStage异步实例。因此,可以调用CompletableFuture.supplyAsync()方法将第二个任务所要调用的普通异步方法包装成一个CompletionStage实例。

  这里使用thenCompose()分两步计算(10+10)*2

public class ThenComposeDemo {public static void main(String[] args) throws ExecutionException, InterruptedException {CompletableFuture<Long> future = CompletableFuture.supplyAsync(new Supplier<Long>() {@Overridepublic Long get() {long firstStep = 10 + 10;System.out.println(Thread.currentThread().getName()+":first out is " + firstStep);return firstStep;}}).thenCompose(new Function<Long, CompletionStage<Long>>() {@Overridepublic CompletionStage<Long> apply(Long firstStepOut) {//将第二个任务所要调用的普通异步方法包装成一个CompletionState异步实例return CompletableFuture.supplyAsync(new Supplier<Long>() {//两个任务所要调用的普通异步方法@Overridepublic Long get() {long secondStep = firstStepOut * 2;System.out.println(Thread.currentThread().getName()+":second Step out is " + secondStep);return secondStep;}});}});Long result = future.get();System.out.println(Thread.currentThread().getName()+":out is " + result);}
}

image-20231004200557549

四个任务串行方法的区别

  thenApply()、thenRun()、thenAccept()这三个方法的不同之处主要在于其核心参数fn、action、consumer的类型不同,分别为Function<T,R>、Runnable、Consumer<? super T>类型。

  但是,thenCompose()方法与thenApply()方法有本质的不同:

  • thenCompose()的返回值是一个新的CompletionStage实例,可以持续用来进行下一轮CompletionStage任务的调度。
  • thenApply()的返回值就是第二个任务的普通异步方法的执行结果,它的返回类型与第二不执行的普通异步方法的返回类型相同,通过thenApply()所返回的值不能进行下一轮CompletionStage链式(或者流式)调用。

异步任务的合并执行

  如果某个任务同时依赖另外两个异步任务的执行结果,就需要对另外两个异步任务进行合并。以泡茶为例,“泡茶喝”任务需要对“烧水”任务与“清洗”任务进行合并。

thenCombine()方法

  thenCombine()会在两个CompletionStage任务都执行完成后,把两个任务的结果一起交给thenCombine()来处理。

//合并代表第二步任务(参数other)的CompletionStage实例,返回第三步任务的CompletionStage
public <U,V> CompletableFuture<V> thenCombine(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn)
//不一定在同一个线程中执行第三步任务的CompletionStage实例
public <U,V> CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn)
//第三步任务的CompletionStage实例在指定的executor线程池中执行
public <U,V> CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn, Executor executor)

  thenCombine()方法的调用者为第一步的CompletionStage实例,该方法的第一个参数为第二步的CompletionStage实例,该方法的返回值为第三步的CompletioStage实例。在逻辑上,thenCombine()方法的功能是将第一步、第二步的结果合并到第三步上。

  thenCombine()系列方法有两个核心参数:

  • other参数:表示待合并的第二部任务的CompletionStage实例。
  • fn参数:表示第一个任务和第二个任务执行完成后,第三步需要执行的逻辑。

  fn参数的类型为BiFunction<? super T,? super U,? extends V>,该类型的声明涉及三个泛型参数:

  • T:表示第一个任务所返回结果的类型
  • U:表示第二个任务所返回结果的类型
  • V:表示第三个任务所返回结果的类型

  调用thenCombine分三步计算(10+10)*(10+10):

//使用thenCombine()计算(10+10)*(10+10)
public class ThenCombineDemo {public static void main(String[] args) throws ExecutionException, InterruptedException {CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(new Supplier<Integer>() {@Overridepublic Integer get() {Integer firstStep = 10 + 10;System.out.println(Thread.currentThread().getName()+":firstStep out is " + firstStep);return firstStep;}});CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(new Supplier<Integer>() {@Overridepublic Integer get() {Integer secondStep = 10 + 10;System.out.println(Thread.currentThread().getName()+":secondStep out is " + secondStep);return secondStep;}});CompletableFuture<Integer> future3 = future1.thenCombine(future2, new BiFunction<Integer, Integer, Integer>() {@Overridepublic Integer apply(Integer step1Out, Integer step2Out) {return step1Out * step2Out;}});Integer result = future3.get();System.out.println(Thread.currentThread().getName()+":out is " + result);}
}

image-20231004201828595

runAfterBoth()方法

  funAfterBoth()方法不关心每一步任务的输入参数和处理结果。

//合并第二步任务的CompletionStage实例,返回第三步任务的CompletionStage
public CompletableFuture<Void> runAfterBoth(CompletionStage<?> other,Runnable action)
//不一定在同一个线程中执行第三步任务的CompletionStage实例
public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other,Runnable action)
//第三步任务的CompletionStage实例在指定的executor线程池中执行
public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other,Runnable action,Executor executor) 
thenAcceptBoth()方法

  该方法是对runAfterBoth()方法和thenCombine()方法的特点进行了折中,调用thenAcceptBoth()方法,第三个任务可以接收其合并过来的第一个任务、第二个任务的处理结果,但是第三个任务(合并任务)却不能返回结果。

public <U> CompletableFuture<Void> thenAcceptBoth(CompletionStage<? extends U> other,BiConsumer<? super T, ? super U> action)
public <U> CompletableFuture<Void> thenAcceptBothAsync(CompletionStage<? extends U> other,BiConsumer<? super T, ? super U> action)
public <U> CompletableFuture<Void> thenAcceptBothAsync(CompletionStage<? extends U> other,BiConsumer<? super T, ? super U> action, Executor executor)
allOf()等待所有的任务结束

  CompletionStage接口的allOf()会等待所有的任务结束,以合并所有的任务。thenCombine()只能合并两个任务,如果需要合并多个异步任务,那么可以调用allOf()

//allOF()会等待所有的任务结束,以合并所有的任务
public class AllOfDemo {public static void main(String[] args) {CompletableFuture<Void> future1 = CompletableFuture.runAsync(() -> System.out.println(Thread.currentThread().getName()+":模拟异步任务1"));CompletableFuture<Void> future2 = CompletableFuture.runAsync(() -> System.out.println(Thread.currentThread().getName()+":模拟异步任务2"));CompletableFuture<Void> future3 = CompletableFuture.runAsync(() -> System.out.println(Thread.currentThread().getName()+":模拟异步任务3"));CompletableFuture<Void> future4 = CompletableFuture.runAsync(() -> System.out.println(Thread.currentThread().getName()+":模拟异步任务4"));CompletableFuture<Void> all = CompletableFuture.allOf(future1, future2, future3, future4);all.join();System.out.println(Thread.currentThread().getName()+":所有异步任务都已经执行完毕");}
}

image-20231004202627949

CompletableFuture中的方法太多了,这里只介绍常用的

泡茶喝实例

  为了异步执行整个排查流程,分别设计三个线程:泡茶线程(MainThread,主线程)、烧水线程(HotWaterThread)和清洗线程(WashThread)。

  • 泡茶线程的工作是:启动清洗线程、启动烧水线程,等清洗、烧水的工作完成后,泡茶喝;

  • 清洗线程的工作是:洗茶壶、洗茶杯;

  • 烧水线程的工作是:洗好水壶,灌上凉水,放在火上,一直等水烧开。

  我们使用CompletableFuture实现整个泡茶喝程序。我们分3个任务:

  任务1负责洗水壶、烧开水

  任务2负责洗茶壶、洗茶杯和拿茶叶

  任务3负责泡茶

public class DrinkTea {public static final int SLEEP_GAP = 3000;    //等待3秒public static void main(String[] args) throws ExecutionException, InterruptedException {//任务1:洗水壶->烧开水CompletableFuture<Boolean> hotJob = CompletableFuture.supplyAsync(() -> {System.out.println(Thread.currentThread().getName() + ":洗好水壶");System.out.println(Thread.currentThread().getName() + ":烧开水");//线程睡眠一段时间,代表烧水中try {Thread.sleep(SLEEP_GAP);} catch (InterruptedException e) {throw new RuntimeException(e);}System.out.println(Thread.currentThread().getName() + ":水开了");return true;});//任务2:洗茶壶->洗茶杯->拿茶叶CompletableFuture<Boolean> washJob = CompletableFuture.supplyAsync(() -> {System.out.println(Thread.currentThread().getName() + ":洗茶壶");System.out.println(Thread.currentThread().getName() + ":洗茶杯");System.out.println(Thread.currentThread().getName() + ":拿茶叶");//线程睡眠一段时间,代表清洗中try {Thread.sleep(SLEEP_GAP);} catch (InterruptedException e) {throw new RuntimeException(e);}System.out.println(Thread.currentThread().getName() + ":洗完了");return true;});//任务3:任务1和任务2完成后执行泡茶CompletableFuture<String> drinkJob = hotJob.thenCombine(washJob, new BiFunction<Boolean, Boolean, String>() {@Overridepublic String apply(Boolean hotOK, Boolean washOK) {if (hotOK && washOK) {System.out.println(Thread.currentThread().getName() + ":泡茶喝,茶喝完");return "茶喝完了";}return "没有喝到茶";}});//等待任务3执行结果System.out.println(Thread.currentThread().getName() + ":" + drinkJob.get());}
}

image-20231004203800659

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.hqwc.cn/news/124937.html

如若内容造成侵权/违法违规/事实不符,请联系编程知识网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

二叉树题目:路径总和 II

文章目录 题目标题和出处难度题目描述要求示例数据范围 前言解法一思路和算法代码复杂度分析 解法二思路和算法代码复杂度分析 题目 标题和出处 标题&#xff1a;路径总和 II 出处&#xff1a;113. 路径总和 II 难度 4 级 题目描述 要求 给你二叉树的根结点 root \tex…

星宿UI2.4资源付费变现小程序源码 支持流量主

第一个小程序为星宿小程序 目前是最新版2.0 搭建星宿需要&#xff1a;备用域名 服务器 微信小程序账号 功能&#xff1a;文章展示 文章分类 资源链接下载 轮播图 直接下载附件功能 很多 很适合做资源类分享 源码下载&#xff1a;https://download.csdn.net/download/m0_6604…

【微信小程序开发】一文学会使用CSS控制样式布局与美化

引言 在微信小程序开发中&#xff0c;CSS样式布局和美化是非常重要的一部分&#xff0c;它能够为小程序增添美感&#xff0c;提升用户体验。本文将介绍如何学习使用CSS进行样式布局和美化&#xff0c;同时给出代码示例&#xff0c;帮助开发者更好地掌握这一技巧。 一、CSS样式布…

怒刷LeetCode的第23天(Java版)

目录 第一题 题目来源 题目内容 解决方法 方法一&#xff1a;贪心算法 方法二&#xff1a;动态规划 方法三&#xff1a;回溯算法 方法四&#xff1a;并查集 第二题 题目来源 题目内容 解决方法 方法一&#xff1a;排序和遍历 方法二&#xff1a;扫描线算法 方法…

专业PDF编辑阅读工具PDF Expert mac中文特点介绍

PDF Expert mac是一款专业的PDF编辑和阅读工具。它可以帮助用户在Mac、iPad和iPhone等设备上查看、注释、编辑、填写和签署PDF文档。 PDF Expert mac软件特点 PDF编辑&#xff1a;PDF Expert提供了丰富的PDF编辑功能&#xff0c;包括添加、删除、移动、旋转、缩放、裁剪等操作…

Qt model/view 理解01

在 Qt 中对数据处理主要有两种方式&#xff1a;1&#xff09;直接对包含数据的的数据项 item 进行操作&#xff0c;这种方法简单、易操作&#xff0c;现实方式单一的缺点&#xff0c;特别是对于大数据或在不同位置重复出现的数据必须依次对其进行操作&#xff0c;如果现实方式改…

JMeter的详细使用及相关问题

一、中文乱码问题 如果出现乱码&#xff0c;需要修改编码集&#xff0c;&#xff08;版本问题有的不需要修改&#xff0c;就不用管&#xff09; 修改后保存重启就好了。 JMeter5.5版本的按照如下修改&#xff1a; 二、JMeter的启动 ①建议直接用ApacheJMeter.jar双击启动…

【2023年11月第四版教材】第18章《项目绩效域》(第一部分)

第18章《项目绩效域》&#xff08;第一部分&#xff09; 1 章节内容2 干系人绩效域2.1 绩效要点2.2 执行效果检查2.3 与其他绩效域的相互作用 3 团队绩效域3.1 绩效要点3.2 与其他绩效域的相互作用3.3 执行效果检查3.4 开发方法和生命周期绩效域 4 绩效要点4.1 与其他绩效域的相…

[React] react-redux基本使用

文章目录 1.redux2.安装redux3.操作redux3.1 创建最为核心的store3.2 创建为store工作的reducer3.3 redux的响应式处理 4.完整版redux4.1 完善actionCreators4.2 thunk中间件 5.react-redux5.1 Count容器组件5.2 connect函数5.3 Provider 1.redux redux原理图 actionCreators:…

Android LitePal byte[]类型字段不被创建

我创建了以下实体类&#xff0c;主要是用户分享的内容、分享的照片、分享的标题&#xff0c;然后百度了一下LitePal可以识别byte[]&#xff0c;因为需要文件的上传与读取&#xff1a; public class Context extends LitePalSupport {private Integer ContextId;private String…

经典网络解析(四) transformer | 自注意力、多头、发展

文章目录 1 背景1.1 困境1.2 基本架构 2 嵌入层3 编码器部分3.1 自注意力层3.2 多头注意力机制3.3 LayerNorm归一化层 4 解码器5 transformer的发展6 代码 1 背景 1.1 困境 transformer可以并行训练&#xff0c;也是用来实现attention注意力机制 之前RNN的困境 &#xff08…

【kubernetes】CRI OCI

1 OCI OCI(Open Container Initiative)&#xff1a;由Linux基金会主导&#xff0c;主要包含容器镜像规范和容器运行时规范&#xff1a; Image Specification(image-spec)Runtime Specification(runtime-spec)runC image-spec定义了镜像的格式&#xff0c;镜像的格式有以下几…