Operating Systems Programming

Operating Systems Programming via Java Producer Consumer Problem with Blocking Queue

Producer Consumer problem is a popular problem domain in SE industry.
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.................");  
   }  
 }  

Hence we had  solved the producer consumer problem.