Aguarde e notifique em Threads de Consumidor e Produtor
Apenas comecei a aprender multi-threading. Eu tenho 5 produtores e 2 consumidores em vários segmentos. Basicamente, este programa adiciona 100 itens à fila. O produtor deixará de adicionar quando o tamanho da fila for 100. Gostaria que o consumidor notificasse o produtor quando remover todos os itens da fila para que o produtor possa começar a adicionar novamente. Atualmente, o produtor espera, mas nunca é notificado pelo consumidor.
Produtor:
public class Producer implements Runnable {
private BlockingQueue sharedQueue;
private final int queueSize;
private Object lock = new Object();
public Producer(BlockingQueue sharedQueue, int queueSize){
this.sharedQueue = sharedQueue;
this.queueSize = queueSize;
}
public void run() {
while(true) {
if(sharedQueue.size()== queueSize){
try {
synchronized (lock) {
sharedQueue.wait();
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
try {
sharedQueue.put("Producer: " + sharedQueue.size());
Thread.sleep(500);
System.out.println("Producer: Queue Size " + sharedQueue.size() + " Current Thread " + Thread.currentThread());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
consumidor:
public class Consumer implements Runnable{
private BlockingQueue sharedQueue;
private final int queueSize;
private final int queueEmpty=0;
private Object lock = new Object();
public Consumer(BlockingQueue sharedQueue, int queueSize){
this.sharedQueue = sharedQueue;
this.queueSize = queueSize;
}
//Notify awaiting thread if the sharedQueue is empty
public void run() {
while (true) {
if(sharedQueue.size()==queueEmpty){
synchronized (lock) {
this.notifyAll();
}
}
try {
sharedQueue.take();
Thread.sleep(800);
System.out.println("Consumer: Queue Size " + sharedQueue.size() + " Current Thread" + Thread.currentThread());
}catch(InterruptedException e){
e.printStackTrace();
}
}
}
}
Classe principal
public class App{
//A simple program to illustrate how producer and consumer pattern works with blocking queue using executor service
public static void main( String[] args )
{
final BlockingQueue<String> sharedQueue = new ArrayBlockingQueue<String> (100);
final int queueSize =100;
final int producerNum = 5;
final int consumerNum = 2;
final ExecutorService executorProducer = Executors.newFixedThreadPool(producerNum);
final ExecutorService executorConsumer = Executors.newFixedThreadPool(consumerNum);
for(int i=0;i<producerNum;i++){
Producer producer = new Producer(sharedQueue,queueSize);
executorProducer.execute(producer);
}
for(int j=0;j<consumerNum;j++){
Consumer consumer = new Consumer(sharedQueue,queueSize);
executorConsumer.execute(consumer);
}
}
}