BlockingQueue Interface in Java
The BlockingQueue interface in Java is added in Java 1.5 along with various other concurrent Utility classes like ConcurrentHashMap, Counting Semaphore, CopyOnWriteArrrayList, etc. BlockingQueue interface supports flow control (in addition to queue) by introducing blocking if either BlockingQueue is full or empty. A thread trying to enqueue an element in a full queue is blocked until some other thread makes space in the queue, either by dequeuing one or more elements or clearing the queue completely. Similarly, it blocks a thread trying to delete from an empty queue until some other threads insert an item. BlockingQueue does not accept a null value. If we try to enqueue the null item, then it throws NullPointerException.
Java provides several BlockingQueue implementations such as LinkedBlockingQueue, ArrayBlockingQueue, PriorityBlockingQueue, SynchronousQueue, etc. Java BlockingQueue interface implementations are thread-safe. All methods of BlockingQueue are atomic in nature and use internal locks or other forms of concurrency control. Java 5 comes with BlockingQueue implementations in the java.util.concurrent package.
Usage of BlockingQueue
A BlockingQueue accessed by a producer(put) thread and a consumer(take) thread
The Hierarchy of BlockingQueue
Declaration
public interface BlockingQueue extends Queue
Here, E is the type of elements stored in the Collection.
Classes that Implement BlockingQueue
We directly cannot provide an instance of BlockingQueue since it is an interface, so to utilize the functionality of the BlockingQueue, we need to make use of the classes implementing it. Also, to use BlockingQueue in your code, use this import statement.
import java.util.concurrent.BlockingQueue; (or) import java.util.concurrent.*;
The implementing class of BlockingDeque is LinkedBlockingDeque. This class is the implementation of the BlockingDeque and the linked list data structure. The LinkedBlockingDeque can be optionally bounded using a constructor, however, if the capacity is unspecified it is Integer.MAX_VALUE by default. The nodes are added dynamically at the time of insertion obeying the capacity constraints.
The syntax for creating objects:
BlockingQueue objectName = new LinkedBlockingDeque(); (or) LinkedBlockingDeque objectName = new LinkedBlockingDeque();
Example: In the code given below we perform some basic operations on a LinkedBlockingDeque, like creating an object, adding elements, deleting elements, and using an iterator to traverse through the LinkedBlockingDeque.
BlockingQueue Types
The BlockingQueue are two types:
1. Unbounded Queue: The Capacity of the blocking queue will be set to Integer.MAX_VALUE. In the case of an unbounded blocking queue, the queue will never block because it could grow to a very large size. when you add elements its size grows.
BlockingQueue blockingQueue = new LinkedBlockingDeque();
2. Bounded Queue: The second type of queue is the bounded queue. In the case of a bounded queue you can create a queue passing the capacity of the queue in queues constructor:
// Creates a Blocking Queue with capacity 5 BlockingQueue blockingQueue = new LinkedBlockingDeque(5);
To implement Bounded Semaphore using BlockingQueue
Interface 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.
BlockingQueue methods come in four forms, with different ways of handling operations that cannot be satisfied immediately, but may be satisfied at some point in the future: one throws an exception, the second returns a special value (either null or false , depending on the operation), the third blocks the current thread indefinitely until the operation can succeed, and the fourth blocks for only a given maximum time limit before giving up. These methods are summarized in the following table:
| Throws exception | Special value | Blocks | Times out | |
|---|---|---|---|---|
| Insert | add(e) | offer(e) | put(e) | offer(e, time, unit) |
| Remove | remove() | poll() | take() | poll(time, unit) |
| Examine | element() | peek() | not applicable | not applicable |
A BlockingQueue does not accept null elements. Implementations throw NullPointerException on attempts to add , put or offer a null . A null is used as a sentinel value to indicate failure of poll operations.
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 designed to be used primarily for producer-consumer queues, but additionally support the Collection interface. So, for example, it is possible to remove an arbitrary element from a queue using remove(x) . However, such operations are in general not performed very efficiently, and are intended for only occasional use, such as when a queued message is cancelled.
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 .
A BlockingQueue does not intrinsically support any kind of «close» or «shutdown» operation to indicate that no more items will be added. The needs and usage of such features tend to be implementation-dependent. For example, a common tactic is for producers to insert special end-of-stream or poison objects, that are interpreted accordingly when taken by consumers.
Usage example, based on a typical producer-consumer scenario. Note that a BlockingQueue can safely be used with multiple producers and multiple consumers.
class Producer implements Runnable < private final BlockingQueue queue; Producer(BlockingQueue q) < queue = q; >public void run() < try < while (true) < queue.put(produce()); >> catch (InterruptedException ex) < . handle . >> Object produce() < . >> class Consumer implements Runnable < private final BlockingQueue queue; Consumer(BlockingQueue q) < queue = q; >public void run() < try < while (true) < consume(queue.take()); >> catch (InterruptedException ex) < . handle . >> void consume(Object x) < . >> class Setup < void main() < BlockingQueue q = new SomeQueueImplementation(); Producer p = new Producer(q); Consumer c1 = new Consumer(q); Consumer c2 = new Consumer(q); new Thread(p).start(); new Thread(c1).start(); new Thread(c2).start(); >> Memory consistency effects: As with other concurrent collections, actions in a thread prior to placing an object into a BlockingQueue happen-before actions subsequent to the access or removal of that element from the BlockingQueue in another thread.
This interface is a member of the Java Collections Framework.
Interface 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.
BlockingQueue methods come in four forms, with different ways of handling operations that cannot be satisfied immediately, but may be satisfied at some point in the future: one throws an exception, the second returns a special value (either null or false , depending on the operation), the third blocks the current thread indefinitely until the operation can succeed, and the fourth blocks for only a given maximum time limit before giving up. These methods are summarized in the following table:
| Throws exception | Special value | Blocks | Times out | |
|---|---|---|---|---|
| Insert | add(e) | offer(e) | put(e) | offer(e, time, unit) |
| Remove | remove() | poll() | take() | poll(time, unit) |
| Examine | element() | peek() | not applicable | not applicable |
A BlockingQueue does not accept null elements. Implementations throw NullPointerException on attempts to add , put or offer a null . A null is used as a sentinel value to indicate failure of poll operations.
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 designed to be used primarily for producer-consumer queues, but additionally support the Collection interface. So, for example, it is possible to remove an arbitrary element from a queue using remove(x) . However, such operations are in general not performed very efficiently, and are intended for only occasional use, such as when a queued message is cancelled.
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 .
A BlockingQueue does not intrinsically support any kind of «close» or «shutdown» operation to indicate that no more items will be added. The needs and usage of such features tend to be implementation-dependent. For example, a common tactic is for producers to insert special end-of-stream or poison objects, that are interpreted accordingly when taken by consumers.
Usage example, based on a typical producer-consumer scenario. Note that a BlockingQueue can safely be used with multiple producers and multiple consumers.
class Producer implements Runnable < private final BlockingQueue queue; Producer(BlockingQueue q) < queue = q; >public void run() < try < while (true) < queue.put(produce()); >> catch (InterruptedException ex) < . handle . >> Object produce() < . >> class Consumer implements Runnable < private final BlockingQueue queue; Consumer(BlockingQueue q) < queue = q; >public void run() < try < while (true) < consume(queue.take()); >> catch (InterruptedException ex) < . handle . >> void consume(Object x) < . >> class Setup < void main() < BlockingQueue q = new SomeQueueImplementation(); Producer p = new Producer(q); Consumer c1 = new Consumer(q); Consumer c2 = new Consumer(q); new Thread(p).start(); new Thread(c1).start(); new Thread(c2).start(); >> Memory consistency effects: As with other concurrent collections, actions in a thread prior to placing an object into a BlockingQueue happen-before actions subsequent to the access or removal of that element from the BlockingQueue in another thread.
This interface is a member of the Java Collections Framework.