java - 如何中断 CompletableFuture 的底层执行

Question

我知道 CompletableFuture 设计不会通过中断来控制其执行，但我想你们中的一些人可能会遇到这个问题。 CompletableFutures 是组合异步执行的非常好的方法，但是如果您希望在取消 future 时中断或停止底层执行，我们该怎么做呢？或者我们必须接受任何已取消或手动完成的 CompletableFuture 不会影响在那里完成它的线程？

在我看来，这显然是一项无用的工作，需要 executor worker 的时间。我想知道在这种情况下哪种方法或设计可能会有所帮助？

更新

这是一个简单的测试

public class SimpleTest {

  @Test
  public void testCompletableFuture() throws Exception {
    CompletableFuture<Void> cf = CompletableFuture.runAsync(()->longOperation());

    bearSleep(1);

    //cf.cancel(true);
    cf.complete(null);

    System.out.println("it should die now already");
    bearSleep(7);
  }

  public static void longOperation(){
    System.out.println("started");
    bearSleep(5);
    System.out.println("completed");
  }

  private static void bearSleep(long seconds){
    try {
      TimeUnit.SECONDS.sleep(seconds);
    } catch (InterruptedException e) {
      System.out.println("OMG!!! Interrupt!!!");
    }
  }
}

Answer 1

一个 CompletableFuture与最终可能完成的异步操作无关。

Since (unlike FutureTask) this class has no direct control over the computation that causes it to be completed, cancellation is treated as just another form of exceptional completion. Method cancel has the same effect as completeExceptionally(new CancellationException()).

甚至可能没有个单独的线程来完成它(甚至可能有许多个线程在处理它)。即使有，也没有从 CompletableFuture 到任何引用它的线程的链接。

因此，您无法通过 CompletableFuture 执行任何操作来中断任何可能正在运行将完成它的任务的线程。您必须编写自己的逻辑来跟踪任何 Thread 实例，这些实例获取对 CompletableFuture 的引用以完成它。

---

这是我认为您可以逃脱的执行类型的示例。

public static void main(String[] args) throws Exception {
    ExecutorService service = Executors.newFixedThreadPool(1);
    CompletableFuture<String> completable = new CompletableFuture<>();
    Future<?> future = service.submit(new Runnable() {
        @Override
        public void run() {
            for (int i = 0; i < 10; i++) {
                if (Thread.interrupted()) {
                    return; // remains uncompleted
                }
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    return; // remains uncompleted
                }
            }
            completable.complete("done");
        }
    });

    Thread.sleep(2000);

    // not atomic across the two
    boolean cancelled = future.cancel(true);
    if (cancelled)
        completable.cancel(true); // may not have been cancelled if execution has already completed
    if (completable.isCancelled()) {
        System.out.println("cancelled");
    } else if (completable.isCompletedExceptionally()) {
        System.out.println("exception");
    } else {
        System.out.println("success");
    }
    service.shutdown();
}

这假定正在执行的任务已设置为正确处理中断。