ForkJoinPool 重置线程中断状态

Question

我刚刚在取消ForkJoinPool 返回的Future 时注意到以下现象。给定以下示例代码：

ForkJoinPool pool = new ForkJoinPool();
Future<?> fut = pool.submit(new Callable<Void>() {

  @Override
  public Void call() throws Exception {
    while (true) {
      if (Thread.currentThread().isInterrupted()) { // <-- never true
        System.out.println("interrupted");
        throw new InterruptedException();
      }
    }
  }
});

Thread.sleep(1000);
System.out.println("cancel");
fut.cancel(true);

程序从不打印interrupted。 ForkJoinTask#cancel(boolean) 的文档说：

mayInterruptIfRunning - 此值在默认实现中无效，因为中断不用于控制取消。

如果 ForkJoinTasks 忽略中断，您还应该如何检查提交到 ForkJoinPool 的 Callables 中的取消？

Answer 1

在@Mkhail 回答的基础上分享更多亮点：

使用 ForkJoinPool execute() 而不是 submit() 将强制失败的 Runnable 抛出一个 worker 异常，并且 此异常将被 Thread UncaughtExceptionHandler 捕获。

取自 Java 8 代码：
提交正在使用 AdaptedRunnableAction()。
执行正在使用 RunnableExecuteAction()（参见 rethrow(ex)）。

 /**
 * Adaptor for Runnables without results
 */
static final class AdaptedRunnableAction extends ForkJoinTask<Void>
    implements RunnableFuture<Void> {
    final Runnable runnable;
    AdaptedRunnableAction(Runnable runnable) {
        if (runnable == null) throw new NullPointerException();
        this.runnable = runnable;
    }
    public final Void getRawResult() { return null; }
    public final void setRawResult(Void v) { }
    public final boolean exec() { runnable.run(); return true; }
    public final void run() { invoke(); }
    private static final long serialVersionUID = 5232453952276885070L;
}

/**
 * Adaptor for Runnables in which failure forces worker exception
 */
static final class RunnableExecuteAction extends ForkJoinTask<Void> {
    final Runnable runnable;
    RunnableExecuteAction(Runnable runnable) {
        if (runnable == null) throw new NullPointerException();
        this.runnable = runnable;
    }
    public final Void getRawResult() { return null; }
    public final void setRawResult(Void v) { }
    public final boolean exec() { runnable.run(); return true; }
    void internalPropagateException(Throwable ex) {
        rethrow(ex); // rethrow outside exec() catches.
    }
    private static final long serialVersionUID = 5232453952276885070L;
}

Answer 2

这是因为Future<?> 是一个扩展ForkJoinTask 的ForkJoinTask.AdaptedCallable，其取消方法是：

public boolean cancel(boolean mayInterruptIfRunning) {
    return setCompletion(CANCELLED) == CANCELLED;
}

private int setCompletion(int completion) {
    for (int s;;) {
        if ((s = status) < 0)
            return s;
        if (UNSAFE.compareAndSwapInt(this, statusOffset, s, completion)) {
            if (s != 0)
                synchronized (this) { notifyAll(); }
            return completion;
        }
    }
}

它不做任何中断，它只是设置状态。我想这是因为ForkJoinPools 的Futures 可能有一个非常复杂的树结构，并且不清楚以什么顺序取消它们。