how to properly implement producer - consumer in Java

Question

I implemented producer consumer like this.

But it is throwing error.

I tried to use this method of using lock. Link

class Testclass {
    Boolean isFresh = false;
    int count = 0;
    public synchronized void GET(String threadName){
        while(!isFresh){
            try {
                isFresh.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        System.out.println("GET method was called : " + count + " " + threadName);
        isFresh = false;
        isFresh.notify();

    }

    public synchronized void PUT(String threadName){
        while(isFresh){
            try{
                isFresh.wait();
            }catch( InterruptedException e){
                e.printStackTrace();
            }
        }
        count++;
        System.out.println("PUT method was called : " + count + " " + threadName);
        isFresh = true;
        isFresh.notify();
    }

}
class Producer implements Runnable{
    Testclass q;
    String name;
    Producer(Testclass q, String name){
        this.q = q;
        this.name = name;
    }
    public void run(){
        while(true){
            int time = (int)(Math.random() * 10000);

            try {
                Thread.sleep(time);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            q.PUT(this.name);
        }

    }
}

class Consumer implements Runnable{
    Testclass q;
    String name ;
    Consumer(Testclass q,String name){
        this.q = q;
        this.name = name;
    }
    public void run(){
        while(true){
            int time = (int)(Math.random() * 10000);

            try {
                Thread.sleep(time);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            q.GET(this.name);
        }

    }
}


public class Main {
    public static void main(String[] args){
        Testclass t = new Testclass();
        Thread t1 = new Thread(new Consumer(t, "consumer 1"));
        Thread t2 = new Thread(new Consumer(t, "consumer 2"));
        Thread t3 = new Thread(new Producer(t, "producer 1"));
        Thread t4 = new Thread(new Producer(t, "producer 2"));
        t1.start();
        t2.start();
        t3.start();
        t4.start();
        try{
            t1.join();
        }catch (InterruptedException e){
            e.printStackTrace();
        }

    }



}

this implementation throws following error.

Please explain why all the threads are throwing Illegal MonitorStateException?

PUT method was called : 1 producer 1
Exception in thread "Thread-2" java.lang.IllegalMonitorStateException: current thread is not owner
    at java.base/java.lang.Object.notify(Native Method)
    at Testclass.PUT(Main.java:34)
    at Producer.run(Main.java:54)
    at java.base/java.lang.Thread.run(Thread.java:832)
GET method was called : 1 consumer 1
Exception in thread "Thread-0" java.lang.IllegalMonitorStateException: current thread is not owner
    at java.base/java.lang.Object.notify(Native Method)
    at Testclass.GET(Main.java:19)
    at Consumer.run(Main.java:76)
    at java.base/java.lang.Thread.run(Thread.java:832)
Exception in thread "Thread-1" java.lang.IllegalMonitorStateException: current thread is not owner
    at java.base/java.lang.Object.wait(Native Method)
    at java.base/java.lang.Object.wait(Object.java:321)
    at Testclass.GET(Main.java:12)
    at Consumer.run(Main.java:76)
    at java.base/java.lang.Thread.run(Thread.java:832)
PUT method was called : 2 producer 2
Exception in thread "Thread-3" java.lang.IllegalMonitorStateException: current thread is not owner
    at java.base/java.lang.Object.notify(Native Method)
    at Testclass.PUT(Main.java:34)
    at Producer.run(Main.java:54)
    at java.base/java.lang.Thread.run(Thread.java:832)

Process finished with exit code 0

I want to know why is the output like that? and what is the correct way of implementing it?

To use `isFresh.await()` or `notify()`, you need to be synchronized on `isFresh`. — Andy Turner, Jul 27 '20 at 09:19
Right now you're using a [synchronized method](https://docs.oracle.com/javase/tutorial/essential/concurrency/syncmeth.html) you either need to change the object you call wait/notify on or use a [synchronized statement](https://docs.oracle.com/javase/tutorial/essential/concurrency/locksync.html). — matt, Jul 27 '20 at 09:43
Java programming conventions use lower case letters to start method names such as get() and put(). — NomadMaker, Jul 27 '20 at 09:45
@AndyTurner that won't help, because he changed the instance of the monitor object during the wait. OP: decouple the monitor from the IsFresh Boolean flag. Booleans are immutable so you're not changing the value of the object - you're changing the reference to the object — Erwin Bolwidt, Jul 27 '20 at 09:52
I think the point was, if you call wait/noitify you need to be synchronized on the object. As Erwin has pointed out, the object is changing. So you could make this broken implementation run by changing `isFresh.notify()` to `notifyAll()` and change `isFresh.wait()` to `wait()`. — matt, Jul 27 '20 at 09:59

matt · Answer 1 · 2020-07-27T09:39:34.087

You should use a BlockingQueue. Then you don't have to use wait/notify or even synchronzied.

class Testclass {
    BlockingQueue<Integer> queue = new ArrayBlockingQueue(1);
    AtomicInteger count = new AtomicInteger(0);
    public void GET(String threadName) throws InterruptedException{
        Integer i = queue.take();
        count.getAndIncrement();
        System.out.println("GET method was called : " + count + " " + threadName);
    }

    public void PUT(String threadName) throws InterruptedException{

        int c = count.get();
        queue.put(c);
        count.getAndIncrement();
        System.out.println("PUT method was called : " + count + " " + threadName);
    }

}

I'm not sure what you want to do with count. This is readily extensible by changing the size of your Queue.

how to properly implement producer - consumer in Java

1 Answers1