Producer consumer problem - java
public class ProducerConsumerSolution {
public static void main(String args[]) {
Vector sharedQueue = new Vector();
int size = 4;
Thread prodThread = new Thread(new Producer(sharedQueue, size), "Producer");
Thread consThread = new Thread(new Consumer(sharedQueue, size), "Consumer");
prodThread.start();
consThread.start();
}
}
class Producer implements Runnable {
private final Vector sharedQueue;
private final int SIZE;
public Producer(Vector sharedQueue, int size) {
this.sharedQueue = sharedQueue;
this.SIZE = size;
}
@Override
public void run() {
for (int i = 0; i < 7; i++) {
System.out.println("Produced: " + i);
try {
produce(i);
} catch (InterruptedException ex) {
Logger.getLogger(Producer.class.getName()).log(Level.SEVERE, null, ex);
}
}
}
private void produce(int i) throws InterruptedException {
//wait if queue is full
while (sharedQueue.size() == SIZE) {
synchronized (sharedQueue) {
System.out.println("Queue is full " + Thread.currentThread().getName()
public static void main(String args[]) {
Vector sharedQueue = new Vector();
int size = 4;
Thread prodThread = new Thread(new Producer(sharedQueue, size), "Producer");
Thread consThread = new Thread(new Consumer(sharedQueue, size), "Consumer");
prodThread.start();
consThread.start();
}
}
class Producer implements Runnable {
private final Vector sharedQueue;
private final int SIZE;
public Producer(Vector sharedQueue, int size) {
this.sharedQueue = sharedQueue;
this.SIZE = size;
}
@Override
public void run() {
for (int i = 0; i < 7; i++) {
System.out.println("Produced: " + i);
try {
produce(i);
} catch (InterruptedException ex) {
Logger.getLogger(Producer.class.getName()).log(Level.SEVERE, null, ex);
}
}
}
private void produce(int i) throws InterruptedException {
//wait if queue is full
while (sharedQueue.size() == SIZE) {
synchronized (sharedQueue) {
System.out.println("Queue is full " + Thread.currentThread().getName()
+ " is waiting , size: " + sharedQueue.size());
sharedQueue.wait();
}
}
//producing element and notify consumers
synchronized (sharedQueue) {
sharedQueue.add(i);
sharedQueue.notifyAll();
}
}
}
class Consumer implements Runnable {
private final Vector sharedQueue;
private final int SIZE;
public Consumer(Vector sharedQueue, int size) {
this.sharedQueue = sharedQueue;
this.SIZE = size;
}
@Override
public void run() {
while (true) {
try {
System.out.println("Consumed: " + consume());
Thread.sleep(50);
} catch (InterruptedException ex) {
Logger.getLogger(Consumer.class.getName()).log(Level.SEVERE, null, ex);
}
}
}
private int consume() throws InterruptedException {
//wait if queue is empty
while (sharedQueue.isEmpty()) {
synchronized (sharedQueue) {
System.out.println("Queue is empty " + Thread.currentThread().getName()
sharedQueue.wait();
}
}
//producing element and notify consumers
synchronized (sharedQueue) {
sharedQueue.add(i);
sharedQueue.notifyAll();
}
}
}
class Consumer implements Runnable {
private final Vector sharedQueue;
private final int SIZE;
public Consumer(Vector sharedQueue, int size) {
this.sharedQueue = sharedQueue;
this.SIZE = size;
}
@Override
public void run() {
while (true) {
try {
System.out.println("Consumed: " + consume());
Thread.sleep(50);
} catch (InterruptedException ex) {
Logger.getLogger(Consumer.class.getName()).log(Level.SEVERE, null, ex);
}
}
}
private int consume() throws InterruptedException {
//wait if queue is empty
while (sharedQueue.isEmpty()) {
synchronized (sharedQueue) {
System.out.println("Queue is empty " + Thread.currentThread().getName()
+ " is waiting , size: " + sharedQueue.size());
sharedQueue.wait();
}
}
//Otherwise consume element and notify waiting producer
synchronized (sharedQueue) {
sharedQueue.notifyAll();
return (Integer) sharedQueue.remove(0);
}
}
}
Output:
Produced: 0
Queue is empty Consumer is waiting , size: 0
Produced: 1
Consumed: 0
Produced: 2
Produced: 3
Produced: 4
Produced: 5
Queue is full Producer is waiting , size: 4
Consumed: 1
Produced: 6
Queue is full Producer is waiting , size: 4
Consumed: 2
Consumed: 3
Consumed: 4
Consumed: 5
Consumed: 6
Queue is empty Consumer is waiting , size: 0
sharedQueue.wait();
}
}
//Otherwise consume element and notify waiting producer
synchronized (sharedQueue) {
sharedQueue.notifyAll();
return (Integer) sharedQueue.remove(0);
}
}
}
Output:
Produced: 0
Queue is empty Consumer is waiting , size: 0
Produced: 1
Consumed: 0
Produced: 2
Produced: 3
Produced: 4
Produced: 5
Queue is full Producer is waiting , size: 4
Consumed: 1
Produced: 6
Queue is full Producer is waiting , size: 4
Consumed: 2
Consumed: 3
Consumed: 4
Consumed: 5
Consumed: 6
Queue is empty Consumer is waiting , size: 0
Read more: http://java67.blogspot.com/2012/12/producer-consumer-problem-with-wait-and-notify-example.html#ixzz3xFTjA1Me
-----------------------------------------------------------------------------------------------------------------
First, the
BlockingQueueExample
class which starts a Producer
and a Consumer
in separate threads. TheProducer
inserts strings into a shared BlockingQueue
, and the Consumer
takes them out.public class BlockingQueueExample { public static void main(String[] args) throws Exception { BlockingQueue queue = new ArrayBlockingQueue(1024); Producer producer = new Producer(queue); Consumer consumer = new Consumer(queue); new Thread(producer).start(); new Thread(consumer).start(); Thread.sleep(4000); } }
Here is the
Producer
class. Notice how it sleeps a second between each put()
call. This will cause theConsumer
to block, while waiting for objects in the queue.public class Producer implements Runnable{ protected BlockingQueue queue = null; public Producer(BlockingQueue queue) { this.queue = queue; } public void run() { try { queue.put("1"); Thread.sleep(1000); queue.put("2"); Thread.sleep(1000); queue.put("3"); } catch (InterruptedException e) { e.printStackTrace(); } } }
Here is the
Consumer
class. It just takes out the objects from the queue, and prints them to System.out
.public class Consumer implements Runnable{ protected BlockingQueue queue = null; public Consumer(BlockingQueue queue) { this.queue = queue; } public void run() { try { System.out.println(queue.take()); System.out.println(queue.take()); System.out.println(queue.take()); } catch (InterruptedException e) { e.printStackTrace(); } } }
Ref: http://tutorials.jenkov.com/java-util-concurrent/blockingqueue.html
Comments
Post a Comment