java - Future 和 ExecutorService，如何知道取消的任务何时终止？

Question

假设我有一些使用 ExecutorService 启动任务的代码。然后调用者通过 Future 取消它由 submit() 返回方法:

execService = Executors.newSingleThreadExecutor ();
Future<String> result = execService.submit ( () -> {
  for ( ... ) {
    if ( Thread.interrupted() ) break;
    // Stuff that takes a while
  }
  return "result";
});
...
result.cancel ( true );
// I'm not sure the corresponding thread/task has finished and the call() method 
// above returned
// result.isDone() is immediately true. result.get() throws CancellationException

因此，我可以取消后台任务并使执行程序准备好重新启动。但是，在被中断的任务完成中断并且相应的方法返回之前，我不能做后者。

请注意，在上面的代码中，主要流程在 result.cancel() 之后直接返回和 result.isCancelled()紧接着就是如此，同时，并行任务可能还需要一段时间才能检查 Thread.interrupted()再次并终止。在继续主线程之前，我需要确保辅助任务已完全完成。

此外，请注意，单线程执行器的使用是偶然的，除了这个简单的例子，我想解决只有一个并行线程的情况和运行多个线程的情况的问题。

确定这一点的最佳方法是什么？

到目前为止，我已经考虑过引入一个对任务和调用者都可见的标志，或者关闭执行程序并等待它完成。如果想要多次重用执行器，后一种解决方案可能效率低下，前者更好，但我想知道是否有另一种更简单或更规范的方法。

Answer 1

我也遇到了类似的问题，我最终制作了这个实用程序:

    private enum TaskState {
        READY_TO_START,
        RUNNING,
        DONE,
        MUST_NOT_START;
    }
    
    /**
     * Runs the given set of tasks ensuring that:
     * 
     * If one fails this will not return while tasks are running and once returned
     * no more tasks will start.
     * 
     * 
     * @param <V>
     * @param pool
     * @param tasksToRun
     */
    public <V> void runAndWait(ExecutorService pool, List<Callable<V>> tasksToRun) {
        
        // We use this to work around the fact that the future doesn't tell us if the task is actually terminated or not.
        List<AtomicReference<TaskState>> taskStates = new ArrayList<>();
        List<Future<V>> futures = new ArrayList<>();
        
        for(Callable<V> c : tasksToRun) {
            AtomicReference<TaskState> state = new AtomicReference<>(TaskState.READY_TO_START);
            futures.add(pool.submit(new Callable<V>() {

                @Override
                public V call() throws Exception {
                    if(state.compareAndSet(TaskState.READY_TO_START, TaskState.RUNNING)) {
                        try {
                            return c.call();
                        } finally {
                            state.set(TaskState.DONE);
                        }
                    } else {
                        throw new CancellationException();
                    }
                }
                
            }));
            taskStates.add(state);
        }
        int i = 0;
        try {
            
            // Wait for all tasks.
            for(; i < futures.size(); i++) {
                futures.get(i).get(7, TimeUnit.DAYS);
            }
        } catch(Throwable t) { 
            try {
                // If we have tasks left that means something failed.
                List<Throwable> exs = new ArrayList<>();
                final int startOfRemaining = i;
                
                // Try to stop all running tasks if any left.
                for(i = startOfRemaining; i < futures.size(); i++) {
                    try {
                        futures.get(i).cancel(true);
                    } catch (Throwable e) {
                        // Prevent exceptions from stopping us from
                        // canceling all tasks. Consider logging this.
                    }
                }
                
                // Stop the case that the task is started, but has not reached our compare and set statement. 
                taskStates.forEach(s -> s.compareAndSet(TaskState.READY_TO_START, TaskState.MUST_NOT_START));
                
                for(i = startOfRemaining; i < futures.size(); i++) {
                    try {
                        futures.get(i).get();
                    } catch (InterruptedException e) {
                        break; // I guess move on, does this make sense should we instead wait for our tasks to finish?
                    } catch (CancellationException e) {
                        // It was cancelled, it may still be running if it is we must wait for it.
                        while(taskStates.get(i).get() == TaskState.RUNNING) {
                            Thread.sleep(1);
                        }
                    } catch (Throwable t1) {
                        // Record the exception if it is interesting, so users can see why it failed.
                        exs.add(t1);
                    }
                }
                
                exs.forEach(t::addSuppressed);
            } finally {
                throw new RuntimeException(t);
            }
        }
    }

我最终将每个任务包装起来，这样我就可以知道是否所有任务都完成了。