Implementing multiple Producer Consumer using custom Blocking queue I am trying to implement multiple producer multiple consumer problem using a custom blocking queue. The requirement is that:


If the queue is not full, all the producers should be able to produce products independently.
If the queue is full, the producers have to wait.
If the queue is not empty, the consumer can consume products independently.
If the queue is empty, the consumers have to wait.


Here is the code I have implemented. Could you please review my code and let me know if there is any flaw?

Custom Blocking Queue

public class CustomQueue&lt;T&gt; {

private static final Object LOCK = new Object();
private int queueSize = 100; // default size
private List&lt;T&gt; queue;

public CustomQueue(int size){
    this.queueSize = size;
    this.queue = new LinkedList&lt;&gt;();
}

public void produce(T t){
    if(queue.size()==queueSize){ // queue is full. Producer need to wait
        synchronized(LOCK){
            while(queue.size()==queueSize){
                try {
                    LOCK.wait();
                } catch (InterruptedException e) {
                    System.out.println("Exception in produce() while waiting. Reason: "+e.getCause());
                }
            }
        }
    }else{ // queue is not full. Producer can produce items
        queue.add(t);
        LOCK.notifyAll();
    }
}

public T consume(){
    if(queue.size()==0){
        while(queue.size()==0){// queue is empty. Consumer need to wait
            try {
                LOCK.wait();
            } catch (InterruptedException e) {
                System.out.println("Exception in consume() while waiting. Reason: "+e.getCause());
            }
        }
    }else{ // queue is not empty. Consumers can consume item
        T t = queue.get(0);
        return t;
    }
    return null;
}


Producer code

public class Producer&lt;T&gt; implements Runnable{

private CustomQueue&lt;T&gt; sharedQueue;

public Producer(CustomQueue&lt;T&gt; sharedQueue){
    this.sharedQueue = sharedQueue;
}

@Override
public void run() {
    produceItems();     
}

private void produceItems() {
    List&lt;T&gt; items = getItems();     
    for(T t: items){
        sharedQueue.produce(t);
    }       
}

private List&lt;T&gt; getItems() {
    //dummy method to get list of items
    return null;
}


Consumer code

public class Consumer&lt;T&gt; implements Runnable {

private CustomQueue&lt;T&gt; sharedQueue;

public Consumer(CustomQueue&lt;T&gt; sharedQueue){
    this.sharedQueue = sharedQueue;
}

@Override
public void run() {
    T t = consumeItem();        
}

private T consumeItem() {
    T t = sharedQueue.consume();
    return t;
}
Your produce and consume methods are both buggy for the same reason: you've got a time of check/time of use bug.

Imagine a queue with size 1. Start two producer threads at the same time. They'll both see queue.size() == 0 in produce() and they'll both happily proceed to queue.add() without a lock, overflowing your queue (and causing other problems if that add() method isn't thread-safe itself).

Same thing exactly on the consumer side. If there's one item in the queue and two threads enter consume() at the same time, both will try to extract an item and at least one will fail.

You need to hold a lock that covers the check and the produce/consume actions.



You can't call notifyAll() on an object unless you own the object's monitor. You'll get a IllegalMonitorStateException if you try to run your code.