Pool de threads acessando fila - Help!

Programação Java
#1

Prezados, estou precisando de algo do tipo:

[code] class Produtor implements Runnable {
//* Adiciona objetos (strings) à uma fila do tipo FIFO
}

class Consumidor extends Socket implements Runnable {
//* Pega o objeto da fila, processa (acessa socket, coleta informação e armazena resultado em um List).
}[/code] Quanto à sockets, tudo ok, o projeto já está rodando. Mas quero dar agilidade na execução da tarefa, preciso de pelo menos 4 linhas de execução simultânea. Tenho pesquisado no google e guj, entretanto como não tenho experiência no uso de threads e filas, gostaria que alguem me indicasse algum link específico sobre o assunto, ou mesmo algum exemplo sobre como implementar essa funcionalidade.

Grato.

#2

Mas sua dúvida diz respeito a como montar a fila FIFO?

Dá uma olhada aqui (Java 5)

#3

Olá

De cara vou avisando que se não tem muito conhecimento de threads então já tem programa para o fim de semana.

Você pode implantar um pool de threads com a classe ThreadPoolExecutor.

As 2 coisas importantes a ter em mente ao usar esta classe é o tamanho do pool e a fila usada para as tasks que podem ser passados no construtor.

A fila deve ser uma BlockingQueue e a escolha depende de:
a) SynchronousQueue = Uma fila de tamanho ZERO, isto é, o que entra é pego logo em seguida. Funciona apenas como um ponto de encontro entre Threads

b) Uma fila ilimitada tal como LinkedBlockingQueue com capacidade ilimitada. Adicionar uma task na fila sempre tem sucesso e nenhuma task é rejeitada.

c) Uma fila limitada tal como LinkedBlockingQueue com capacidade limitada ou um ArrayBlockingQueue

Leia com calma o javadoc da classe ThreadPoolExecutor e também das interfaces que implementa. Há exemplos que te interessam. E google por mais explicações sobre estas classes cujo uso é difícil de explicar em poucas palavras.

[]s
Luca

#4

Sim, e várias threads acessando-a sincronizadamente.

#5

Nada como o [COMPLETE ANSWER MODE ON] do Luca… hehehe…

Bem legal! :wink:

#6

Olá

Bem, para ficar mais completa eu deveria ter sugerido a ele o uso do java.nio como alternativa ao modelo dele. Com um ServerSocketChannel ele resolve o problema mais rapidamente e pode passar o fim de semana curtindo o Redibul-er-reice.

[]s
Luca

#7

Obrigado pelas sugestões. Vou iniciar a caminhada…

Valeu.

#8

Além do ThreadPoolExecutors, não custa dar uma olhada em uma fábrica bem prático para ele, a classe Executors.

Basicamente, você faz:

//Cria um pool para 4 linhas de execução simultânetas. ExecutorService threadPool = Executors.newFixedThreadPool(4);

O resto, é como o Luca falou… E esta introdução do Brian Goetz pode te ajudar.

#9

//Cria um pool para 4 linhas de execução simultânetas.
ExecutorService threadPool = Executors.newFixedThreadPool(4);[/code][/quote]De acordo com as dicas, problema resolvido. Para constar no tópico:

class Teste { public static void main(String[] args) { BlockingQueue q = new LinkedBlockingQueue(); ExecutorService threadPool = Executors.newCachedThreadPool(); threadPool.execute(new Produtor(q)); threadPool.execute(new Consumidor(q)); threadPool.execute(new Consumidor(q)); threadPool.shutdown(); } } class Produtor implements Runnable{ private final BlockingQueue queue; Produtor(BlockingQueue q) {queue = q;} public void run() { try { for (int i = 0; i &lt= 100; i++) { System.out.println("Adicionando tarefa " + i); queue.put("tarefa " + i);} } catch (InterruptedException ex) { ex.printStackTrace();} } } class Consumidor implements Runnable{ private final BlockingQueue queue; Consumidor(BlockingQueue q) {queue = q;} public void run() { while(queue.isEmpty() == false) { try { System.out.println("Executando " + queue.take()); Thread.sleep(500); // Simula delay real do consumidor } catch (InterruptedException ex) { ex.printStackTrace();}} } }

#10

Ficou legal tua implementação. :slight_smile:

Se quiser conhecer um pouco mais do java.util.concurrent tente reformular o seu problema para executar com um CompletionService.

Dá uma olhada no artigo do Goetz que eu te passei pra ver como fazer isso. :wink:

#11

Pessoal,

Eu usei a implementação acima funcionou perfeitamente, mas estou com um problema.

A versão do java é 1.4 e não posso usar o ExecutorService.

Alguém teria uma implementação de um PoolThread compatível com esta versão para me ajudar?

Encontrei alguns exemplos no site da IBM, mas o comportamento na execução foi muito estranho. A cada execução eu tenho um comportamento diferente e a thread nunca morre no fim das execuções.

[]'s

#12

pessoal, eu peço desculpa pro ter revivido o tópico, se bem que o assunto é ± o mesmo.

eu quero criar um pool de threads para fazer determinado trabalho, então aproveitei o exemplo do genildof, porém usando FixedThreadPool, ja que quero um tamanho fixo para o pool de threads. O problema é que as vezes alguma thread não é finalizada, enquanto que outras vezes executa certinho, a quantidade de threads do pool que executa por ciclo, e depois fecha todas, se alguma só pode entrar para executar depois que a fila estava vazia ela simplesmente termina, o ruim é quando essa ultima thread não termina, o que é o mais comum…

alguém saberia me dizer como corrigir isso?

segue o código do nosso amigo ai abaixo, depois que adaptei:


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

public class testeThread {
    public static void main(String[] args) {
        BlockingQueue q = new LinkedBlockingQueue();
        ExecutorService threadPool = Executors.newFixedThreadPool(3);
        //ExecutorService threadPool = Executors.newCachedThreadPool();
        
        Produtor p = new Produtor(q);
        
        threadPool.execute(new Produtor(q));
        threadPool.execute(new Consumidor(q, "primeira"));
        threadPool.execute(new Consumidor(q, "segunda "));
        threadPool.execute(new Consumidor(q, "terceira"));
        threadPool.execute(new Consumidor(q, "quarta  "));
        threadPool.execute(new Consumidor(q, "quinta  "));
        threadPool.shutdown();
    }
}
class Produtor implements Runnable{
    private final BlockingQueue queue;
    Produtor(BlockingQueue q) {queue = q;}
    public void run() {
    	Thread.currentThread().setPriority(10);
        try { 
            for (int i = 0; i <= 30; i++) {
                System.out.println("Adicionando tarefa " + i);
                queue.put("tarefa " + i);}
        } catch (InterruptedException ex) {
            ex.printStackTrace();}
    }
}
class Consumidor implements Runnable{
    private final BlockingQueue queue;
    private String nome;
    
