5

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 内のキャンセルをチェックするには、他にどのような方法がありますか?

4

2 に答える 2

7

これFuture<?>は、 がForkJoinTask.AdaptedCallablewhich extendsForkJoinTaskであり、そのキャンセル メソッドが次の場合に発生します。

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のFutureが非常に複雑なツリー構造を持っている可能性があり、どの順序でそれらをキャンセルするかが明確でないために発生すると思われます。

于 2014-01-24T06:49:21.533 に答える
1

@Mkhailの回答に加えて、さらに光を共有します。

submit() の代わりに ForkJoinPool execute() を使用すると、失敗した Runnable が強制的にワーカー例外をスローし、この例外はThread UncaughtExceptionHandler によってキャッチされます。

Java 8 コードから取得:
送信は AdaptedRunnableAction() を使用しています。
execute は 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;
}
于 2018-05-13T16:53:11.117 に答える