Encadeamentos de Produtor / Consumidor usando uma Fila

Gostaria de criar algum tipo de aplicativo de segmentação Producer/Consumer . Mas não tenho certeza qual é a melhor maneira de implementar uma fila entre os dois.

Então, eu tenho algumas com duas ideias (ambas podem estar totalmente erradas). Eu gostaria de saber qual seria melhor e se os dois sugam, qual seria a melhor maneira de implementar a fila. É principalmente minha implementação da fila nesses exemplos que me preocupa. Estou estendendo uma class Queue que é uma class interna e segura para threads. Abaixo estão dois exemplos com 4 classs cada.

Main class-

 public class SomeApp { private Consumer consumer; private Producer producer; public static void main (String args[]) { consumer = new Consumer(); producer = new Producer(); } } 

Classe de consumidor

 public class Consumer implements Runnable { public Consumer() { Thread consumer = new Thread(this); consumer.start(); } public void run() { while(true) { //get an object off the queue Object object = QueueHandler.dequeue(); //do some stuff with the object } } } 

Classe de produtor

 public class Producer implements Runnable { public Producer() { Thread producer = new Thread(this); producer.start(); } public void run() { while(true) { //add to the queue some sort of unique object QueueHandler.enqueue(new Object()); } } } 

Fila de class

 public class QueueHandler { //This Queue class is a thread safe (written in house) class public static Queue readQ = new Queue(100); public static void enqueue(Object object) { //do some stuff readQ.add(object); } public static Object dequeue() { //do some stuff return readQ.get(); } } 

OU

Main class-

 public class SomeApp { Queue readQ; private Consumer consumer; private Producer producer; public static void main (String args[]) { readQ = new Queue(100); consumer = new Consumer(readQ); producer = new Producer(readQ); } } 

Classe de consumidor

 public class Consumer implements Runnable { Queue queue; public Consumer(Queue readQ) { queue = readQ; Thread consumer = new Thread(this); consumer.start(); } public void run() { while(true) { //get an object off the queue Object object = queue.dequeue(); //do some stuff with the object } } } 

Classe de produtor

 public class Producer implements Runnable { Queue queue; public Producer(Queue readQ) { queue = readQ; Thread producer = new Thread(this); producer.start(); } public void run() { while(true) { //add to the queue some sort of unique object queue.enqueue(new Object()); } } } 

Fila de class

 //the extended Queue class is a thread safe (written in house) class public class QueueHandler extends Queue { public QueueHandler(int size) { super(size); //All I'm thinking about now is McDonalds. } public void enqueue(Object object) { //do some stuff readQ.add(); } public Object dequeue() { //do some stuff return readQ.get(); } } 

E ir!

O Java 5+ tem todas as ferramentas que você precisa para esse tipo de coisa. Você vai querer:

  1. Coloque todos os seus produtores em um ExecutorService ;
  2. Coloque todos os seus Consumidores em outro ExecutorService ;
  3. Se necessário, comunique-se entre os dois usando um BlockingQueue .

Eu digo “se necessário” para (3) porque da minha experiência é um passo desnecessário. Tudo o que você faz é enviar novas tarefas para o serviço executor do consumidor. Assim:

 final ExecutorService producers = Executors.newFixedThreadPool(100); final ExecutorService consumers = Executors.newFixedThreadPool(100); while (/* has more work */) { producers.submit(...); } producers.shutdown(); producers.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS); consumers.shutdown(); consumers.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS); 

Então os producers submetem diretamente aos consumers .

OK, como os outros observam, a melhor coisa a fazer é usar o pacote java.util.concurrent . Eu recomendo altamente “Java Concurrency in Practice”. É um ótimo livro que cobre quase tudo que você precisa saber.

Quanto à sua implementação em particular, como observei nos comentários, não inicie Threads from Constructors – pode ser inseguro.

Deixando isso de lado, a segunda implementação parece melhor. Você não quer colocar filas em campos estáticos. Você provavelmente está perdendo a flexibilidade para nada.

Se você quiser prosseguir com sua própria implementação (para fins de aprendizado, eu acho?), Forneça um método start() pelo menos. Você deve construir o object (você pode instanciar o object Thread ) e, em seguida, chamar start() para iniciar o thread.

Editar: ExecutorService tem sua própria fila para que isso possa ser confuso .. Aqui está algo para você começar.

 public class Main { public static void main(String[] args) { //The numbers are just silly tune parameters. Refer to the API. //The important thing is, we are passing a bounded queue. ExecutorService consumer = new ThreadPoolExecutor(1,4,30,TimeUnit.SECONDS,new LinkedBlockingQueue(100)); //No need to bound the queue for this executor. //Use utility method instead of the complicated Constructor. ExecutorService producer = Executors.newSingleThreadExecutor(); Runnable produce = new Produce(consumer); producer.submit(produce); } } class Produce implements Runnable { private final ExecutorService consumer; public Produce(ExecutorService consumer) { this.consumer = consumer; } @Override public void run() { Pancake cake = Pan.cook(); Runnable consume = new Consume(cake); consumer.submit(consume); } } class Consume implements Runnable { private final Pancake cake; public Consume(Pancake cake){ this.cake = cake; } @Override public void run() { cake.eat(); } } 

EDIT adicional: Para o produtor, em vez de while(true) , você pode fazer algo como:

 @Override public void run(){ while(!Thread.currentThread().isInterrupted()){ //do stuff } } 

Desta forma, você pode desligar o executor chamando .shutdownNow() . Se você usar while(true) , ele não será encerrado.

Observe também que o Producer ainda está vulnerável a RuntimeExceptions (ou seja, uma RuntimeException interromperá o processamento)

Você está reinventando a roda.

Se você precisar de persistência e outros resources corporativos, use o JMS (sugiro o ActiveMq ).

Se você precisar de filas na memory rápidas, use uma das impalações da fila do java.

Se você precisar dar suporte ao java 1.4 ou anterior, use o excelente pacote simultâneo da Doug Lea.

Eu estendi a resposta proposta cletus para o exemplo de código de trabalho.

  1. Um ExecutorService (pes) aceita tarefas do Producer .
  2. Um ExecutorService (ces) aceita tarefas do Consumer .
  3. Tanto o Producer quanto o Consumer compartilham o BlockingQueue .
  4. Tarefas de múltiplos Producer geram números diferentes.
  5. Qualquer uma das tarefas do Consumer pode consumir o número gerado pelo Producer

Código:

 import java.util.concurrent.*; public class ProducerConsumerWithES { public static void main(String args[]){ BlockingQueue sharedQueue = new LinkedBlockingQueue(); ExecutorService pes = Executors.newFixedThreadPool(2); ExecutorService ces = Executors.newFixedThreadPool(2); pes.submit(new Producer(sharedQueue,1)); pes.submit(new Producer(sharedQueue,2)); ces.submit(new Consumer(sharedQueue,1)); ces.submit(new Consumer(sharedQueue,2)); // shutdown should happen somewhere along with awaitTermination / * https://stackoverflow.com/questions/36644043/how-to-properly-shutdown-java-executorservice/36644320#36644320 */ pes.shutdown(); ces.shutdown(); } } class Producer implements Runnable { private final BlockingQueue sharedQueue; private int threadNo; public Producer(BlockingQueue sharedQueue,int threadNo) { this.threadNo = threadNo; this.sharedQueue = sharedQueue; } @Override public void run() { for(int i=1; i<= 5; i++){ try { int number = i+(10*threadNo); System.out.println("Produced:" + number + ":by thread:"+ threadNo); sharedQueue.put(number); } catch (Exception err) { err.printStackTrace(); } } } } class Consumer implements Runnable{ private final BlockingQueue sharedQueue; private int threadNo; public Consumer (BlockingQueue sharedQueue,int threadNo) { this.sharedQueue = sharedQueue; this.threadNo = threadNo; } @Override public void run() { while(true){ try { int num = sharedQueue.take(); System.out.println("Consumed: "+ num + ":by thread:"+threadNo); } catch (Exception err) { err.printStackTrace(); } } } } 

saída:

 Produced:11:by thread:1 Produced:21:by thread:2 Produced:22:by thread:2 Consumed: 11:by thread:1 Produced:12:by thread:1 Consumed: 22:by thread:1 Consumed: 21:by thread:2 Produced:23:by thread:2 Consumed: 12:by thread:1 Produced:13:by thread:1 Consumed: 23:by thread:2 Produced:24:by thread:2 Consumed: 13:by thread:1 Produced:14:by thread:1 Consumed: 24:by thread:2 Produced:25:by thread:2 Consumed: 14:by thread:1 Produced:15:by thread:1 Consumed: 25:by thread:2 Consumed: 15:by thread:1 

Nota. Se você não precisar de vários Produtores e Consumidores, mantenha um único Produtor e Consumidor. Eu adicionei vários produtores e consumidores para mostrar os resources do BlockingQueue entre vários produtores e consumidores.

Este é um código muito simples.

 import java.util.*; // @author : rootTraveller, June 2017 class ProducerConsumer { public static void main(String[] args) throws Exception { Queue queue = new LinkedList<>(); Integer buffer = new Integer(10); //Important buffer or queue size, change as per need. Producer producerThread = new Producer(queue, buffer, "PRODUCER"); Consumer consumerThread = new Consumer(queue, buffer, "CONSUMER"); producerThread.start(); consumerThread.start(); } } class Producer extends Thread { private Queue queue; private int queueSize ; public Producer (Queue queueIn, int queueSizeIn, String ThreadName){ super(ThreadName); this.queue = queueIn; this.queueSize = queueSizeIn; } public void run() { while(true){ synchronized (queue) { while(queue.size() == queueSize){ System.out.println(Thread.currentThread().getName() + " FULL : waiting...\n"); try{ queue.wait(); //Important } catch (Exception ex) { ex.printStackTrace(); } } //queue empty then produce one, add and notify int randomInt = new Random().nextInt(); System.out.println(Thread.currentThread().getName() + " producing... : " + randomInt); queue.add(randomInt); queue.notifyAll(); //Important } //synchronized ends here : NOTE } } } class Consumer extends Thread { private Queue queue; private int queueSize; public Consumer(Queue queueIn, int queueSizeIn, String ThreadName){ super (ThreadName); this.queue = queueIn; this.queueSize = queueSizeIn; } public void run() { while(true){ synchronized (queue) { while(queue.isEmpty()){ System.out.println(Thread.currentThread().getName() + " Empty : waiting...\n"); try { queue.wait(); //Important } catch (Exception ex) { ex.printStackTrace(); } } //queue not empty then consume one and notify System.out.println(Thread.currentThread().getName() + " consuming... : " + queue.remove()); queue.notifyAll(); } //synchronized ends here : NOTE } } } 
  1. Código Java “BlockingQueue” que sincronizou o método put e get.
  2. Código Java “Producer”, encadeamento produtor para produzir dados.
  3. Código Java “Consumidor”, segmento consumidor para consumir os dados produzidos.
  4. Código Java “ProducerConsumer_Main”, function principal para iniciar o encadeamento produtor e consumidor.

BlockingQueue.java

 public class BlockingQueue { int item; boolean available = false; public synchronized void put(int value) { while (available == true) { try { wait(); } catch (InterruptedException e) { } } item = value; available = true; notifyAll(); } public synchronized int get() { while(available == false) { try { wait(); } catch(InterruptedException e){ } } available = false; notifyAll(); return item; } } 

Consumer.java

 package com.sukanya.producer_Consumer; public class Consumer extends Thread { blockingQueue queue; private int number; Consumer(BlockingQueue queue,int number) { this.queue = queue; this.number = number; } public void run() { int value = 0; for (int i = 0; i < 10; i++) { value = queue.get(); System.out.println("Consumer #" + this.number+ " got: " + value); } } } 

ProducerConsumer_Main.java

 package com.sukanya.producer_Consumer; public class ProducerConsumer_Main { public static void main(String args[]) { BlockingQueue queue = new BlockingQueue(); Producer producer1 = new Producer(queue,1); Consumer consumer1 = new Consumer(queue,1); producer1.start(); consumer1.start(); } }