2

Durante um projeto me sugeriram o uso de Fork in Join da API do Java ao invés de threads, não achei nada fácil de entender pelos exemplos achados no Google.

Eu entendi que é possível passar uma lista de tarefas e ele subdivide e depois reúne os resultados.

Gostaria de saber se é melhor que usar threads e se alguém tem um exemplo para mostrar ou até mesmo contar sua experiencia de uso com esse modelo.

3 Respostas 3

4

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 Callables. 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.

3
  • 1
    Pois é, meu problema na verdade já implementei threads para manipular e-mail da API JavaMail, pois eu realizo procedimento de separar mensagens por remetente, separar mensagens por conteúdo, criar pastas. Coisas deste tipo, mas pretendo usar concorrência para outras funcionalidades no sistema. Commented 6/02/2014 às 13:27
  • @Macario1983 Corrigido!
    – utluiz
    Commented 6/02/2014 às 13:59
  • eu usei um código com forkInJoin, cara muito show de bola, problema é como você colocou, ele não permite sincronização, algo que tive que alterar em meu código pois eu atribuía a um vetor quais mensagens não tem anexo. Então optei por passar um list para facilitar. Vou postar o código também para caso alguém queira ver, ou alguém possa corrigir. Commented 6/02/2014 às 16:11
2

Existe um projeto chamado Java Concurrent Animated que criou uma aplicação de demonstração das capacidades de Java em programação concorrente / paralela.

Você pode baixar um demo (com os fontes) que você pode usar para estudar tudo relacionado à esse tema.

Existem vários tipos de animações que demonstram como usar cada modelo de programação concorrente. Durante a animação, trechos do código de exemplo vão sendo destacados, dando uma boa ideia de como implementar.

O demo é apenas um JAR autoexecutável. É só dar duplo clique.

É muito legal e muito bem feito!

1
  • Muito boa sua recomendação! Commented 6/02/2014 às 13:23
0

Estou postando o código como exemplo para quem quiser ver como foi feito. Este pode ser alterado por quem quiser, desde que tenha melhorado.

package service.forkinjoin;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.RecursiveAction;

import javax.mail.Message;
import javax.mail.MessagingException;
import javax.mail.Multipart;
import javax.mail.Part;
import javax.mail.internet.MimeBodyPart;

import service.FileUtil;

public class ForkSortMessages extends RecursiveAction {

    private static final long serialVersionUID = -1092415796824205832L;
    private List<Message> listMessages;
    private List<Message> listMessagesToDelete;

    public ForkSortMessages(List<Message> listMessages, List<Message> listMessagesToDelete) {
        this.listMessages = listMessages;
        this.listMessagesToDelete = listMessagesToDelete;
    }

    @Override
    protected void compute() {

        List<RecursiveAction> actions = new ArrayList<>();

        if (this.listMessages.size() <= Runtime.getRuntime().availableProcessors()) {
            try {
                this.separateMessages();
            } catch (MessagingException | IOException e) {
                e.printStackTrace();
            }
        } else {

            int end = this.listMessages.size() / 2;
            actions.add(new ForkSortMessages(this.listMessages.subList(0, end), this.listMessagesToDelete));
            end += this.listMessages.size() % 2 == 0 ? 0 : 1;
            actions.add(new ForkSortMessages(this.listMessages.subList(end, this.listMessages.size()), this.listMessagesToDelete));
            invokeAll(actions);
        }
    }

    private void separateMessages() throws MessagingException, IOException {

        for (Message message : this.listMessages) {

            if ((this.hasNoAttachment(message.getContentType()) || (this.hasNoXmlAttachment(message)))) {
                listMessagesToDelete.add(message);
            }
        }
    }

    private boolean hasNoAttachment(String content) {
        return !content.contains("multipart/MIXED");
    }

    private boolean hasNoXmlAttachment(Message message) throws IOException, MessagingException {

        Multipart multipart = (Multipart) message.getContent();

        for (int i = 0; i < multipart.getCount(); i++) {

            MimeBodyPart mimeBodyPart = (MimeBodyPart) multipart.getBodyPart(i);

            if (Part.ATTACHMENT.equalsIgnoreCase(mimeBodyPart.getDisposition())) {

                if (FileUtil.isXmlFile(mimeBodyPart.getFileName())) {
                    return false;
                }
            }
        }

        return true;
    }
}
2
  • Esta é uma resposta como ficou em definitivo ou faz parte da sua pergunta?
    – Maniero
    Commented 8/02/2014 às 14:25
  • 1
    @bigown foi um código que fiz, não sei esta 100% como é a ideia, mas fiz baseado no que vi em alguns exemplos. Estou deixando para que possa ajudar. Commented 8/02/2014 às 16:17

Você deve fazer log-in para responder a esta pergunta.

Esta não é a resposta que você está procurando? Pesquise outras perguntas com a tag .