【问题标题】:CompletableFuture exception handling runAsync & thenRunCompletableFuture 异常处理 runAsync & thenRun
【发布时间】:2020-01-28 02:23:58
【问题描述】:

假设我有这个示例代码并且在runAsync 中遇到了异常。我的问题是这个异常是否会阻止thenRun 被执行,因为thenRun 与此代码的调用者方法在同一线程中运行。

private void caller() {
    CompletableFuture.runAsync(() -> {
          try {
              // some code
          } catch (Exception e) {
              throw new CustomException(errorMessage, e);
          }
         }, anInstanceOfTaskExecutor).thenRun(
         // thenRun code
     ));
}

我已经了解了this 线程,它解释了如何处理异步块抛出的异常(即通过阻塞和使用join)。我想知道如果CompletableFuture completesExceptionally 是否会执行thenRun 块内的代码。

更新

我运行了一些代码来测试这个:

CompletableFuture.runAsync(() -> {
      List<Integer> integerList = new ArrayList<>();
      integerList.get(1);    // throws exception
    }).thenRun(() -> {
      System.out.println("No exception occurred");
    });

它不打印任何东西,这意味着异常没有从异步块“传播到/到达”调用方方法的线程。我现在了解这里的预期行为,但我有以下问题:

  1. 为什么即使 CompletableFuture 异常完成,它也会静默失败?
  2. 它如何在后台工作?
  3. 是不是因为这两个线程(调用者线程和异步线程)都有自己的堆栈空间?

