Especificação da parte do código que decide dividir o problema ou resoler diretamente o problema.


if (problem is small)
  directly solve problem
else {
  split problem into independent parts
  fork new subtasks to solve each part
  join all subtasks
  compose result from subresults
}

Veja a inmlementação desta especifição a seguir:

@Override
  protected void compute() {
    if (data.size() <= SEQUENTIAL_THRESHOLD) { // base case
      long sum = computeSumDirectly();
      System.out.format("Sum of %s: %d\n", data.toString(), sum);
    } else { // recursive case
      // Calcuate new range
      int mid = data.size() / 2;
      SumAction firstSubtask =
          new SumAction(data.subList(0, mid));
      SumAction secondSubtask =
          new SumAction(data.subList(mid, data.size()));

      firstSubtask.fork(); // queue the first task
      secondSubtask.compute(); // compute the second task
      firstSubtask.join(); // wait for the first task result

      // Or simply call
      //invokeAll(firstSubtask, secondSubtask);
    }
  }

Este código está presente no código abaixo:

========================================================================================

Pacotes do Java 7 em diante para rodar o Framework Fork/Join


package <nome do seu pacote>.fork;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.RecursiveTask;
import java.util.concurrent.ThreadLocalRandom;


========================================================================================

Resolvendo com RecursiveAction para obter os resultados de somais parciais.

public class SumAction extends RecursiveAction {
  
  private static final int SEQUENTIAL_THRESHOLD = 5;
  
  private List<Long> data;

  public SumAction(List<Long> data) {
    this.data = data;
  }
  
  @Override
  protected void compute() {
    if (data.size() <= SEQUENTIAL_THRESHOLD) { // base case
      long sum = computeSumDirectly();
      System.out.format("Sum of %s: %d\n", data.toString(), sum);
    } else { // recursive case
      // Calcuate new range
      int mid = data.size() / 2;
      SumAction firstSubtask =
          new SumAction(data.subList(0, mid));
      SumAction secondSubtask =
          new SumAction(data.subList(mid, data.size()));

      firstSubtask.fork(); // queue the first task
      secondSubtask.compute(); // compute the second task
      firstSubtask.join(); // wait for the first task result

      // Or simply call
      //invokeAll(firstSubtask, secondSubtask);
    }
  }
  
  private long computeSumDirectly() {
    long sum = 0;
    for (Long l: data) {
      sum += l;
    }
    return sum;
  }
  
  public static void main(String[] args) {
    Random random = new Random();

    List<Long> data = random
        .longs(10, 1, 5)
        .boxed()
        .collect(toList());

    ForkJoinPool pool = new ForkJoinPool();
    SumAction task = new SumAction(data);
    pool.invoke(task);
  }
  
}


  Sum of [1, 4, 4, 2, 3]: 14
  Sum of [4, 1, 2, 1, 1]: 9

=========================================================================================

Resolvendo com RecursiveTask para obter o resultado da soma total.

public class SumTask extends RecursiveTask<Long> {

 private static final int SEQUENTIAL_THRESHOLD = 5;

  private List<Long> data;

  public SumTask(List<Long> data) {
    this.data = data;
  }

  private long computeSumDirectly() {
    long sum = 0;
    for (Long l: data) {
      sum += l;
    }
    return sum;
  }
  
  @Override
  protected Long compute() {
    if (data.size() <= SEQUENTIAL_THRESHOLD) { // base case
      long sum = computeSumDirectly();
      System.out.format("Sum of %s: %d\n", data.toString(), sum);
      return sum;
    } else { // recursive case
      // Calcuate new range
      int mid = data.size() / 2;
      SumTask firstSubtask =
          new SumTask(data.subList(0, mid));
      SumTask secondSubtask =
          new SumTask(data.subList(mid, data.size()));

      // queue the first task
      firstSubtask.fork();

      // Return the sum of all subtasks
      return secondSubtask.compute()
              +
              firstSubtask.join();
    }
  }

  public static void main(String[] args) {
    Random random = new Random();

    List<Long> data = random
        .longs(10, 1, 5)
        .boxed()     // returns a Stream consisting of the elements of this stream, each boxed to an Integer.
        .collect(toList());

    ForkJoinPool pool = new ForkJoinPool();
    SumTask task = new SumTask(data);
    System.out.println("Sum: " + pool.invoke(task));
  }
}

Sum of [4, 3, 1, 1, 1]: 10
Sum of [1, 1, 1, 2, 1]: 6
Sum: 16

==============================================================================================================

Teste de Desempenho 

1. Execute um teste informal para ver o desempenho da classe SumTask versus uma implementação seqüencial, em uma 
lista de dez milhões de elementos.

public class ComparePerformance {

  public static void main(String[] args) {
    Random random = new Random();

    List<Long> data = random
        .longs(10_000_000, 1, 100)
        .boxed()
        .collect(toList());

    testForkJoin(data);
    //testSequentially(data);
  }

  private static void testForkJoin(List<Long> data) {
    final long start = System.currentTimeMillis();

    ForkJoinPool pool = new ForkJoinPool();
    SumTask task = new SumTask(data);
    pool.invoke(task);

    System.out.println("Executed with fork/join in (ms): " + (System.currentTimeMillis() - start));
  }

  private static void testSequentially(List<Long> data) {
    final long start = System.currentTimeMillis();

    long sum = 0;
    for (Long l: data) {
      sum += l;
    }

    System.out.println("Executed sequentially in (ms): " + (System.currentTimeMillis() - start));
  }
}

=======================================================================================================================

Sabemos que a validade de um teste como este é questionável (usando System.currentTimeMillis() para medir o tempo de execução). Dependendo do hardware que você está testando, você pode obter resultados diferentes.
No entanto, não precisamos nos preocupar muito com os números atuais porque como eles se comparam uns com os outros é muito mais interessante !
Execute o programa dez vezes para cada teste para obter um tempo médio de execução. 

O primeiro teste, você pode ver o Framework Fork/Join, usando um limite threshold de 1000 elementos, depois de um threshold de 100.000 e, em seguida, um threshold de 1 milhão de elementos.  

O segundo teste, você pode ver a implementação sequencial.

Aqui estão os resultados:


     (ver os resultados no link Implementando ...)



=======================================================================================================================
Por outro lado, o método computeSumDirectly na implementação Fork/Join possui o mesmo loop da implementação sequencial 
para somar todos os elementos.
 

2. Mas o que acontece se mudarmos a implementação sequencial para algo como o seguinte:

private static void testSequentiallyStream(List<Long> data) {
  final long start = System.currentTimeMillis();

  data.stream().reduce(0L, Long::sum);

  System.out.println("Executed with a sequential stream in (ms): " + (System.currentTimeMillis() - start));
}


=======================================================================================================================