Future.cancel(true) 不能可靠地取消/中断线程

Question

我尝试使用 CompletionService 并行执行多个任务。当我尝试实施取消时，问题就出现了。

这是我使用的代码草图：

void startTasks(int numberOfTasks) throws Exception {

    ExecutorService executor = Executors.newFixedThreadPool(numberOfThreads);
    CompletionService<TaskResultType> completionService = new ExecutorCompletionService<TaskResultType>(executor);
    ConcurrentLinkedQueue<TaskResultType> results = new ConcurrentLinkedQueue<BenchmarkResult>();
    ArrayList<Future> futures = new ArrayList<Future>();

    for (int i = 0; i < numberOfTasks ; i++) {
        TypeOfTask task = ... ; 
        Future future = completionService.submit(task);
        futures.add(future);
    }

    boolean failed = false;
    Throwable cause = null;
    for (int i = 0; i < numberOfThreads; i++) {
        try {
            Future<TaskResultType> resultFuture = completionService.take();
            TaskResultType result = resultFuture.get();
            results.add(result);
        } catch (ExecutionException e) {
            failed = true;
            cause = e.getCause();
            /* cancel all other running tasks in case of failure in one task */
            for (Future future : futures) {
                future.cancel(true);
            }
        } catch (CancellationException e) {
            // consume (planned cancellation from calling future.cancel())
        }

    }
    executor.shutdown();
    // code to throw an exception using cause
}

任务实现Callable。

当我现在在大多数情况下在其中一个任务中抛出异常时，它运行良好，即我立即从其他任务中获取 CancellationExceptions，并且任务立即完成（让我们称之为案例 A）。但有时（我们称之为案例 B），一些任务先完成然后抛出 CancellationException。 Future.cancel(true) 在这两种情况下对所有任务都返回 true（除了具有初始 ExecutionException 的那个，因为这个已经被取消了）。

我使用 Thread.currentThread.isInterrupted() 检查中断标志，在完成的任务（即取消不成功的任务）中，中断标志设置为 false。

在我看来，这一切似乎都是非常奇怪的行为。有人知道问题可能是什么吗？

更新

到目前为止我最好的想法是，在包含任务的代码深处的某个地方（只有一些高级代码来自我自己），中断状态被消耗，例如通过捕获的 InterruptedException 不调用 Thread.interrupt() 来重新建立状态。由于线程的调度，Future.cancel() 设置中断标志的确切时间可能会略有不同，这可以解释不一致的行为。消费中断状态能否解释案例 B 的行为？

Answer 1

但有时，有些任务会先完成，然后再抛出 CancellationException。

会不会是你取消了一个任务，它通常是中断的（在这种状态下你可能认为它返回了一个结果，但是对于 CompletionService 它被取消了），future 是由take() 返回的，你调用future.get() 还有CancellationException。

你也可以看看 Guava 的 Futures.allAsList，它似乎在做一个非常相似的事情：

创建一个新的 ListenableFuture，其值是一个列表，其中包含其所有输入期货的值（如果全部成功）。如果任何输入失败，则返回的 future 失败。

Answer 2

但有时，某些任务先完成，然后抛出 CancellationException。 Future.cancel(true) 在这两种情况下对所有任务都返回 true（除了具有初始 ExecutionException 的那个，因为这个已经被取消了）。

如果您的程序未完成（或在引发异常后需要很长时间才能完成），那么我怀疑您的问题是其中一项任务正在执行 IO 或以其他方式阻塞并且未检查 Thread.currentThread().isInterrupted()。因此，即使您取消了 Future 并且线程被中断，也不会被检测到。

但是，该程序似乎即将结束。所以我不确定这里的错误情况是什么。如果您发现异常，您可以在列表中的所有期货上调用future.cancel(true)。投掷的和已经完成的都应该返回false，因为它们不能被取消。从取消返回true的那些应该已经被中断了。

例如。如果倒数第二个线程抛出异常，那么future.cancel(true) 应该只返回最后一个线程运行的true。

要做的一件事是在未来完成时删除它们，这样您就不会重新取消已经完成的工作。但这可能会掩盖您现在看到的问题：

Future<TaskResultType> resultFuture = completionService.take();
futures.remove(resultFuture);

更新：

很有可能某些代码正在吞噬中断。不幸的是，它一直在发生。如果某些线程在被取消并运行到完成时没有立即完成，那么这可能就是正在发生的事情。