Meilleure façon de gérer les exceptions non interceptées dans les tâches/l'action ForkJoinPool

Question

Quelle est la meilleure façon de gérer les exceptions (non interceptées) en utilisant ForkJoinPool pour soumettre des tâches (RecursiveAction ou RecursiveTask)?

ForkJoinPool accepte un Thread.UncaughtExceptionHandler pour gérer les exceptions lorsque le WorkerThread se termine de manière abrupte (ce qui n'est de toute façon pas sous notre contrôle) mais ce gestionnaire n'est pas utilisé lorsque ForkJoinTask déclenche une exception. J'utilise la manière standard submit/invokeAll dans mon implémentation.

Voici mon scénario:

J'ai un fil en cours d'exécution dans des données de lecture de boucle infinie d'un système 3ème partie. Avec dans cette discussion, je soumets les tâches à l'ForkJoinPool

new Thread() { 
     public void run() { 
     while (true) { 
      ForkJoinTask<Void> uselessReturn = 
        ForkJoinPool.submit(RecursiveActionTask); 
     } 
     } 
} 

J'utilise un RecursiveAction et dans quelques scénarios un RecursiveTask. Ces tâches sont soumises à FJPool en utilisant la méthode submit(). Je souhaite disposer d'un gestionnaire d'exceptions générique similaire à UncaughtExceptionHandler où si une tâche génère une exception non contrôlée/non interceptée, je peux traiter l'exception et soumettre de nouveau la tâche si nécessaire. La gestion de l'exception garantit également que les tâches mises en file d'attente ne seront pas annulées si une ou plusieurs des tâches lancent une exception.

méthode invokeAll() retourne un ensemble de ForkJoinTasks mais ces tâches sont dans un bloc récursif (chaque tâche appelle la méthode compute() et peut être divisé plus [scénario hypothétique])

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 
      } 
      } 
     } 
    } 

} 

J'ai remarqué que lorsque 3-4 les tâches échouent toute l'unité de soumission de file d'attente est rejetée. Actuellement, j'ai mis un try/catch autour de processusTask que je n'aime pas personnellement. Je cherche plus générique.

1. Je veux aussi connaître toute la liste des tâches qui ont échoué afin que je puisse les soumettre à nouveau
2. Lorsque les tâches lèvent des exceptions font les fils s'évincés de la piscine (bien que mon analyse a révélé qu'ils doesn 't [mais pas sûr])?
3. L'appel de la méthode get() sur FutureTask mettrait probablement mon flux en séquence tant qu'il attendra que la tâche se termine.
4. Je veux connaître l'état de la tâche seulement si elle échoue. Je ne me soucie pas quand il se termine (évidemment ne veut pas attendre une heure plus tard)

Des idées sur la façon de gérer les exceptions dans le scénario ci-dessus?

Answer 1

Ce que je montre que nous avons résolu 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 
    } 
}