I am trying to write produce and consumer that should print output in below order
consumer 1
produce 1
consumer 2
produce 2
But it's not giving outputs in order. To achieve order when using synchronized keyword, it's not printing even any output.
import java.util.concurrent.ArrayBlockingQueue;
import java.util.stream.IntStream;
public class ProducerAndCosumer {
ArrayBlockingQueue<Integer> blockingQueue ;
public ProducerAndCosumer(int capacity) {
this.blockingQueue = new ArrayBlockingQueue<Integer>(capacity);
}
public synchronized void consume(int data) {
System.out.println(" consumer " + data);
blockingQueue.offer(data);
}
public synchronized int produce() throws InterruptedException {
int data = blockingQueue.take();
System.out.println(" produce " + data);
return data;
}
public static void main(String[] args) throws InterruptedException {
ProducerAndCosumer pc = new ProducerAndCosumer(10);
Runnable run1 = new Runnable() {
@Override
public void run() {
System.out.println(" Entered consumer runner ");
IntStream.range(1, 20).forEach(pc::consume);
}
};
new Thread(run1).start();
Runnable run2 = new Runnable() {
@Override
public void run() {
System.out.println(" Entered producer runner ");
try {
for (int i = 0 ; i < 30 ; i++) {
pc.produce();
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
};
new Thread(run2).start();
}
}
Please, suggest how can I resolve it. And one more question can we achieve it without wait and notify.
in your code, you have the race condition problem. You should provide concurrent access to your ArrayBlockingQueue
. One of a lot of possible variants is the next example(with two semaphores):
public class ProducerAndConsumer {
private final ArrayBlockingQueue<Integer> blockingQueue ;
private final Semaphore semaphoreOne;
private final Semaphore semaphoreTwo;
public ProducerAndConsumer(int capacity) {
this.blockingQueue = new ArrayBlockingQueue<>(capacity, true);
this.semaphoreOne = new Semaphore(1);
this.semaphoreTwo = new Semaphore(1);
}
public void consume(int data) throws InterruptedException {
semaphoreOne.acquire();
System.out.println(" consumer " + data);
blockingQueue.offer(data);
semaphoreTwo.release();
}
public void produce() throws InterruptedException {
semaphoreTwo.acquire();
int data = blockingQueue.take();
System.out.println(" produce " + data);
semaphoreOne.release();
}
public static void main(String[] args) throws InterruptedException {
ProducerAndConsumer pc = new ProducerAndConsumer(10);
Runnable run1 = () -> {
System.out.println(" Entered consumer runner ");
for (int i = 1; i < 20; i++) {
try {
pc.consume(i);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
};
Runnable run2 = () -> {
System.out.println(" Entered producer runner ");
try {
for (int i = 0 ; i < 30 ; i++) {
pc.produce();
}
} catch (InterruptedException e) {
e.printStackTrace();
}
};
new Thread(run1).start();
new Thread(run2).start();
}
}
Output for this program is:
Entered consumer runner
Entered producer runner
consumer 1
produce 1
consumer 2
produce 2
consumer 3
produce 3
consumer 4
produce 4
consumer 5
produce 5
consumer 6
produce 6
consumer 7
produce 7
consumer 8
produce 8
consumer 9
produce 9
consumer 10
produce 10
consumer 11
produce 11
consumer 12
produce 12
consumer 13
produce 13
consumer 14
produce 14
consumer 15
produce 15
consumer 16
produce 16
consumer 17
produce 17
consumer 18
produce 18
consumer 19
produce 19
Besides, when you declare both your methods consume
produce
synchronized
you will have a deadlock since both methods will try to synchronize by one monitor(object ProducerAndConsumer
), see ThreadDump