java - CompletableFuture with Runnable-delegation - 在委托(delegate)类中忽略异常

Question

我在使用 CompletableFuture 将代码转换为非阻塞代码时遇到问题。为了尽量减少问题的范围，我创建了一个 sample code当我使用 CompletableFuture 时，它​​的行为有所不同。问题是 CompletableFuture 吞下了来自 Runnable-delegation 的异常。

我在 Runnable 和 ExecutorService 之上使用委托(delegate)来提供我的原始应用程序所需的一些包装器代码。

示例代码:

• MyRunnable:我的示例 runnable，它总是抛出异常。

  public class MyRunnable implements Runnable {
  
      @Override
      public void run() {
          System.out.println("This is My Thread throwing exception : " + Thread.currentThread().getName());
          throw new RuntimeException("Runtime exception from MyThread");
      }
  }
  

• DelegatingRunnable - 这是委托(delegate) runnable，它围绕传递给它的 Runnable 委托(delegate)和包装逻辑，以及用于异常处理的占位符。

  public class DelegatingRunnable implements Runnable {
  
      private Runnable delegate; 
  
      public DelegatingRunnable(Runnable delegate) {
          this.delegate = delegate;
      }
  
      @Override
      public void run() {
          System.out.println("Delegating Thread start : " + Thread.currentThread().getName());
          try {
              // Some code before thread execution
              delegate.run();
              // Some code after thread execution
          } catch (Exception e) {
              // While using CompletableFuture, could not catch exception here
              System.out.println("###### Delegating Thread Exception Caught : " + Thread.currentThread().getName());
              //throw new RuntimeException(e.getMessage());
          } catch (Throwable t) {
              System.out.println("!!!!!!! Delegating Thread Throwable Caught : " + Thread.currentThread().getName());
          }
          System.out.println("Delegating Thread ends : " + Thread.currentThread().getName());
      }
  
  }
  

• DelegatingExecutorService - 此委托(delegate)执行方法。它只是用 DelegatingRunnable 包装了 runnable。

  public class DelegatingExecutorService extends AbstractExecutorService {
  
      private ExecutorService executor;
  
      public DelegatingExecutorService(ExecutorService executor) {
          this.executor = executor;
      }
  
      @Override
      public void execute(Runnable command) {
          executor.execute(new DelegatingRunnable(command));
      }
  
      // Othere delegating methods
  
  }       
  

• MainClass - 我使用两种方法。 Way1 - 使用不带 CompletableFuture 的 ExecutorService。方式 2 - 使用 CompletableFuture

  public class MainClass {
  
      public static void main(String[] arg) {
          //way1();
          way2();
      }
  
      public static void way2() {
          System.out.println("Way:2 # This is main class : " + Thread.currentThread().getName());
  
          ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()+1);
          DelegatingExecutorService executorService = new DelegatingExecutorService(executor);
  
          CompletableFuture.runAsync(new MyRunnable(), executorService)
              .whenComplete((res, ex) -> {
                  if (ex != null) {
                      System.out.println("whenComplete - exception  : " + Thread.currentThread().getName());
                  } else {
                      System.out.println("whenComplete - success  : " + Thread.currentThread().getName());
                  }
              });
  
          executor.shutdown();
          System.out.println("main class completed : " + Thread.currentThread().getName());
      }
  
      public static void way1() {
          System.out.println("Way:1 # This is main class : " + Thread.currentThread().getName());
          ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()+1);
  
          DelegatingExecutorService executorService = new DelegatingExecutorService(executor);
  
          executorService.execute(new MyRunnable());
  
          executor.shutdown();
          System.out.println("main class completed : " + Thread.currentThread().getName());
      }
  }
  

问题: 当我运行 way1() 时，输出是

    Way:1 # This is main class : main
    Delegating Thread start : pool-1-thread-1
    This is My Thread throwing exception : pool-1-thread-1
    ###### Delegating Thread Exception Caught : pool-1-thread-1
    main class completed : main
    Delegating Thread ends : pool-1-thread-1

你可以注意到 'DelegatingRunnable' 的 catch block 可以捕获这里的异常，这是从 MyRunnable 引发的。但是，如果我使用 CompletableFuture 使用 way2()，MyRunnable 的异常不会在 DelegatingRunnable 下咳嗽，尽管我看到它在 CompletableFuture 的“whenComplete”回调下咳嗽。

way2的输出是

    Way:2 # This is main class : main
    Delegating Thread start : pool-1-thread-1
    This is My Thread throwing exception : pool-1-thread-1
    Delegating Thread ends : pool-1-thread-1
    whenComplete - exception  : main
    main class completed : main

您会注意到 CompletableFuture 在内部使用相同的 DelegatingExecutionService 和 DelegatingRunnable。我不明白为什么 DelegatingRunnable 在这种情况下无法捕获异常。

(为什么我要使用 CompletableFuture？-这只是一个示例代码来解释我所面临的确切问题。但总的来说，我需要使用 CompletableFuture 以非阻塞方式最终创建任务链)

Answer 1

在 CompletableFuture 的源代码中，您可以看到它将给定的 Runnable 包装在一个 AsyncRun 类型的对象中，该对象本身实现了 可运行。 此 AsyncRun 将传递给执行程序的 execute 方法。 当内部/原始 Runnable 抛出异常时，它会被 AsyncRun 的代码捕获并且 CompletableFuture 完成为失败但异常将不会被重新抛出。

这就是为什么您的包装器 (DelegatingRunnable) 永远不会看到异常的原因。