Fork/Join
A funcionalidade fork/join
, presente a partir do Java 7, pode ser um pouco difícil de entender a princípio, pois ela atende uma classe específica de problemas.
Segundo a documentação da classe ForkJoinTask
, a ideia é permitir um método mais eficiente de processamento paralelo impondo algumas restrições sobre o modo como as threads operam. O objetivo é conseguir dividir um problema em tarefas menores e executá-las independentemente, sem uso de sincronização (synchronized
).
Se pensarmos um pouco, isso resolve muito bem questões de Programação Dinâmica. Um dos exemplos clássicos é o número de Fibonacci. O código a seguir é um exemplo de uso do fork/join
que calcula o número de Fibonacci:
public class Fibonacci extends RecursiveTask<Long> {
long n;
public Fibonacci(long n) {
this.n = n;
}
@Override
protected Long compute() {
if (n <= 1) {
return n;
}
Fibonacci f1 = new Fibonacci(n - 1);
f1.fork();
Fibonacci f2 = new Fibonacci(n - 2);
return f2.compute() + f1.join();
}
}
Podemos salientar os seguintes pontos:
- Estendemos
RecursiveTask
, uma das duas implementações disponíveis de ForkJoinTask
. Uma "tarefa recursiva" tem o propósito de executar subtarefas recursivas em paralelo e retornar um valor final.
- Dentro do método
compute()
ocorre a "mágica", onde definimos o limite da recursar no primeiro if
e invocamos recursivamente instâncias da classe em novas threads.
- Ao invocar
f1.fork()
, estamos pedindo para que o cálculo de f(n-2)
seja feito em outra thread. Isso libera o processamento paralelo de f(n-1)
.
- Finalmente, invocamos
f2.compute()
para calcular o valor imediatamente e f1.join()
para recuperar o valor do fork
ou aguardar que o processamento termine.
Você pode executar o código acima da seguinte forma:
ForkJoinPool pool = new ForkJoinPool(4);
Fibonacci fibonacci = new Fibonacci(10);
long resultado = pool.invoke(fibonacci);
System.out.println(resultado);
A class ForkJoinPool
permite o gerenciamento das tarefas paralelas e o parâmetro do construtor estabelece o nível de paralelismo, isto é, quantas threads serão usadas simultaneamente.
O método pool.invoke(fibonacci)
inicia o processamento, aguarda o cálculo e retorna o número calculado.
ThreadPoolExecutor
Porém, se o seu problema não entra na categoria de problema onde "dividir para conquistar" recursivamente é a melhor estratégia, você pode usar APIs mais genéricas, tal como ThreadPoolExecutor
.
Ela faz parte do mesmo pacote que fork/join
, pois ambas implementam ExecutorService
, mas sem a natureza recursiva e limitada da "irmã".
Para criar uma instência de ThreadPoolexecutor
:
ExecutorService threadPoolExecutor = new ThreadPoolExecutor(
4, //tamanho inicial do pool
8, //tamanho máximo do pool
10000, //tempo de espera máximo para cada thread na fila
TimeUnit.MILLISECONDS, //unidade de tempo
new LinkedBlockingQueue<Runnable>() //implementação da fila de theads
);
Depois, pedimos para executar threads e usamos futures para guardar a promessa de um resultado futuro:
Future<Integer> f1 = threadPoolExecutor.submit(new Callable<Integer>() {
@Override
public Integer call() throws Exception {
return 1;
}
});
Future<Integer> f2 = threadPoolExecutor.submit(new Callable<Integer>() {
@Override
public Integer call() throws Exception {
return 2;
}
});
Nota que o código acima usa o método submit()
para solicitar a execução de dois Callable
s. Um Callable
representa uma thread, assim como um Runnable
, mas retornando um valor.
A classe Future
guarda uma referência para a thread de forma que podemos continuar a execução e recuperar o resultado quando estiver pronto, assim:
try {
Integer r1 = f1.get(1000, TimeUnit.MILLISECONDS);
Integer r2 = f2.get(1000, TimeUnit.MILLISECONDS);
System.out.println(r1 + r2);
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
} catch (TimeoutException e) {
e.printStackTrace();
}
No exemplo acima, usei o método get()
das instâncias de Future
para aguardar o processamento e recuperar efetivamente o resultado.
Os parâmetros indicam que o tempo limite de espera é de mil milissegundos. Se passar disso, um TimeoutException
será lançado.
Considerações finais
A escolha entre usar threads clássicas, fork/join
, ThreadPoolExecutor
ou qualquer outro mecanismo depende da natureza do problema que você tenta resolver. Uma abordagem não é necessariamente melhor que as outras para todos os tipos de problemas.
Além disso, determinadas situações exigirão implementações mais específicas, onde você vai precisar estender as classes e interfaces da API java.util.concurrent
. A boa notícia é que essa API é feita com isso em mente.