如何取消多线程繁忙任务的未来？

Question

在我的代码中，我必须运行一个大量使用递归和并行流处理的任务，以便深入了解可能的游戏动作树并确定最佳动作。这需要很多时间，所以为了防止用户等待太长时间让计算机“思考”，我想设置一个时间，比如 1000 毫秒。如果在 1000 毫秒内没有找到最佳着法，则计算机将随机播放。 我的问题是，虽然我在 Future 上调用了取消（可能中断设置为 true），但任务没有中断，繁忙的线程继续在后台运行。 我试图定期检查当前的 isInterrupted()，然后尝试退出，但这没有帮助。 有什么想法吗？

下面是我的代码：

public Move bestMove() {
    ExecutorService executor = Executors.newSingleThreadExecutor();
    Callable<Move> callable = () -> bestEntry(bestMoves()).getKey();
    Future<Move> future = executor.submit(callable);
    try {
        return future.get(1000, TimeUnit.MILLISECONDS);
    } catch (InterruptedException e) {
        System.exit(0);
    } catch (ExecutionException e) {
        throw new RuntimeException(e);
    } catch (TimeoutException e) {
        future.cancel(true);
        return randomMove();
    }
    return null;
}

private Move randomMove() {
    Random random = new Random();
    List<Move> moves = state.possibleMoves();
    return moves.get(random.nextInt(moves.size()));
}

private <K> Map.Entry<K, Double> bestEntry(Map<K, Double> map) {
    List<Map.Entry<K, Double>> list = new ArrayList<>(map.entrySet());
    Collections.sort(list, (e1, e2) -> (int) (e2.getValue() - e1.getValue()));
    return list.get(0);
}

private <K> Map.Entry<K, Double> worstEntry(Map<K, Double> map) {
    List<Map.Entry<K, Double>> list = new ArrayList<>(map.entrySet());
    Collections.sort(list, (e1, e2) -> (int) (e1.getValue() - e2.getValue()));
    return list.get(0);
}

private Map<Move, Double> bestMoves() {
    Map<Move, Double> moves = new HashMap<>();
    state.possibleMoves().stream().parallel().forEach(move -> {
        if (!Thread.currentThread().isInterrupted()) {
            Game newState = state.playMove(move);
            Double score = newState.isTerminal() ? newState.utility()
                    : worstEntry(new (newState).worstMoves()).getValue();
            moves.put(move, score);
        }
    });
    return moves;
}

private Map<Move, Double> worstMoves() {
    Map<Move, Double> moves = new HashMap<>();
    state.possibleMoves().stream().parallel().forEach(move -> {
        if (!Thread.currentThread().isInterrupted()) {
            Game newState = state.playMove(move);
            Double score = newState.isTerminal() ? -newState.utility()
                    : bestEntry(new (newState).bestMoves()).getValue();
            moves.put(move, score);
        }
    });
    return moves;
}

ps：我也试过不使用“parallel()”，但仍然有一个线程在运行。

提前致谢。

Answer 1

谢谢大家的回答。我想我找到了一个更简单的解决方案。 首先，我认为future.cancel(true) 不起作用的原因是它可能只在启动任务的线程上设置了中断标志。 （即与未来关联的线程）。 然而，由于任务本身使用并行流处理，它在不同的线程上产生工作者，这些线程永远不会被中断，因此我不能定期检查 isInterrupted() 标志。 我发现的“解决方案”（或者更多的解决方法）是在我的算法对象中保留我自己的中断标志，并在取消任务时手动将其设置为 true。因为所有线程都在同一个实例上工作，所以它们都可以访问中断标志并且它们服从。

Answer 2

Future.cancel 只需将线程设置为interrupted，那么您的代码必须如下处理：

public static void main(String[] args) throws InterruptedException {

    final ExecutorService executor = Executors.newSingleThreadExecutor();
    final Future<Integer> future = executor.submit(() -> count());
    try {
        System.out.println(future.get(1, TimeUnit.SECONDS));
    } catch (Exception e){
        future.cancel(true);
        e.printStackTrace();
    }finally {
        System.out.printf("status=finally, cancelled=%s, done=%s%n", future.isCancelled(), future.isDone());
        executor.shutdown();
    }

}

static int count() throws InterruptedException {
    while (!Thread.interrupted());
    throw new InterruptedException();
}

正如您所看到的count 不断检查线程是否可以继续运行，您必须了解，实际上并不能保证如果她不想停止正在运行的线程。

参考：

• Why set the interrupt bit in a Callable
• how to suspend thread using thread's id?

---

更新 2017-11-18 23:22

我编写了一个 FutureTask 扩展，即使代码不尊重interrupt 信号，它也能够尝试停止线程。请记住，它是不安全的，因为Thread.stop 方法is deprecated，无论如何它正在工作，如果你真的需要你可以使用它（请阅读 Thread.stop 弃用说明之前，例如，如果你正在使用锁，然后运行 ​​.stop 会导致死锁）。

测试代码

public static void main(String[] args) throws InterruptedException {
    final ExecutorService executor = newFixedSizeExecutor(1);
    final Future<Integer> future = executor.submit(() -> count());
    try {
        System.out.println(future.get(1, TimeUnit.SECONDS));
    } catch (Exception e){
        future.cancel(true);
        e.printStackTrace();
    }
    System.out.printf("status=finally, cancelled=%s, done=%s%n", future.isCancelled(), future.isDone());
    executor.shutdown();
}

static int count() throws InterruptedException {
    while (true);
}

自定义执行器

static ThreadPoolExecutor newFixedSizeExecutor(final int threads) {
    return new ThreadPoolExecutor(threads, threads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>()){
        protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
            return new StoppableFutureTask<>(new FutureTask<>(callable));
        }
    };
}

static class StoppableFutureTask<T> implements RunnableFuture<T> {

    private final FutureTask<T> future;
    private Field runnerField;

    public StoppableFutureTask(FutureTask<T> future) {
        this.future = future;
        try {
            final Class clazz = future.getClass();
            runnerField = clazz.getDeclaredField("runner");
            runnerField.setAccessible(true);
        } catch (Exception e) {
            throw new Error(e);
        }
    }

    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
        final boolean cancelled = future.cancel(mayInterruptIfRunning);
        if(cancelled){
            try {
                ((Thread) runnerField.get(future)).stop();
            } catch (Exception e) {
                throw new Error(e);
            }
        }
        return cancelled;
    }

    @Override
    public boolean isCancelled() {
        return future.isCancelled();
    }

    @Override
    public boolean isDone() {
        return future.isDone();
    }

    @Override
    public T get() throws InterruptedException, ExecutionException {
        return future.get();
    }

    @Override
    public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
        return future.get(timeout, unit);
    }

    @Override
    public void run() {
        future.run();
    }
}

输出

java.util.concurrent.TimeoutException
    at java.util.concurrent.FutureTask.get(FutureTask.java:205)
    at com.mageddo.spark.sparkstream_1.Main$StoppableFutureTask.get(Main.java:91)
    at com.mageddo.spark.sparkstream_1.Main.main(Main.java:20)
status=finally, cancelled=true, done=true

Process finished with exit code 0