    public String getNome(){ return nome; }
    
    Consumidor(BlockingQueue q, String nome) {this.queue = q; this.nome = nome;}
    public void run() {
    	Thread.currentThread().setPriority(2);
        while(queue.isEmpty() == false) {
            try {
                System.out.println("Executando " + nome + " - " + queue.take()); 
                Thread.sleep(500); // Simula delay real do consumidor
            } catch (InterruptedException ex) {
                ex.printStackTrace();}
        }System.out.println("terminando thread "+nome);
    }
}

um exemplo de saida onde todas as threads terminam certinho:

Adicionando tarefa 0 Adicionando tarefa 1 Adicionando tarefa 2 Adicionando tarefa 3 Adicionando tarefa 4 Adicionando tarefa 5 Adicionando tarefa 6 Adicionando tarefa 7 Adicionando tarefa 8 Adicionando tarefa 9 Adicionando tarefa 10 Adicionando tarefa 11 Adicionando tarefa 12 Adicionando tarefa 13 Adicionando tarefa 14 Adicionando tarefa 15 Executando terceira - tarefa 0 Executando segunda - tarefa 1 Executando primeira - tarefa 2 Executando segunda - tarefa 3 Executando primeira - tarefa 5 Executando terceira - tarefa 4 Executando primeira - tarefa 6 Executando terceira - tarefa 8 Executando segunda - tarefa 7 Executando primeira - tarefa 9 Executando segunda - tarefa 11 Executando terceira - tarefa 10 Executando primeira - tarefa 12 Executando terceira - tarefa 14 Executando segunda - tarefa 13 Executando primeira - tarefa 15 terminando thread terceira terminando thread quarta terminando thread quinta terminando thread segunda terminando thread primeira

agora um exemplo onde alguma não termina, eu acabo precisando fecha pelo botão do proprio eclipse:

Adicionando tarefa 0 Adicionando tarefa 1 Adicionando tarefa 2 Adicionando tarefa 3 Adicionando tarefa 4 Adicionando tarefa 5 Adicionando tarefa 6 Adicionando tarefa 7 Adicionando tarefa 8 Adicionando tarefa 9 Adicionando tarefa 10 Adicionando tarefa 11 Adicionando tarefa 12 Adicionando tarefa 13 Adicionando tarefa 14 Adicionando tarefa 15 Executando terceira - tarefa 0 Executando primeira - tarefa 1 Executando segunda - tarefa 2 Executando terceira - tarefa 3 Executando primeira - tarefa 4 Executando segunda - tarefa 5 Executando terceira - tarefa 6 Executando primeira - tarefa 7 Executando segunda - tarefa 8 Executando terceira - tarefa 9 Executando primeira - tarefa 10 Executando segunda - tarefa 11 Executando primeira - tarefa 12 Executando terceira - tarefa 13 Executando segunda - tarefa 14 Executando terceira - tarefa 15 terminando thread segunda terminando thread quarta terminando thread quinta terminando thread terceira

nessa execução ai, foi a primeira que não terminou, mais nem sempre é ela, tiveram outras vezes também que quem não termina é a segunda por exemplo, conforme abaixo:

Adicionando tarefa 0 Adicionando tarefa 1 Adicionando tarefa 2 Adicionando tarefa 3 Adicionando tarefa 4 Adicionando tarefa 5 Adicionando tarefa 6 Adicionando tarefa 7 Adicionando tarefa 8 Adicionando tarefa 9 Adicionando tarefa 10 Adicionando tarefa 11 Adicionando tarefa 12 Adicionando tarefa 13 Adicionando tarefa 14 Adicionando tarefa 15 Executando terceira - tarefa 0 Executando segunda - tarefa 1 Executando primeira - tarefa 2 Executando terceira - tarefa 3 Executando primeira - tarefa 5 Executando segunda - tarefa 4 Executando primeira - tarefa 6 Executando segunda - tarefa 8 Executando terceira - tarefa 7 Executando segunda - tarefa 9 Executando terceira - tarefa 11 Executando primeira - tarefa 10 Executando segunda - tarefa 12 Executando primeira - tarefa 14 Executando terceira - tarefa 13 Executando primeira - tarefa 15 terminando thread terceira terminando thread quarta terminando thread quinta terminando thread primeira

#13