【问题讨论】:

    标签: java asynchronous exception java-8 completable-future


    【解决方案1】:

    一般信息

    CompletionStage的文档解释了接口的一般规则:

    可能是异步计算的一个阶段,当另一个CompletionStage 完成时执行一个操作或计算一个值。一个阶段在其计算终止时完成,但这可能反过来触发其他相关阶段。此接口中定义的功能仅采用几种基本形式,可扩展为更大的方法集以捕获一系列使用风格:

    • 阶段执行的计算可以表示为FunctionConsumerRunnable(使用名称包括applyaccept的方法,或 run)取决于它是否需要参数和/或产生结果。例如:

      stage.thenApply(x -> square(x))
           .thenAccept(x -> System.out.print(x))
           .thenRun(() -> System.out.println());
      

      另一种形式 (compose) 允许从返回完成阶段的函数构造计算管道。

      阶段计算的任何参数都是触发阶段计算的结果。

    • 一个阶段的执行可以由一个阶段的完成触发,也可以由两个阶段的完成触发,或者两个阶段的其中一个触发。使用前缀为 then 的方法排列单个阶段的依赖关系。由两个阶段完成触发的那些可以使用相应命名的方法组合它们的结果或效果。由两个阶段中的任一触发的不保证哪个结果或效果用于相关阶段的计算。

    • 阶段之间的依赖关系控制计算的触发,但不保证任何特定的顺序。此外,新阶段计算的执行可以通过以下三种方式中的任何一种进行安排:默认执行、默认异步执行(使用带有后缀的方法 async 采用阶段的默认异步执行工具)或自定义(通过提供的Executor)。默认和异步模式的执行属性由CompletionStage 实现指定,而不是这个接口。具有显式 Executor 参数的方法可能具有任意执行属性,甚至可能不支持并发执行,但以适应异步的方式安排处理。

    • 两种方法形式(handlewhenComplete)支持无条件计算触发阶段是否正常完成或异常完成。方法exceptionally仅在触发阶段异常完成时才支持计算,计算替换结果,类似于java [sic] catch关键字。在所有其他情况下,如果一个阶段的计算因(未经检查的)异常或错误而突然终止,则所有需要其完成的相关阶段也会异常完成,CompletionException 将异常作为其原因。如果一个阶段依赖于两个阶段中的both,并且都异常完成,则CompletionException 可能对应于这些异常中的任何一个。如果一个阶段依赖于其他两个阶段中的任一个,并且其中只有一个异常完成,则无法保证依赖阶段是正常完成还是异常完成。在方法whenComplete 的情况下,当提供的操作本身遇到异常时,该阶段将异常完成,除非源阶段也异常完成,在这种情况下,源阶段的异常完成优先并传播到依赖阶段。

    所有方法都遵循上述触发、执行和异常完成规范(在单个方法规范中不再重复)。 [...]

    [...]

    CompletableFuture 的文档解释了线程规则(和其他策略),如上所述,其中一些由CompletionStage 的实现决定:

    可以显式完成的Future(设置其值和状态),并且可以用作CompletionStage,支持在完成时触发的相关函数和操作。

    当两个或更多线程尝试completecompleteExceptionallycancelCompletableFuture 时,只有一个线程成功。

    除了直接操作状态和结果的这些和相关方法之外,CompletableFuture 还使用以下策略实现接口CompletionStage

    • 为非异步方法的依赖完成提供的操作可以由完成当前 CompletableFuture 的线程或完成方法的任何其他调用者执行。

    • 所有没有显式 Executor 参数的异步方法都使用 ForkJoinPool.commonPool() 执行(除非它不支持至少两个并行级别,在这种情况下,将创建一个新的 Thread 来运行每个任务)。这可以通过定义方法defaultExecutor() 来覆盖子类中的非静态方法。为了简化监控、调试和跟踪,所有生成的异步任务都是标记接口CompletableFuture.AsynchronousCompletionTask 的实例。有时间延迟的操作可以使用该类中定义的适配器方法,例如:supplyAsync(supplier, delayedExecutor(timeout, timeUnit))。为了支持具有延迟和超时的方法,该类最多维护一个用于触发和取消操作的守护线程,而不是用于运行它们。

    • 所有CompletionStage 方法都是独立于其他公共方法实现的,因此一个方法的行为不会受到子类中其他方法的覆盖的影响。

    • 所有CompletionStage 方法都返回CompletableFutures。要将用法限制为仅在接口CompletionStage 中定义的那些方法,请使用方法minimalCompletionStage()。或者为了确保客户端不会自己修改未来,请使用方法copy()

    CompletableFuture 还使用以下策略实现 Future

    • 由于(与FutureTask 不同)此类无法直接控制导致其完成的计算,因此取消被视为另一种形式的异常完成。方法cancelcompleteExceptionally(new CancellationException()) 的效果相同。方法isCompletedExceptionally() 可用于确定CompletableFuture 是否以任何异常方式完成。

    • 如果使用CompletionException 异常完成,方法get()get(long, TimeUnit) 会抛出一个ExecutionException,原因与对应的CompletionException 中的原因相同。为了在大多数情况下简化使用,这个类还定义了 join()getNow(T) 方法,在这些情况下直接抛出 CompletionException

    [...]


    您的问题

    这是您的示例代码:

    CompletableFuture.runAsync(() -> {
          List<Integer> integerList = new ArrayList<>();
          integerList.get(1);    // throws exception
        }).thenRun(() -> {
          System.out.println("No exception occurred");
        });
    

    如果您不知道,thenRun 等方法会返回 new CompletionStage。所以你的代码类似于以下:

    CompletableFuture<Void> runAsyncStage = CompletableFuture.runAsync(() -> List.of().get(0));
    CompletableFuture<Void> thenRunStage =
        runAsyncStage.thenRun(() -> System.out.println("thenRun executing!"));
    

    thenRunStagerunAsyncStage 的完成触发,在这种情况下,IndexOutOfBoundsException 保证异常完成。至于为什么Runnable没有被执行,那是因为CompletionStage#thenRun(Runnable)的合约:

    返回一个新的CompletionStage,当这个阶段正常完成时,它会执行给定的操作。有关异常完成的规则​​,请参阅 CompletionStage 文档。

    由于触发阶段异常完成,thenRunStage 阶段也异常完成,这意味着 Runnable 被跳过。

    1。 “为什么即使 CompletableFuture 异常完成,它也会默默地失败?”

    示例代码相当于使用 try-catch 块吞下异常。您没有看到异常,因为您还没有编写会报告异常的代码。 runAsyncStagethenRunStage两个阶段都异常完成,后者是因为前者异常完成。

    如果您想了解阶段“链内”的异常,则必须使用 exceptionally[Async]handle[Async]whenComplete[Async] 等阶段。这样做可以让您根据触发阶段的正常或异常完成来更改链的行为。

    如果您想了解阶段“链外”的异常,则必须使用join()get()get(long,TimeUnit) 等方法。如果阶段异常完成,则第一个阶段将抛出一个 CompletionException 包装失败原因,而后两个阶段将抛出一个 ExecutionException 包装失败原因。

    2。 “它在后台是如何工作的?”

    CompletableFuture 的实现过于复杂,无法在 Stack Overflow 答案中进行解释。如果您想研究实现,可以查看源代码。您的 JDK 应该带有一个包含 Java 源文件的 src.zip 文件。也可以在线查看源代码OpenJDK repositories。例如,这里是CompletableFuture的JDK 13源代码:

    https://hg.openjdk.java.net/jdk/jdk13/file/0368f3a073a9/src/java.base/share/classes/java/util/concurrent/CompletableFuture.java

    3。 “是不是因为这两个线程(调用者线程和异步线程)都有自己的堆栈空间?”

    除非两个线程之间存在某种通信,否则一个线程不会意识到另一个线程中的异常。诸如join() 之类的调用方法将在适当的时候将异常传达给将抛出所述异常的调用线程。但是,如您的第一个问题的答案所示,它比这稍微复杂一些。即使一个线程在单个阶段抛出异常,您也不会看到堆栈跟踪或类似的东西。这是因为异常被捕获,并且阶段被标记为失败以该异常作为原因。然后,其他代码必须根据需要显式检索和处理该异常。

    这与使用 ExecutorService 并返回 Future 对象没有什么不同。该任务可能在后台失败,但在查询 Future 之前,其他代码不会意识到这一点。

    来自赏金:“我希望了解线程之间如何交互的细节。”

    我不确定还要添加什么。 CompletionStage API 是“高于”线程的抽象。您只需告诉 API 您希望如何执行命令链,包括每个阶段使用哪些线程池,并且实现为您处理所有线程间通信。也就是说,每个线程都做自己的事情,只是 API 旨在提供一种更简单和反应性的方式在线程之间进行通信。如果您对如何实现感兴趣,那么我建议您研究源代码(上面链接)。

    【讨论】:

      【解决方案2】:

      这取决于您添加的步骤exceptionally

      以下情况,会跳过thenRun,直接执行异常阻塞。

      CompletableFuture.runAsync(() -> { 
           //process and throw exception
           }, anInstanceOfTaskExecutor )
          .thenRun(() -> {})
          .exceptionally(exception -> {
            // do something, handle exception
          })
       ));
      

      在这种情况下,它将执行thenRun

         CompletableFuture.runAsync(() -> { 
           //process and throw exception
           }, anInstanceOfTaskExecutor )
          .exceptionally(exception -> {
            // do something, handle exception
          })
          .thenRun(() -> {})
       ));
      

      希望对你有帮助

      【讨论】:

      • 谢谢@lucid。你会碰巧知道为什么这是预期的行为吗?为什么 thenRun 块会被跳过异步代码块完成异常?
      • 您可以查看官方文档中的第 4 点。这将有助于理解内部工作。 docs.oracle.com/javase/8/docs/api/java/util/concurrent/…
      • @Miguel 因为某些阶段仅适用于正常完成(例如thenRun),其他阶段仅适用于异常完成(例如exceptionally),而其他阶段则适用于任何一个(例如handle)。如何调用阶段取决于构建的阶段链。在您的问题示例中,thenRun 阶段异常完成,因为它的触发阶段异常竞争;在这种情况下Runnable 不会被执行,因为thenRun 的合约规定Runnable 只有在触发阶段正常完成时才会被执行。
      猜你喜欢
      • 1970-01-01
      • 1970-01-01
      • 2018-12-11
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      • 2018-06-17
      • 2021-07-03
      • 1970-01-01
      相关资源
      最近更新 更多