After the feedback I have got from this question, I tried to to re-implement the classic Producer Consumer problem using multithreading.
The requirements are:
- 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.
It's running and seems to be giving correct output. However, I have a small concern. In my code, if a producer is producing an item, consumer won't be able to consume simultaneously. Is there a way to break this dependency? i.e. A consumer should be able to consume item (provided the queue is not empty) irrespective of whether the producer is producing item or not and vice versa.
Custom Blocking Queue
public class CustomQueue<E> {
private Queue<E> queue;
private int limit;
public CustomQueue(int limit){
this.queue = new LinkedList<>();
this.limit = limit;
}
public synchronized void enQueue(E e) throws InterruptedException{
while(queue.size()==limit){
System.out.println("Queue is full."+ Thread.currentThread().getName() + " is waiting....");
wait();
}
System.out.println("Queue is not full. "+ Thread.currentThread().getName() + " is producing "+ e);
queue.add(e);
notifyAll();
}
public synchronized E deQueue() throws InterruptedException{
while(queue.size()==0){
System.out.println("Queue is empty."+ Thread.currentThread().getName() + " is waiting....");
wait();
}
E e = queue.remove();
System.out.println("Queue is not empty. "+ Thread.currentThread().getName() + " is consuming "+ e);
notifyAll();
return e;
}
}
Producer Code
public class Producer implements Runnable{
private CustomQueue<Integer> sharedQueue;
public Producer(CustomQueue<Integer> sharedQueue){
this.sharedQueue = sharedQueue;
}
@Override
public void run() {
while(true){
produceItems();
}
}
private void produceItems() {
for(int i=0; i<10; i++){
try {
sharedQueue.enQueue(i);
} catch (InterruptedException e) {
System.out.println("Exception in producer."+ e);
}
}
}
}
Consumer Code
public class Consumer implements Runnable {
private CustomQueue<Integer> sharedQueue;
public Consumer(CustomQueue<Integer> sharedQueue){
this.sharedQueue = sharedQueue;
}
@Override
public void run() {
while(true){
int item = consumeItem();
}
}
private int consumeItem() {
int item;
try {
item = sharedQueue.deQueue();
} catch (InterruptedException e) {
item = Integer.MIN_VALUE;
System.out.println("Exception in consumer."+ e);
}
return item;
}
}
Test Code
public class Test {
public static void main(String args[]){
CustomQueue<Integer> sharedQueue = new CustomQueue<>(10);
Producer p1 = new Producer(sharedQueue);
Producer p2 = new Producer(sharedQueue);
Consumer c1 = new Consumer(sharedQueue);
Consumer c2 = new Consumer(sharedQueue);
Thread pt1 = new Thread(p1);
Thread ct1 = new Thread(c1);
Thread pt2 = new Thread(p2);
Thread ct2 = new Thread(c2);
pt1.setName("Producer 1");
pt2.setName("Producer 2");
ct1.setName("Consumer 1");
ct2.setName("Consumer 2");
ct1.start();
pt2.start();
ct2.start();
pt1.start();
}
}
Output snippet
- Queue is not empty. Consumer 1 is consuming 5
- Queue is not empty. Consumer 1 is consuming 6
- Queue is not empty. Consumer 1 is consuming 7
- Queue is not empty. Consumer 1 is consuming 8
- Queue is empty.Consumer 1 is waiting....
- Queue is empty.Consumer 2 is waiting....
- Queue is not full. Producer 2 is producing 6
- Queue is not full. Producer 2 is producing 7
main()
as well? \$\endgroup\$