Только начал изучать многопоточность. У меня 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);
    }



   }
 }