Programação concorrente com java.util.concurrent (parte 1)

Atualmente para tirar um melhor proveito dos processadores com múltiplos núcleos, é necessário que os desenvolvedores se preocupem com a execução de tarefas em paralelo. Para facilitar nisso, foi adicionado o pacote chamado java.util.concurrent no Java 5. Neste post vou apresentar através de alguns exemplos algumas facilidades que as classes deste pacote provêem.

Exemplo 1: Execução assíncrona

Download do Exemplo

Este primeiro exemplo, baseado neste exercício, consiste no cálculo do valor do PI otimizado para rodar em vários núcleos separando-o em várias tarefas. Este valor pode ser calculado aproximadamente pela série de Gregory utilizando a seguinte fórmula:

Ou seja, teremos o seguinte nas primeiras iterações: PI/4 = 1 – 1/3 + 1/5 – 1/7 + …

Quanto mais iterações, mais preciso o valor fica. Veja esse link: http://www.eveandersson.com/pi/gregory-leibniz

O código abaixo realiza o cálculo do somatório recebendo o valor inicial (k) e o final (n). Veja que nossa classe implementa a interface Callable através do método call(), o qual retorna o tipo Double como resultado do somatório.

import java.util.concurrent.Callable;

public class CalculoPI implements Callable<Double> {

private int inicio;
private int fim;

public CalculoPI(int inicio, int fim) {
this.inicio = inicio;
this.fim = fim;
}

public Double call() throws Exception {
double valor = 0.0;

for (int i=inicio; i <= fim; i++) {
valor += Math.pow(-1.0, i + 1) / (2.0 * (double)i - 1.0);
}

return valor;
}
}

Para separarmos este cálculo em várias tarefas concorrentes podemos definir os valores iniciais e finais de cada uma delas. Veja o código a seguir:

import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class Main {

public static void main(String[] args) throws InterruptedException, ExecutionException {
//cria um pool de threads para realizar o cálculo
ExecutorService es = Executors.newCachedThreadPool();

//separa o cálculo em 4 partes definindo o valor de n inicial e final para cada uma
Future<Double> parte1 =  es.submit(new CalculoPI(1,100000000));
Future<Double> parte2 = es.submit(new CalculoPI(100000001,200000000));
Future<Double> parte3 = es.submit(new CalculoPI(200000001,300000000));
Future<Double> parte4 = es.submit(new CalculoPI(300000001,400000000));

//junta os valores cálculados das 4 partes e multiplica por 4
double pi = 4.0 * (parte1.get() + parte2.get() + parte3.get() + parte4.get());

        es.shutdown();

System.out.println("Valor calculado de PI é " + pi);
}
}

Aqui foram utilizadas 500.000.000 iterações, as quais são dividas em 4 partes. Inicialmente, foi criado um pool de threads através de Executors.newCachedThreadPool() que retorna um objeto ThreadPoolExcutor configurado para criar novas threads assim que necessário e reaproveitar as já existentes que estejam disponíveis. Você pode também utilizar métodos como Executors.newFixedThreadPool(int) para criar um pool com número fixo de threads ou criar um objeto do tipo ThreadPoolExcutor explicitamente definindo seus parâmetros no construtor. Em seguida, as partes dos cálculos foram submetidas para execução de forma assíncrona, ou seja, ao chamar submit() não vamos aguardar o cálculo finalizar porque ele será executado em paralelo.

Quando um serviço é submetido, é retornado um objeto Future. É através dele que pegamos o resultado do cálculo na linha 19 chamando seu método get(). Este método é bloqueante, ou seja, ao chamá-lo é necessário aguardar o final da execução da tarefa.

Exemplo 2: Produtor/Consumidor com BlockingQueue

Download do Exemplo

Implementar esse algoritmo usando apenas blocos de código com synchronized e os métodos wait() e notify()/notifyAll() muitas vezes é trabalhoso. Para facilitar isso, podemos utilizar a collection BlockingQueue. Veja abaixo como fica simples o código:

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public class Main {

public static void main(String[] args) {
//cria fila bloqueante com 10 lugares disponíveis
BlockingQueue<Integer> queue = new LinkedBlockingQueue<Integer>(10);

//cria e inicia thread para o produtor
new Thread(new Produtor(queue)).start();

//cria e inicia thread para o consumidor
new Thread(new Consumidor(queue)).start();
}
}

import java.util.concurrent.BlockingQueue;

public class Produtor implements Runnable {

private BlockingQueue<Integer> queue;

public Produtor(BlockingQueue<Integer> queue) {
this.queue = queue;
}

public void run() {
int i=0;

try {
while(true) {
/*adiciona valor na fila imediatamente se houver espaço,
caso contrario aguarda até que o espaço fique disponível*/
queue.put(i++);

Thread.sleep(500);
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}

import java.util.concurrent.BlockingQueue;

public class Consumidor implements Runnable {

private BlockingQueue<Integer> queue;

public Consumidor(BlockingQueue<Integer> queue) {
this.queue = queue;
}

public void run() {
try {
while(true) {
/*se houver valor retira ele da fila, senão fica bloqueado
até que um valor seja adicionado */
int valor = queue.take();

System.out.println(valor);
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}

Conforme pode ser visto nos comentários no código, o método put() adiciona valor na fila imediatamente se houver espaço, caso contrario aguarda até que o espaço fique disponível. No consumidor, o método take() é chamado e, se houver algum valor disponível ele é retirado, senão fica bloqueado até que um valor seja adicionado.

Na próxima parte deste artigo irei apresentar algumas outras ferramentas deste pacote, como o uso de locks e semáforos.