BlockingQueue Implementation

BlockingQueue :

java.util.concurrent.BlockingQueue is an interface that extends the java.util.Queue that add special support for blocking that it will wait for space to become available in the queue while storing and will wait for the queue to become non empty while retrieving elements.

Implementation of BlockingQueue :

util.concurrent.ArrayBlockingQueue : ArrayBlockingQueue is a fixed size bounded queue
util.concurrent.DelayQueue : The queue is unbounded and Elements added to the queue must implement thejava.util.concurrent.Delayed interface. Elements can be retrieved from the queue when the delay time has expired.
util.concurrent.LinkedBlockingQueue : LinkedBlockingQueue is an optionally-bounded blocking queue. This queue orders elements FIFO (first-in-first-out). The LinkedBlockingQueue keeps the elements internally in a linked structure
util.concurrent.PriorityBlockingQueue : A priority queue is unbounded queue where elements are ordered by natural order or by a Comparator provided at queue construction time .It does not permit null elements.
util.concurrent.SynchronousQueue : SynchronousQueue is a queue that can only contain a single element .In a SynchronousQueue an insert will wait for a remove operation by another thread and vice versa.

ArrayBlockingQueue implementation of BlockingQueue.

As already mentioned above, an ArrayBlockingQueue is a fixed size bounded queue. It is backed by array. As it is a BlockingQueue , its put(e) method will block a thread and waits for space to become available in the queue, and its take() method will block other thread and wait for an element become available in the queue. It is declared as given below

private static BlockingQueue<String> bq = new ArrayBlockingQueue<String>(no_of_element);

Methods of ArrayBlockingQueue :

put(E item) method in java –This method checks whether space is available or not.If space is not available it will wait else it will insert element and notify all waiting threads in java.

E take() method in java -This method Checks whether element is available or not.If element is not available it will wait else it will remove element and notify all waiting threads in java.

Peek(): This method gets the element from head but not removed and returns null if not available

element(): This method gets the element from head but not removed and throw exception if not available

Example of ArrayBlockingQueue:

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class BlockingQueueDemo {
   public static void main(String args[]){
      BlockingQueue<Integer> sharedQueue = new ArrayBlockingQueue<Integer>(5);
      Thread producerThread = new Thread(new ProducerThread(sharedQueue), "ProducerThread");
      Thread consumerThread = new Thread(new ConsumerThread(sharedQueue), "ConsumerThread");
      producerThread.start();
      consumerThread.start();
  }
}

import java.util.concurrent.ArrayBlockingQueue;

import java.util.concurrent.BlockingQueue;

public class BlockingQueueDemo {

public static void main(String args[]){

BlockingQueue<Integer> sharedQueue = new ArrayBlockingQueue<Integer>(5);

Thread producerThread = new Thread(new ProducerThread(sharedQueue), "ProducerThread");

Thread consumerThread = new Thread(new ConsumerThread(sharedQueue), "ConsumerThread");

producerThread.start();

consumerThread.start();

}

import java.util.concurrent.BlockingQueue;

class ConsumerThread implements Runnable{
      private BlockingQueue<Integer> sQueue;
      public ConsumerThread (BlockingQueue<Integer> sQueue) {
         this.sQueue = sQueue;
      }

      @Override
      public void run() {
       while(true){
         try {
              System.out.println("Consumer retrieved item : "+ sQueue.take());
             } catch (InterruptedException ex) {
            }
       }
      }
    }

import java.util.concurrent.BlockingQueue;

class ConsumerThread implements Runnable{

private BlockingQueue<Integer> sQueue;

public ConsumerThread (BlockingQueue<Integer> sQueue) {

this.sQueue = sQueue;

}

@Override

public void run() {

while(true){

try {

System.out.println("Consumer retrieved item : "+ sQueue.take());

} catch (InterruptedException ex) {

}

import java.util.concurrent.BlockingQueue;

class ProducerThread implements Runnable {

      private final BlockingQueue<Integer> sQueue;
      public ProducerThread(BlockingQueue<Integer> sQueue) {
         this.sQueue = sQueue;
      }

      @Override
      public void run() {
       for(int i=1; i<=5; i++){
        try {
              System.out.println("Producer produced item: " + i);
              sQueue.put(i);
            } catch (InterruptedException ex) {
            }
       }
     }
   }

import java.util.concurrent.BlockingQueue;

class ProducerThread implements Runnable {

private final BlockingQueue<Integer> sQueue;

public ProducerThread(BlockingQueue<Integer> sQueue) {

this.sQueue = sQueue;

}

@Override

public void run() {

for(int i=1; i<=5; i++){

try {

System.out.println("Producer produced item: " + i);

sQueue.put(i);

} catch (InterruptedException ex) {

}

Output:
Producer produced item: 1
Producer produced item: 2
Producer produced item: 3
Producer produced item: 4
Producer produced item: 5
Consumer retrieved item : 1
Consumer retrieved item : 2
Consumer retrieved item : 3
Consumer retrieved item : 4
Consumer retrieved item : 5

Realtime Use cases:
1) Realtime use case for BlockingQueue is Chat application in which messages are put into a queue. Produces will produce the messages from one end and consumer will consume messages.

2) ThreadPoolExecutor uses BlockingQueue to put executable task which get executed by worker threads.

3) A real world example could be an application-wide notification framework, which sends mails to the stakeholders at various points during the course of application usage.

Related Posts

Reader Interactions

Leave a Reply Cancel reply

Footer