从 CompletableFuture 调用 ExecutorService.shutdownNow

Question

当已经运行任务之一抛出异常时，我需要取消所有已安排但尚未运行 CompletableFuture 任务。

尝试了以下示例，但大多数时候 main 方法不会退出（可能是由于某种类型的死锁）。

public static void main(String[] args) {
    ExecutorService executionService = Executors.newFixedThreadPool(5);

    Set< CompletableFuture<?> > tasks = new HashSet<>();

    for (int i = 0; i < 1000; i++) {
        final int id = i;
        CompletableFuture<?> c = CompletableFuture

        .runAsync( () -> {
            System.out.println("Running: " + id); 
            if ( id == 400 ) throw new RuntimeException("Exception from: " + id);
        }, executionService )

        .whenComplete( (v, ex) -> { 
            if ( ex != null ) {
                System.out.println("Shutting down.");
                executionService.shutdownNow();
                System.out.println("shutdown.");
            }
        } );

        tasks.add(c);
    }

    try{ 
        CompletableFuture.allOf( tasks.stream().toArray(CompletableFuture[]::new) ).join(); 
    }catch(Exception e) { 
        System.out.println("Got async exception: " + e); 
    }finally { 
        System.out.println("DONE"); 
    }        
}

最后的打印输出是这样的：

Running: 402
Running: 400
Running: 408
Running: 407
Running: 406
Running: 405
Running: 411
Shutting down.
Running: 410
Running: 409
Running: 413
Running: 412
shutdown.

在单独的线程上尝试了运行 shutdownNow 方法，但在大多数情况下，它仍然会产生相同的死锁。

知道是什么导致了这个死锁吗？

当抛出异常时，您认为取消所有已安排但尚未运行 CompletableFuture 的最佳方法是什么？

正在考虑迭代 tasks 并在每个 CompletableFuture 上调用 cancel。但我不喜欢的是 CancellationException from join.

Answer 1

你应该记住

CompletableFuture<?> f = CompletableFuture.runAsync(runnable, executionService);

基本等同于

CompletableFuture<?> f = new CompletableFuture<>();
executionService.execute(() -> {
    if(!f.isDone()) {
        try {
            runnable.run();
            f.complete(null);
        }
        catch(Throwable t) {
            f.completeExceptionally(t);
        }
    }
});

因此ExecutorService对CompletableFuture一无所知，因此，一般无法取消。它所拥有的只是一些工作，表示为 Runnable.

的实现

换句话说，shutdownNow()会阻止pending jobs的执行，因此，剩余的futures不会正常完成，但不会取消它们。然后，您调用 join() 由 allOf 编辑的未来 return 由于从未完成的期货，它永远不会 return。

但请注意，计划作业会在执行任何昂贵的操作之前检查未来是否已经完成。

因此，如果您将代码更改为

ExecutorService executionService = Executors.newFixedThreadPool(5);
Set<CompletableFuture<?>> tasks = ConcurrentHashMap.newKeySet();
AtomicBoolean canceled = new AtomicBoolean();

for(int i = 0; i < 1000; i++) {
    final int id = i;
    CompletableFuture<?> c = CompletableFuture
        .runAsync(() -> {
            System.out.println("Running: " + id); 
            if(id == 400) throw new RuntimeException("Exception from: " + id);
        }, executionService);
        c.whenComplete((v, ex) -> {
            if(ex != null && canceled.compareAndSet(false, true)) {
                System.out.println("Canceling.");
                for(CompletableFuture<?> f: tasks) f.cancel(false);
                System.out.println("Canceled.");
            }
        });
    tasks.add(c);
    if(canceled.get()) {
        c.cancel(false);
        break;
    }
}

try {
    CompletableFuture.allOf(tasks.toArray(new CompletableFuture[0])).join();
} catch(Exception e) {
    System.out.println("Got async exception: " + e);
} finally {
    System.out.println("DONE");
}
executionService.shutdown();

一旦相关的 future 被取消，runnables 将不会被执行。由于取消和普通执行之间存在竞争，因此将操作更改为

可能会有所帮助

.runAsync(() -> {
    System.out.println("Running: " + id); 
    if(id == 400) throw new RuntimeException("Exception from: " + id);
    LockSupport.parkNanos(1000);
}, executionService);

模拟一些实际工作量。然后，你会看到遇到异常后执行的动作变少了。

由于异步异常甚至可能在提交循环仍在运行时发生，它使用AtomicBoolean检测这种情况并在这种情况下停止循环。

请注意，对于 CompletableFuture，取消与任何其他异常完成之间没有区别。调用 f.cancel(…) 等同于 f.completeExceptionally(new CancellationException())。因此，由于 CompletableFuture.allOf 在异常情况下报告任何异常，因此很可能是 CancellationException 而不是触发异常。

如果用 complete(null) 替换两个 cancel(false) 调用，您会得到类似的效果，runnables 不会为已经完成的 futures 执行，但是 allOf 会报告原始异常，因为它是当时唯一的异常。它还有另一个积极的影响：用 null 值完成比构建 CancellationException 便宜得多（对于每个未决的未来），因此通过 complete(null) 强制完成运行得更快，防止更多执行期货。

Answer 2

另一种仅依赖于CompletableFuture的解决方案是使用“取消器”未来，这将导致所有未完成的任务在完成时被取消：

Set<CompletableFuture<?>> tasks = ConcurrentHashMap.newKeySet();
CompletableFuture<Void> canceller = new CompletableFuture<>();

for(int i = 0; i < 1000; i++) {
    if (canceller.isDone()) {
        System.out.println("Canceller invoked, not creating other futures.");
        break;
    }
    //LockSupport.parkNanos(10);
    final int id = i;
    CompletableFuture<?> c = CompletableFuture
            .runAsync(() -> {
                //LockSupport.parkNanos(1000);
                System.out.println("Running: " + id);
                if(id == 400) throw new RuntimeException("Exception from: " + id);
            }, executionService);
    c.whenComplete((v, ex) -> {
        if(ex != null) {
            canceller.complete(null);
        }
    });
    tasks.add(c);
}
canceller.thenRun(() -> {
    System.out.println("Cancelling all tasks.");
    tasks.forEach(t -> t.cancel(false));
    System.out.println("Finished cancelling tasks.");
});

从 CompletableFuture 调用 ExecutorService.shutdownNow

Calling ExecutorService.shutdownNow from CompletableFuture

java

java-8

java-threads

completable-future