处理 ForkJoinPool 任务/操作中未捕获异常的更好方法

Question

使用 ForkJoinPool 提交任务（RecursiveAction 或 RecursiveTask）？

ForkJoinPool 接受 Thread.UncaughtExceptionHandler来处理 WorkerThread 突然终止时的异常（无论如何，这不在我们的控制之下），但是当 ForkJoinTask 抛出一个例外。我正在使用标准 提交/在我的实现中调用All 方式。

这是我的场景：

我有一个线程在无限循环中运行，从第三方系统读取数据。在此线程中，我将任务提交到 ForkJoinPool

新线程() {
      公共无效运行（）{
         而（真）{
             ForkJoinTask;无用返回= 
                   ForkJoinPool.submit(RecursiveActionTask);
         }
      }
 }

我正在使用 RecursiveAction，并且在少数情况下使用 RecursiveTask。这些任务使用submit()方法提交到FJPool。 我想要一个类似于 UncaughtExceptionHandler 的通用异常处理程序，如果任务抛出未检查/未捕获的异常，我可以处理该异常并在需要时重新提交任务。处理异常还可以确保如果一个/某些任务抛出异常，排队的任务不会被取消。

invokeAll() 方法返回一组 ForkJoinTask，但这些任务位于递归块中（每个任务调用 compute() 方法，并且可以进一步拆分[假设场景]）

class RecursiveActionTask extends RecursiveAction {

    public void compute() {
       if <task.size() <= ACCEPTABLE_SIZE) {
          processTask() // this might throw an checked/unchecked exception
       } else {
          RecursiveActionTask[] splitTasks = splitTasks(tasks)
          RecursiveActionTasks returnedTasks = invokeAll(splitTasks);
          // the below code never executes as invokeAll submits the tasks to the pool 
          // and the flow never comes to the code below.
          // I am looking for some handling like this
          for (RecusiveActionTask task : returnedTasks) {
             if (task.isDone()) {
                task.getException() // handle this exception
             }
          }
       }
    }

}

我注意到，当 3-4 个任务失败时，整个队列提交单元都会被丢弃。目前，我在 processTask 周围放置了一个我个人不喜欢的 try/catch 。我正在寻找更通用的。

1. 我还想知道所有失败的任务列表，以便我可以重新提交它们
2. 当任务抛出异常时，线程是否会从池中逐出（尽管我的分析发现它们不会[但不确定]）？
3. 在 FutureTask 上调用 get() 方法更有可能使我的流程按顺序排列，因为它会等待任务完成。
4. 我只想知道任务失败时的状态。我不在乎它何时完成（显然不想等一个小时后）

有什么想法如何处理上述场景中的异常吗？

Answer 1

这就是我们在 Akka 中解决这个问题的方法：

/**
 * INTERNAL AKKA USAGE ONLY
 */
final class MailboxExecutionTask(mailbox: Mailbox) extends ForkJoinTask[Unit] {
  final override def setRawResult(u: Unit): Unit = ()
  final override def getRawResult(): Unit = ()
  final override def exec(): Boolean = try { mailbox.run; true } catch {
    case anything ⇒
      val t = Thread.currentThread
      t.getUncaughtExceptionHandler match {
        case null ⇒
        case some ⇒ some.uncaughtException(t, anything)
      }
      throw anything
  }
 }

Answer 2

@Rajendra，你做的一切都是正确的，除了你应该使用 ForkJoinPoolexecute() 而不是submit()。

这样，如果 Runnable 失败，它将强制产生一个工作异常，并且它将被 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;
}