Producer Consumer problem is a popular problem domain in SE industry. Its hard find anyone who doesn't have heard it.
So in this post I will solve this producer consumer problem by using Blocking Queue in java.
Before that I will give a quick introduction of Blocking Queue.
Blocking Queue is an interface locate in java concurrent package.It mainly support operations that wait for the queue to become non empty when retrieving and removing element and wait for space become available when adding an element. All the blocking queue implementation are thread-safe and methods are atomic.
In this demo I will use ArrayBlockingQueue as the implementation of BlockQueue.
So first we create domain model for this demo.
Message.java
package rd.domain;
public class Message {
private String description;
public String getDescription() {
return description;
}
public void setDescription(String description) {
this.description = description;
}
}
then we create
Producer.java
create 100 message and finish message and add in to the queue.
package rd.concurent;
import rd.domain.Message;
import java.util.concurrent.BlockingQueue;
public class Producer implements Runnable {
private BlockingQueue<Message> queue;
public Producer(BlockingQueue<Message> queue) {
this.queue = queue;
}
@Override
public void run() {
// create messages and adding to queue
for (int i = 1; i <= 100; i++) {
Message message = new Message();
message.setDescription(" Message " + i);
try {
Thread.sleep(i);
queue.put(message);
System.out.println("Produced " + message.getDescription());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
// adding exit message
Message message = new Message();
message.setDescription("finish");
try {
queue.put(message);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
Consumer.java
package rd.concurent;
import rd.domain.Message;
import java.util.concurrent.BlockingQueue;
public class Consumer implements Runnable {
private BlockingQueue<Message> queue;
public Consumer(BlockingQueue<Message> queue) {
this.queue = queue;
}
@Override
public void run() {
Message message = null;
try {
while (!(message = queue.take()).getDescription().endsWith("finish")){
Thread.sleep(10);
System.out.println("Consumed " + message.getDescription());
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
So the basic implementation are done. Now its time to test it. Here I create
Main.java and create producer consumer thread and start those threads.
Main.java
package rd;
import rd.concurent.Consumer;
import rd.concurent.Producer;
import rd.domain.Message;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
public class Main {
public static void main(String[] args) {
BlockingQueue<Message> blockingQueue = new ArrayBlockingQueue<>(10);
Producer producer = new Producer(blockingQueue);
Consumer consumer = new Consumer(blockingQueue);
new Thread(consumer).start();
new Thread(producer).start();
System.out.println("Started.................");
}
}
So that's it. We just solved the producer consumer problem.
