BlockingQueue

This Article is part of Series of Articles on Java 8 Concurrency Tutorial.
In this article, we’ll focus on a the concept of BlockingQueue in the Java language.

BlockingQueue

A Queue that additionally supports operations that wait for the queue to become non-empty when retrieving an element, and wait for space to become available in the queue when storing an element.

A BlockingQueue may be capacity bounded. At any given time it may have a remainingCapacity beyond which no additional elements can be put without blocking. A BlockingQueue without any intrinsic capacity constraints always reports a remaining capacity of Integer.MAX_VALUE.

BlockingQueue implementations are thread-safe. All queuing methods achieve their effects atomically using internal locks or other forms of concurrency control. However, the bulk Collection operations addAll, containsAll, retainAll and removeAll are not necessarily performed atomically unless specified otherwise in an implementation. So it is possible, for example, for addAll(c) to fail (throwing an exception) after adding only some of the elements in c.

let’s implement Producer and Consumer Problem using BlockingQueue

BlockingQueue Example

class BlockingQueueProducer implements Runnable
{
	private BlockingQueue<Integer> blockingQueue;
	private Random random;
	public BlockingQueueProducer(BlockingQueue<Integer> blockingQueue) {
		this.blockingQueue=blockingQueue;
		this.random = new Random();
		
	}
	@Override
	public void run() {
		while(true)
		{
			System.out.println("BlockingQueueProducer - Adding Elements");
			try {
				blockingQueue.put(random.nextInt(3000));
				System.out.println("Added Element : Current Size of Q " + blockingQueue.size());
				Thread.sleep(2000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}		
	}	
}

class BlockingQueueConsumer implements Runnable
{
	BlockingQueue<Integer> blockingQueue;
	public BlockingQueueConsumer(BlockingQueue<Integer> blockingQueue) {
		this.blockingQueue=blockingQueue;
	}

	@Override
	public void run() {
		while (true) {

			try {
				System.out.println("BlockingQueueConsumer : iteam recived from Q " + blockingQueue.take() );
				Thread.sleep(2500);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}	
}
public class BlockingQueueExample {

	public static void main(String[] args) {
		BlockingQueue<Integer> blockingQueue = new ArrayBlockingQueue<Integer>(5);
		ExecutorService newFixedThreadPool = Executors.newFixedThreadPool(2);
		newFixedThreadPool.submit(new BlockingQueueConsumer(blockingQueue));
		newFixedThreadPool.submit(new BlockingQueueProducer(blockingQueue));
		newFixedThreadPool.shutdown();
	}
}

Key Points

put(E e)

Inserts the specified element into this queue, waiting if necessary for space to become available.

take()

Retrieves and removes the head of this queue, waiting if necessary until an element becomes available.

poll()

Retrieves and removes the head of this queue, waiting up to the specified wait time if necessary for an element to become available.