具有Runnable-delegation的CompletableFuture - 在委派类时忽略异常

Question

我在使用CompletableFuture将代码转换为非阻塞代码时遇到了问题.为了最小化问题的范围,我创建了一个示例代码,当我使用CompletableFuture时,该代码的行为有所不同.问题是CompletableFuture从Runnable-delegation吞下异常.

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

示例代码:

• 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 - 这是委托runnable,它委托并包装传递给它的Runnable的逻辑,以及用于异常处理的占位符.

  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 - 此委托执行方法.它只是使用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.Way2 - 使用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块可以在这里捕获异常,这是从MyRunnable引发的.但是如果我使用CompletableFuture使用way2(),则在DelegatingRunnable下MyRunnable的异常并不合适,尽管我看到它在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的类型的对象中.这将传递给执行者的方法.当内部/原始引发异常时,它会被代码捕获并且完成为失败但异常不会被重新抛出.AsyncRunRunnableAsyncRunexecuteRunnableAsyncRunCompletableFuture

这就是你的包装器(DelegatingRunnable)永远不会看到异常的原因.