如何在ForkJoinPool中单个任务异常时终止所有任务执行？

问题背景 #

我们有一段基于树结构的并行任务代码，逻辑如下：

• 叶子节点：任务立即执行
• 内部节点（含子节点）：执行结果依赖所有子节点的执行结果

代码使用ForkJoinPool实现并行处理，任务继承RecursiveTask：根节点任务通过pool.execute(task)提交至线程池，子任务通过ForkJoinTask.invokeAll调度。

当前问题：当某一任务抛出异常时，所有计算仍会执行完毕，异常仅在全部任务结束后才抛出，无法快速终止其他任务。

已尝试的解决方案 #

• 为线程池传入UncaughtExceptionHandler，但任务异常未传递至该处理器
• 使用pool.invoke替代execute，无效果
• 在computeSelf方法开头检查任务取消状态isCancelled()，但线程池未自动设置该标记

完整示例代码 #

任务实现类 #

class MyBiTreeTask extends RecursiveTask<Long> {

    final String name;

    MyBiTreeTask(String aName) {
        name = aName;
    }

    @Override
    protected Long compute() {
        long subTiming = computeChildren();
        long selfTiming = computeSelf();
        return subTiming + selfTiming;
    }

    private long computeChildren() {
        if (name.length() >= 5) {
            return 0;
        }
        List<ForkJoinTask<Long>> subTasks = IntStream.range(0, 2)
                                                     .mapToObj(i -> new MyBiTreeTask(name + i))
                                                     .collect(Collectors.toList());
        Collection<ForkJoinTask<Long>> subResults = ForkJoinTask.invokeAll(subTasks);
        AtomicLong subTimings = new AtomicLong(0);
        subResults.forEach(r -> {
            try {
                subTimings.addAndGet(r.get());
            } catch (InterruptedException | ExecutionException aE) {
                throw new RuntimeException(aE);
            }
        });
        return subTimings.get();
    }

    private long computeSelf() {
        long t0 = System.currentTimeMillis();

        if (name.equals("r1100")) {
            throw new IllegalArgumentException("imagine something's wrong here: " + name);
        }
        try {
            // 模拟耗时任务，异常分支外的任务耗时更长
            Thread.sleep(1000 * (name.equals("r0") ? 10 : 1));
        } catch (InterruptedException aE) {
            throw new RuntimeException(aE);
        }

        long t1 = System.currentTimeMillis();
        return t1 - t0;
    }
}

启动代码 #

public static void main(String[] args) {
    long t0 = System.currentTimeMillis();
    try {
        ForkJoinTask<Long> root = new MyBiTreeTask("r");
        ForkJoinPool pool = new ForkJoinPool(ForkJoinPool.getCommonPoolParallelism(),
                                             ForkJoinPool.defaultForkJoinWorkerThreadFactory,
                                             (t, e) -> {
                                                throw new RuntimeException("exception in thread " + t.getId(), e);
                                             },
                                             false);
        pool.execute(root);
        Long accumulatedTiming = root.join();
        System.out.println("accumulated timing = " + accumulatedTiming);
    } catch (Exception aE) {
        System.err.println("couldn't accumulate timings");
        aE.printStackTrace();
    }
    long t1 = System.currentTimeMillis();
    long realTiming = t1 - t0;
    System.out.println("real timing = " + realTiming);
}

预期：当叶子节点r1100抛出异常时，realTiming应较短（仅数秒）；实际：所有子任务执行完毕才抛出异常，耗时过长。

解决方案 #

ForkJoinPool不会自动取消其他任务，需手动实现异常传播与任务取消逻辑，核心步骤如下：

1. 保存父任务引用，实现取消信号向上传播 #

在任务类中添加父任务引用，当子任务抛出异常时，递归取消所有父任务，同时取消未完成的兄弟任务。

2. 在任务执行节点检查取消状态 #

在computeSelf和computeChildren的关键执行点检查任务是否已被取消，若已取消则立即终止执行。

修改后的代码示例 #

class MyBiTreeTask extends RecursiveTask<Long> {

    final String name;
    private final MyBiTreeTask parent;

    MyBiTreeTask(String aName, MyBiTreeTask parent) {
        name = aName;
        this.parent = parent;
    }

    @Override
    protected Long compute() {
        if (isCancelled()) {
            throw new CancellationException("Task cancelled due to upstream failure");
        }
        long subTiming = computeChildren();
        long selfTiming = computeSelf();
        return subTiming + selfTiming;
    }

    private long computeChildren() {
        if (name.length() >= 5 || isCancelled()) {
            return 0;
        }
        List<MyBiTreeTask> subTasks = IntStream.range(0, 2)
                                               .mapToObj(i -> new MyBiTreeTask(name + i, this))
                                               .collect(Collectors.toList());
        ForkJoinTask.invokeAll(subTasks);
        
        long subTimings = 0;
        for (MyBiTreeTask task : subTasks) {
            if (isCancelled()) {
                throw new CancellationException("Task cancelled");
            }
            try {
                subTimings += task.get();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                cancel(true);
                propagateCancelUp();
                subTasks.forEach(t -> t.cancel(true));
                throw new RuntimeException(e);
            } catch (ExecutionException e) {
                cancel(true);
                propagateCancelUp();
                subTasks.forEach(t -> t.cancel(true));
                throw new RuntimeException(e.getCause());
            }
        }
        return subTimings;
    }

    private long computeSelf() {
        if (isCancelled()) {
            throw new CancellationException("Task cancelled");
        }
        long t0 = System.currentTimeMillis();

        if (name.equals("r1100")) {
            throw new IllegalArgumentException("imagine something's wrong here: " + name);
        }
        try {
            // 模拟耗时任务，异常分支外的任务耗时更长
            Thread.sleep(1000 * (name.equals("r0") ? 10 : 1));
        } catch (InterruptedException aE) {
            Thread.currentThread().interrupt();
            throw new RuntimeException(aE);
        }

        long t1 = System.currentTimeMillis();
        return t1 - t0;
    }

    // 向上传播取消信号到所有父任务
    private void propagateCancelUp() {
        MyBiTreeTask currentParent = parent;
        while (currentParent != null && !currentParent.isCancelled()) {
            currentParent.cancel(true);
            currentParent = currentParent.parent;
        }
    }
}

修改后的启动代码 #

public static void main(String[] args) {
    long t0 = System.currentTimeMillis();
    try {
        ForkJoinTask<Long> root = new MyBiTreeTask("r", null);
        ForkJoinPool pool = new ForkJoinPool(ForkJoinPool.getCommonPoolParallelism(),
                                             ForkJoinPool.defaultForkJoinWorkerThreadFactory,
                                             (t, e) -> {
                                                throw new RuntimeException("exception in thread " + t.getId(), e);
                                             },
                                             false);
        pool.execute(root);
        Long accumulatedTiming = root.join();
        System.out.println("accumulated timing = " + accumulatedTiming);
    } catch (Exception aE) {
        System.err.println("couldn't accumulate timings");
        aE.printStackTrace();
    }
    long t1 = System.currentTimeMillis();
    long realTiming = t1 - t0;
    System.out.println("real timing = " + realTiming);
}

原理说明 #

• 当子任务抛出异常时，首先取消当前任务，然后通过propagateCancelUp将取消信号传递至根任务，确保整个任务树都收到终止信号
• 在任务执行的关键节点（compute、computeChildren、computeSelf开头）检查取消状态，一旦发现已取消则立即终止执行
• 捕获异常时主动取消所有兄弟子任务，避免不必要的计算

内容的提问来源于stack exchange，提问作者geronimo