Concurrencia de Java: marco de unión de bifurcación

El framework fork-join permite dividir una determinada tarea en varios trabajadores y luego esperar el resultado para combinarlos. Aprovecha en gran medida la capacidad de la máquina multiprocesador. A continuación se muestran los conceptos y objetos básicos que se utilizan en el marco de unión de bifurcación.

Tenedor

Fork es un proceso en el que una tarea se divide en subtareas más pequeñas e independientes que se pueden ejecutar al mismo tiempo.

Sintaxis

Sum left  = new Sum(array, low, mid);
left.fork();

Aquí Sum es una subclase de RecursiveTask y left.fork () divide la tarea en subtareas.

Unirse

Join es un proceso en el que una tarea une todos los resultados de las subtareas una vez que las subtareas han terminado de ejecutarse, de lo contrario sigue esperando.

Sintaxis

left.join();

Aquí a la izquierda hay un objeto de la clase Sum.

HorquillaUnirsePiscina

es un grupo de subprocesos especial diseñado para trabajar con la división de tareas de bifurcación y unión.

Sintaxis

ForkJoinPool forkJoinPool = new ForkJoinPool(4);

Aquí un nuevo ForkJoinPool con un nivel de paralelismo de 4 CPU.

RecursiveAction

RecursiveAction representa una tarea que no devuelve ningún valor.

Sintaxis

class Writer extends RecursiveAction {
   @Override
   protected void compute() { }
}

RecursiveTask

RecursiveTask representa una tarea que devuelve un valor.

Sintaxis

class Sum extends RecursiveTask<Long> {
   @Override
   protected Long compute() { return null; }
}

Ejemplo

El siguiente programa TestThread muestra el uso del marco Fork-Join en un entorno basado en subprocesos.

import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;

public class TestThread {

   public static void main(final String[] arguments) throws InterruptedException, 
      ExecutionException {
      
      int nThreads = Runtime.getRuntime().availableProcessors();
      System.out.println(nThreads);
      
      int[] numbers = new int[1000]; 

      for(int i = 0; i < numbers.length; i++) {
         numbers[i] = i;
      }

      ForkJoinPool forkJoinPool = new ForkJoinPool(nThreads);
      Long result = forkJoinPool.invoke(new Sum(numbers,0,numbers.length));
      System.out.println(result);
   }  

   static class Sum extends RecursiveTask<Long> {
      int low;
      int high;
      int[] array;

      Sum(int[] array, int low, int high) {
         this.array = array;
         this.low   = low;
         this.high  = high;
      }

      protected Long compute() {
         
         if(high - low <= 10) {
            long sum = 0;
            
            for(int i = low; i < high; ++i) 
               sum += array[i];
               return sum;
         } else {	    	
            int mid = low + (high - low) / 2;
            Sum left  = new Sum(array, low, mid);
            Sum right = new Sum(array, mid, high);
            left.fork();
            long rightResult = right.compute();
            long leftResult  = left.join();
            return leftResult + rightResult;
         }
      }
   }
}

Esto producirá el siguiente resultado.

Salida

32
499500