3 ForkJoinTask¶
Intro¶
La clase ForkJoinTask<V> es la clase base de las tareas que pueden ser ejecutadas por un ejecutor ForkJoinPool, e implementa la interfaz Future<V>.
El objetivo de esta clase es asociar una determinada colección de datos con la computación que se debe realizar con ellos. Proporciona mecanismos para ejecutar las operaciones de descomposición (fork) y de reunificación (join) dentro de la tarea y los métodos de control del estado de la tarea, permitiéndonos utilizar una estrategia divide y vencerás para llevar a cabo el trabajo encomendado.
Para ello, dentro de la tarea usemos algún algoritmo del tipo de descomposición y reunificación recursivo, como el siguiente:
Si el tamaño de la tarea es suficientemente pequeño
Resolver la tarea directamente de forma secuencial
Obtener resultado (opcional)
sino
Subdividir la tarea en subtareas independientes
Ejecutar cada una de las subtareas en paralelo, incluso en hilo diferentes
Esperar a que todas las subtareas terminen de ejecutarse
Obtener resultado componiendo el resultado de cada una de las subtareas (opcional)
Retornar resultado (opcional)
La ventaja principal de aplicar este algoritmo de divide y vencerás es que podemos ejecutar un alto número de ForkJoinTasks en un número pequeño de hilos del fork-join pool, sin tener que sobrecargar el sistema creando muchos hilos.
Como vemos en este sencillo algoritmo, la propia tarea deberá ejecutar una serie de subtareas. Y esta es una de las diferencias más notables entre el Executor Framework y el Fork/Join Framework. Mientras que en el primero todas las tareas deben ser enviadas al ejecutor, en el segundo las propias tareas incluyen métodos para ejecutar y controlar subtareas.
A la hora de crear nuestras tareas no extenderemos directamente la clase ForkJoinTask, sino alguna de sus subclases:
RecursiveAction: Lo usaremos si nuestra tarea no va a retornar ningún resultado.RecursiveTask: Lo usaremos si nuestra tarea retornará un resultado.CountedCompleter: Lo usaremos si al completarse nuestra tarea se deben lanzar otras tareas. Representa árboles jerárquicos de tareas recursivas. Fue agregada en Java 8.
El método principal de estas clases, y que deberemos sobrescribir para implementar su funcionalidad, es el método void compute(), que no retorna nada, en el caso de la clase RecursiveAction, o V compute(), que sí retorna un valor cuyo tipo V corresponde al tipo genérico de la clase, en el caso de la clase RecursiveTask<V>.
Desgraciadamente, estas clases no son interfaces funcionales, por lo que no podemos usar expresiones lambda, sino que estamos obligados a crear clases que hereden de ellas. Debemos tener en cuenta que estas clases fueron diseñadas para Java 7, versión que aún no incorporaba las lambdas.
Dentro de este método se realizará la descomposición de la tarea en subtareas. Una vez creadas las subtareas, deberemos enviárselas al ejecutor ForkJoinPool para que sean ejecutadas. Para ello usaremos alguno de los siguiente métodos existentes en la clase ForkJoinTask, y que son heredados por RecursiveAction y RecursiveTask.
La primera opción es usar el método fork() sobre una subtarea, para que ésta sea enviada internamente al mismo ForkJoinPool, en el que se se inicie su ejecución de manera asíncrona, de manera que la tarea padre continuar su ejecución (no es bloqueado). Como consecuencia, la subtarea es insertada al principio de la cola de trabajo asociada al mismo hilo en el que se está ejecutando la tarea padre. La llamada a este método retorna la propia subtarea sobre la que se ejecuta.
Si la tarea padre desea posteriormente esperar a que termine la subtarea, deberá llamar al método join() de la subtarea, que hará que la tarea padre quede suspendida hasta que la subtarea finalice su ejecución, de manera que el hilo trabajador que estaba ejecutando la tarea padre es asignado a otra tarea del ejecutor (algoritmo de robo de trabajo). Es importante recalcar que el hilo en el que se está ejecutando la tarea padre NO es bloqueado, lo que es bloqueada es la tarea padre en sí, y de hecho el hilo que estaba asignado a la tarea padre es resignado a otra tarea. Cuando la subtarea sobre la que se ha hecho join() es completado, satisfactoriamente o con una excepción, cuando isDone() sea true, la tarea padre es reactivada y recibe el valor retornado el join(), pudiendo continuar con su ejecución.
Si la ejecución de la subtarea retorna un valor, dicho valor será retornado por el propio método join(), de manera que la tarea padre pueda hacer uso de él. Si la tarea no retorna un valor, el método join() retornará null.
Si al ejecutar la subarea se produjera una excepción, al hacer join() sobre ella se lanzará una RuntimeException o Error, pero no ExecutionException, a diferencia de lo que ocurría el método get() de la interfaz Future. Otra diferencia, es que si el hilo desde el que se ha llamado a join(), que está bloqueado esperando que termine de ejecutarse la subtarea, es interrumpido, dicha interrupción no provocará a que la llamada retorne abruptamente lanzando la excepción InterruptedException.
Una segunda opción es usar sobre la subtarea el método invoke() que realiza ambas operaciones, la de fork y la de join, por lo envía la subarea al ForkJoinPool para que sea ejecutada, de manera que la tarea padre quedará suspendida hasta que la subtarea finalice su ejecución, por lo que el hilo trabajador que estaba ejecutando la tarea padre es asignado a otra tarea del ejecutor (algoritmo de robo de trabajo). La llamada a este método retornará el valor retornado por la subtarea. Si la ejecución de la subtarea lanzara una excepción, la llamada al método lanzará una RuntimeException.
Debemos recalcar que fork(), join() y invoke() deben ser ejecutados sobre el objeto subtarea que hayamos creado en el método compute() de la tarea, y NO sobre la tarea padre en sí.
Una tercera opción es que la tarea padre llame al método estático ForkJoinTask.invokeAll(forkJoinSubTask<?>...), que comienza la ejecución de cada una de las subtareas pasadas como argumento, quedando la tarea padre desde la que se realiza la llamada suspendida hasta que terminen de ejecutarse las subtareas, de manera que hilo trabajador que estaba ejecutando la tarea padre es asignado a otra tarea del ejecutor (algoritmo de robo de trabajo). Este método está sobrecargado, de manera que pueda recibir como parámetros un par de subtareas, ForkJoinTask.invokeAll(forkJoinSubTask1, forkJoinSubTask2). La llamada a estos métodos no retorna nada. Este comportamiento ofrece una gestión de tareas más eficiente que la proporcionada por los objetos Runnable y Callable.
Finalmente, también es posible obtener un objeto ForkJoinTask a partir de un objeto Runnable o Callable, mediante los siguientes métodos estáticos:
ForkJoinTask.adapt(runnable): Retorna unForkJoinTask<?>.ForkJoinTask.adapt(runnable, resultT): Retorna unForkJoinTask<T>.ForkJoinTask.adapt(callableT): Retorna unForkJoinTask<T>.