Java BlockingQueue Example
January 11, 2023
BlockingQueue interface in our Java application.
1. The
BlockingQueue is a Queue with additional capability that are
a. The
BlockingQueue waits for the queue to become non-empty when retrieving an element.
b. The
BlockingQueue waits for space to become available in the queue when storing an element.
2. The
BlockingQueue methods can start its operation either without waiting or with waiting. It offers four types of methods.
a. Throws exception : These methods do not wait to start its operation, if they are unable to start immediately, they throw exception. The methods are
add(e), remove() and element().
b. Returns null or false value : These methods return null are false value immediately depending on the operation. These methods are
offer(e), poll() and peek().
c. Blocks : These methods block the current thread indefinitely until the operation can succeed. These methods are
put(e) and take().
d. Times out : These methods block the current thread only up to given time. The methods are
offer(e, time, unit) and poll(time, unit).
3. The
BlockingQueue implementation classes are
ArrayBlockingQueue
LinkedBlockingDeque
LinkedBlockingQueue
LinkedTransferQueue
PriorityBlockingQueue
SynchronousQueue
DelayQueue
4. The
BlockingQueue does not accept null elements. When we add, put or offer a null value, it throws NullPointerException.
5. The default size of a
BlockingQueue is Integer.MAX_VALUE. If we have constructed BlockingQueue with a specified size and if capacity is full, then no element can be put without blocking.
6. The
BlockingQueue implementations are thread safe. Their methods are atomic that are handled either by using locks or other forms of concurrency control. However, the bulk operations such as addAll, containsAll, retainAll and removeAll are not necessarily performed atomically unless specified otherwise in an implementation.
7. The
BlockingQueue are primarily used for producer-consumer queues.
Producer-Consumer Example
Find the producer-consumer example usingBlockingQueue and its implementation class ArrayBlockingQueue with capacity 1. Producer produces element and waits for consumer to consume it. Consumer consumes the element and waits for producer to produce it, and so on.
ProducerConsumer.java
package com.concretepage;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
class Producer implements Runnable {
private final BlockingQueue<String> queue;
Producer(BlockingQueue<String> q) {
queue = q;
}
public void run() {
try {
while (true) {
queue.put(produceData());
}
} catch (InterruptedException ex) {
System.out.println(ex);
}
}
int cnt = 1;
String produceData() {
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
String s = "A-" + cnt++;
System.out.println("Produced: " + s);
return s;
}
}
class Consumer implements Runnable {
private final BlockingQueue<String> queue;
Consumer(BlockingQueue<String> q) {
queue = q;
}
public void run() {
try {
while (true) {
consumeData(queue.take());
}
} catch (InterruptedException ex) {
System.out.println(ex);
}
}
void consumeData(String s) {
System.out.println("Consumed: " + s);
}
}
public class ProducerConsumer {
public static void main(String[] args) {
int capacity = 1;
BlockingQueue<String> q = new ArrayBlockingQueue<>(capacity);
Producer p = new Producer(q);
Consumer c = new Consumer(q);
new Thread(p).start();
new Thread(c).start();
}
}
Produced: A-1 Consumed: A-1 Produced: A-2 Consumed: A-2 Produced: A-3 Consumed: A-3 Produced: A-4 Consumed: A-4 ------
BlockingQueue Methods
TheBlockingQueue implements Collection, Iterable and Queue interfaces. Here we will discuss methods declared in BlockingQueue interface.
1. add
boolean add(E e)
If no space is currently available, it throws
IllegalStateException.
2. put
void put(E e) throws InterruptedException
3. Offer
boolean offer(E e)
With timeout:
boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException
4. take
E take() throws InterruptedException
5. poll
E poll(long timeout, TimeUnit unit) throws InterruptedException
6. remove
boolean remove(Object o)
7. drainTo
int drainTo(Collection<? super E> c)
With max elements:
int drainTo(Collection<? super E> c, int maxElements)
8. contains
boolean contains(Object o)
9. remainingCapacity
int remainingCapacity()
