Java Multithreading: Slow Producer and Fast Consumer

Question

I'm trying to understand synchronized blocks. Here I have implemented a single producer thread and 2 consumer threads.

I keep getting exception in thread since LinkedList is empty.

package com.Main;

import com.qed.Consumer;
import com.qed.Producer;
import com.qed.Store;

public class Main {

public static void main(String[] args) throws InterruptedException {
    Store st = new Store();

    Thread populate = new Thread(new Producer(st)); 
    Thread con1 = new Thread(new Consumer(st));
    Thread con2 = new Thread(new Consumer(st));
    con1.setName("A");
    con2.setName("B");
    populate.start();
    con1.start();
    con2.start();

    populate.join();
    con1.join();
    con2.join();
    if(populate.isAlive()){
        con1.interrupt();
        con2.interrupt();
    }
  }
}

package com.qed;

import java.util.LinkedList;

public class Store {

private LinkedList<Integer> qu = new LinkedList<Integer>();
private final Object lock = new Object();

public void add(int data){
    try{
        while(qu.size() ==10){
            Thread.sleep(1);
            }
        qu.add(data);
    }catch(InterruptedException ie){
        ie.printStackTrace();
    }
}

public int remove(){
    int data=0;
    try{
        synchronized(lock){
            while(qu.size() == 0){
                Thread.sleep(1);
                }
            data = qu.removeFirst();
         }
    }catch(InterruptedException ie){
        ie.printStackTrace();
    }
    return data;
  }
}

package com.qed;

public class Consumer implements Runnable{

private Store st;
public Consumer(Store st){
    this.st=st;
}


public void run(){
    while(true){
        System.out.println(Thread.currentThread().getName() + ". " +st.remove());
    }
  }
}

package com.qed;

public class Producer implements Runnable{

private Store st;
private final int runs = 5000;
public Producer(Store st){
    this.st = st;
}

public void run(){
    int data = 0;
    int curRun =0;
    while(++curRun < runs){
        st.add(data+=200);
        }
    System.out.println("DONE.");
  }
}

Stack trace:

Exception in thread "B" Exception in thread "A"  
java.util.NoSuchElementException    
    at java.util.LinkedList.removeFirst(Unknown Source)    
    at com.qed.Store.remove(Store.java:46)    
    at com.qed.Consumer.run(Consumer.java:20)    
    at java.lang.Thread.run(Unknown Source)    
java.util.NoSuchElementException    
    at java.util.LinkedList.removeFirst(Unknown Source)    
    at com.qed.Store.remove(Store.java:46)    
    at com.qed.Consumer.run(Consumer.java:20)    
    at java.lang.Thread.run(Unknown Source)    

Answer 1

You have to lock on adding as well. Your code allows a producer to update the queue while the consumer might want to remove an entry!

When two threads modify the same queue in parallel, all bets are off!

That single lock usage would only prevent multiple consumers to step on each other!

Thus: add the same kind of locking for the section that adds values.

Beyond that, EJP is correct - a real solution would make use of low level signaling methods such as wait() and notify(). But of course, using these would lead to a very different behavior.

And given your comment: keep in mind that these are TWO different things: A) consumer/producer sending signals to each other B) consumer/producing synchronizing on the same look.

I understand that you don't want "A)" - but you need "B)"; otherwise your queue gets corrupted, and surprises occur.

Answer 2

Problem is your Store class implementation. Instead of sleeping, you need to implement a wait() and notify mechanism there while adding and removing elements.

You are correct in sharing a single Store instance among all consumers and producers but your store needs to behave like a BlockingQueue

So either you use an existing implementation of BlockingQueue from JDK or modify your Store class to implement similar mechanism.

implement-your-own blocking queue in java

Hope it helps !!

Answer 3

You should call wait() method here.
wait() makes your thread wait until some other thread calls notify to wake him up.
sleep() simply don't execute the next statement for specified time period.
And if you see your program snippet, you are using synchronize block which uses an object to check for the monitor availability. But you are not using any of the object monitor methods wait/notify/notifyAll and you're trying to acquire and release the lock without calling these methods. Since, list object used by both the consumer and producer you should use this list objects monitor to synch all the threads. If one thread acquired its monitor then other thread won't be able to access it. Because every object has only one monitor. This approach will solve the synchronisation issue among all the working threads.