Ждите и уведомляйте в Потоке Потребителя и Производителя
Только начал изучать многопоточность. У меня 5 производителей и 2 потребителя в нескольких потоках. В основном эта программа добавляет 100 элементов в очередь. Производитель прекращает добавление, когда размер очереди равен 100. Я бы хотел, чтобы потребитель уведомил производителя, когда потребитель удалит все элементы из очереди, чтобы производитель мог начать добавление снова. В настоящее время производитель будет ждать, но никогда не получит уведомление от потребителя.
Режиссер:
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();
}
}
}
}
потребитель:
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();
}
}
}
}
Основной класс
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);
}
}
}