Understanding producer-consumer using java sync
I am working on a PC problem to understand Java sync and cross-thread communication. Using the code below, the output was
Producer produced-0
Producer produced-1
Producer produced-2
Consumer consumed-0
Consumer consumed-1
Consumer consumed-2
Producer produced-3
Producer produced-4
Producer produced-5
Consumer consumed-3
Consumer consumed-4
But something like below shouldn't be output
Producer produced-0
Consumer consumed-0
Producer produced-1
Consumer consumed-1
Producer produced-2
Consumer consumed-2
Producer produced-3
I expect this kind of output because, as I understand it, the consumer is notified of the received value as soon as the release method releases the lock when the method completes. As a result, the consumer block that was waiting enters a synchronized state sync lock to consume the produced value, while the producer method blocks. this lock is released at the end of the consuming method, which is obtained by the producer thread that was blocked due to synchronization, and the loop continues as each method blocks due to blocking.
Please let me know what I have misunderstood? Thanks to
package MultiThreading;
//Java program to implement solution of producer
//consumer problem.
import java.util.LinkedList;
public class PCExample2
{
public static void main(String[] args)
throws InterruptedException
{
// Object of a class that has both produce()
// and consume() methods
final PC pc = new PC();
// Create producer thread
Thread t1 = new Thread(new Runnable()
{
@Override
public void run()
{
try
{
while (true) {
pc.produce();
}
}
catch(InterruptedException e)
{
e.printStackTrace();
}
}
});
// Create consumer thread
Thread t2 = new Thread(new Runnable()
{
@Override
public void run()
{
try
{
while (true) {
pc.consume();
}
}
catch(InterruptedException e)
{
e.printStackTrace();
}
}
});
// Start both threads
t1.start();
t2.start();
// t1 finishes before t2
t1.join();
t2.join();
}
// This class has a list, producer (adds items to list
// and consumber (removes items).
public static class PC
{
// Create a list shared by producer and consumer
// Size of list is 2.
LinkedList<Integer> list = new LinkedList<>();
int capacity = 12;
int value = 0;
// Function called by producer thread
public void produce() throws InterruptedException
{
synchronized (this)
{
// producer thread waits while list
// is full
while (list.size()==capacity)
wait();
System.out.println("Producer produced-"
+ value);
// to insert the jobs in the list
list.add(value++);
// notifies the consumer thread that
// now it can start consuming
notify();
// makes the working of program easier
// to understand
Thread.sleep(1000);
}
}
// Function called by consumer thread
public void consume() throws InterruptedException
{
synchronized (this)
{
// consumer thread waits while list
// is empty
while (list.size()==0)
wait();
//to retrive the ifrst job in the list
int val = list.removeFirst();
System.out.println("Consumer consumed-"
+ val);
// Wake up producer thread
notify();
// and sleep
Thread.sleep(1000);
}
}
}
}
source to share
It is not necessary that the first thread calls for the current lock (let's call it Thread A) will have the lock as soon as the current lock thread has canceled it, if other threads also made calls to the lock because Thread A tried to acquire it. There is no ordered "queue". See here and here . Thus, judging by the output of the program, it seems that after the producer releases the lock, there may not be enough time for the consumer to acquire the lock before the loop while
in the producer thread is repeated and the producer thread makes another lock call (like pointed out other answers,Thread.sleep()
does not cause the sleeping stream to give up the lock) and if the consumer is unlucky, the producer will re-acquire the lock even if the consumer was the first.
However, there seems to be another misunderstanding. No producer will ever "wait" on PC
until the list contains 12 items, so the consumer thread is only guaranteed the fact that the producer has released at least 12 items (which incidentally happens when I run the program - the consumer will never have chance until the producer thread calls wait()
on the PC, but then it consumes the entire list). This also means that if it is a consumer turn and the list contains less than 12 items, the producer thread will not be notified because it will not wait , but will only be notified when blocked and already, say, "waiting" or "waiting" for blocking on PC (see also hereabout the difference between "waiting" and "blocked" "). So even if you put the two Thread.sleep()
invocations outside the sync blocks, thereby giving the consumer thread (hopefully you shouldn't rely on this) enough time to acquire the lock, the call notify()
from the consumer thread will have no effect, because the manufacturer's thread will never be in a waiting state.
To make sure that both threads change PC
alternately, you will only need to make the renewal thread wait if the list size is greater than zero, not if the list contains 12 (or as many) items.
source to share
Notice two mothods: notify && Thread.sleep
Object.notify ():
Wakes up a single thread waiting for this monitor object. If there are any threads waiting on this object, one of them is selected to wake up. The choice is arbitrary and is at the discretion of the implementation. A thread waits for a monitor object by calling one of the wait methods.
The awakened thread will not be able to act until the current thread releases the lock on this object. The awakened thread will compete as usual with any other threads that may be actively competing for synchronization on this entity ; for example, an awakened thread does not have a secure privilege or lack on the next thread to lock this object.
Thread.sleep ():
Causes the currently executing thread to sleep (temporarily stop execution) in the specified number of milliseconds plus the specified number of nanoseconds, taking into account the precision and accuracy of system timers and schedulers. The thread does not lose ownership of any monitors .
OK. Now you know that the notification will just wake up the thread that also controls this object, but the awakened thread will compete to synchronize with this object. If your producer notifies the consumer and releases the lock, then the producer and consumer are at the same point to compete. And Thread.sleep doesn't do the work you want, it won't release the lock when it sleeps like the doc said. This can happen.
In conclusion, Thread.sleep is not very good with synchronization. and even if you remove that, the first exit is due to the notification mechanism.
@ Andrew S's answer will work.
source to share
From the API : the awakened thread will compete in the usual way with any other threads that may actively compete for synchronization to this object; for example, an awakened thread does not have a secure privilege or lack on the next thread to lock this object.
Move sleep()
outside of the synchronized block to give another thread an advantage to acquire the lock.
source to share
Simply adding the appropriate condition will do the job.
import java.util.LinkedList;
import java.util.Queue;
class Producer extends Thread {
public Queue<Integer> producerQueue;
public int size;
public int count = 0;
Producer(Queue<Integer> queue, int size) {
producerQueue = queue;
this.size = size;
}
public void produce() throws InterruptedException {
synchronized (producerQueue) {
while (producerQueue.size() > 0) {
producerQueue.wait();
}
System.out.println("Produced : " + count);
producerQueue.add(count++);
producerQueue.notify();
Thread.sleep(100);
}
}
public void run() {
try {
while (true) produce();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
class Consumer extends Thread {
public Queue<Integer> consumerQueue;
public int size;
Consumer(Queue<Integer> queue, int size) {
consumerQueue = queue;
this.size = size;
}
public void consume() throws InterruptedException {
synchronized (consumerQueue) {
while (consumerQueue.size() == 0) {
consumerQueue.wait();
Thread.sleep(100);
}
System.out.println("Consumed : " + consumerQueue.poll());
consumerQueue.notify();
}
}
public void run() {
try {
while (true) consume();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
public class Test {
public static void main(String[] args) {
Queue<Integer> commonQueue = new LinkedList<>();
int size = 10;
new Producer(commonQueue, size).start();
new Consumer(commonQueue, size).start();
}
}
source to share