The BlockingQueue interface in Java represents a thread-safe queue that supports operations that wait for the queue to become non-empty when retrieving elements, and wait for space to become available when storing elements.

Core BlockingQueue Implementations

1. ArrayBlockingQueue

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
public class ArrayBlockingQueueExample {
public static void main(String[] args) throws InterruptedException {
// Bounded queue with fixed capacity
BlockingQueue<String> queue = new ArrayBlockingQueue<>(3);
// Producer thread
Thread producer = new Thread(() -> {
try {
queue.put("Item 1");
System.out.println("Added: Item 1");
boolean added = queue.offer("Item 2", 2, TimeUnit.SECONDS);
System.out.println("Added Item 2: " + added);
queue.put("Item 3");
System.out.println("Added: Item 3");
// This will block until space is available
queue.put("Item 4");
System.out.println("Added: Item 4");
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
// Consumer thread
Thread consumer = new Thread(() -> {
try {
Thread.sleep(2000); // Let producer fill the queue
while (!queue.isEmpty()) {
String item = queue.take();
System.out.println("Consumed: " + item);
Thread.sleep(1000);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
producer.start();
consumer.start();
producer.join();
consumer.join();
}
}

2. LinkedBlockingQueue

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.BlockingQueue;
public class LinkedBlockingQueueExample {
public static void main(String[] args) {
// Optionally bounded queue
BlockingQueue<Integer> boundedQueue = new LinkedBlockingQueue<>(100);
BlockingQueue<Integer> unboundedQueue = new LinkedBlockingQueue<>(); // Essentially Integer.MAX_VALUE
// Producer-Consumer example
BlockingQueue<Order> orderQueue = new LinkedBlockingQueue<>();
// Multiple producers
for (int i = 0; i < 3; i++) {
new Thread(new OrderProducer(orderQueue, "Producer-" + i)).start();
}
// Multiple consumers
for (int i = 0; i < 2; i++) {
new Thread(new OrderConsumer(orderQueue, "Consumer-" + i)).start();
}
}
}
class Order {
private final String id;
private final String product;
public Order(String id, String product) {
this.id = id;
this.product = product;
}
@Override
public String toString() {
return "Order{id='" + id + "', product='" + product + "'}";
}
}
class OrderProducer implements Runnable {
private final BlockingQueue<Order> queue;
private final String name;
private int orderId = 1;
public OrderProducer(BlockingQueue<Order> queue, String name) {
this.queue = queue;
this.name = name;
}
@Override
public void run() {
try {
while (true) {
Order order = new Order(name + "-Order-" + orderId++, "Product-" + (orderId % 5));
queue.put(order);
System.out.println(name + " produced: " + order);
Thread.sleep(500 + (int)(Math.random() * 1000));
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
class OrderConsumer implements Runnable {
private final BlockingQueue<Order> queue;
private final String name;
public OrderConsumer(BlockingQueue<Order> queue, String name) {
this.queue = queue;
this.name = name;
}
@Override
public void run() {
try {
while (true) {
Order order = queue.take();
System.out.println(name + " consumed: " + order);
// Process order
Thread.sleep(1000 + (int)(Math.random() * 1500));
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}

3. PriorityBlockingQueue

import java.util.concurrent.PriorityBlockingQueue;
import java.util.Comparator;
public class PriorityBlockingQueueExample {
public static void main(String[] args) throws InterruptedException {
// Natural ordering
BlockingQueue<Integer> priorityQueue = new PriorityBlockingQueue<>();
// Custom ordering
BlockingQueue<Task> taskQueue = new PriorityBlockingQueue<>(
11, // initial capacity
Comparator.comparing(Task::getPriority).reversed() // Higher priority first
);
// Add tasks with different priorities
taskQueue.put(new Task("Low Priority Task", Task.Priority.LOW));
taskQueue.put(new Task("High Priority Task", Task.Priority.HIGH));
taskQueue.put(new Task("Medium Priority Task", Task.Priority.MEDIUM));
taskQueue.put(new Task("Urgent Task", Task.Priority.URGENT));
// Consumers will process tasks in priority order
for (int i = 0; i < 2; i++) {
new Thread(new TaskProcessor(taskQueue, "Processor-" + i)).start();
}
}
}
class Task {
enum Priority { LOW, MEDIUM, HIGH, URGENT }
private final String name;
private final Priority priority;
public Task(String name, Priority priority) {
this.name = name;
this.priority = priority;
}
public Priority getPriority() {
return priority;
}
@Override
public String toString() {
return String.format("Task{name='%s', priority=%s}", name, priority);
}
}
class TaskProcessor implements Runnable {
private final BlockingQueue<Task> queue;
private final String name;
public TaskProcessor(BlockingQueue<Task> queue, String name) {
this.queue = queue;
this.name = name;
}
@Override
public void run() {
try {
while (true) {
Task task = queue.take();
System.out.println(name + " processing: " + task);
Thread.sleep(1000); // Simulate processing
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}

4. SynchronousQueue

import java.util.concurrent.SynchronousQueue;
public class SynchronousQueueExample {
public static void main(String[] args) {
// Direct handoff between producers and consumers
SynchronousQueue<String> queue = new SynchronousQueue<>(true); // fair ordering
// Producer
new Thread(() -> {
try {
String[] items = {"Item1", "Item2", "Item3", "Item4"};
for (String item : items) {
System.out.println("Producer trying to put: " + item);
queue.put(item); // Blocks until consumer takes
System.out.println("Producer successfully put: " + item);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}).start();
// Consumer (slower)
new Thread(() -> {
try {
Thread.sleep(2000); // Start consuming after delay
while (true) {
String item = queue.take(); // Blocks until producer puts
System.out.println("Consumer took: " + item);
Thread.sleep(2000); // Simulate slow processing
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}).start();
}
}

5. DelayQueue

import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
public class DelayQueueExample {
public static void main(String[] args) throws InterruptedException {
DelayQueue<DelayedTask> delayQueue = new DelayQueue<>();
// Add tasks with different delays
long now = System.currentTimeMillis();
delayQueue.put(new DelayedTask("Task 1", now + 3000)); // 3 seconds
delayQueue.put(new DelayedTask("Task 2", now + 1000)); // 1 second
delayQueue.put(new DelayedTask("Task 3", now + 5000)); // 5 seconds
System.out.println("Tasks added to delay queue");
// Consumer that processes tasks when they become available
new Thread(() -> {
try {
while (!delayQueue.isEmpty()) {
DelayedTask task = delayQueue.take(); // Blocks until delay expires
System.out.println("Processed: " + task);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}).start();
}
}
class DelayedTask implements Delayed {
private final String name;
private final long startTime;
public DelayedTask(String name, long startTime) {
this.name = name;
this.startTime = startTime;
}
@Override
public long getDelay(TimeUnit unit) {
long diff = startTime - System.currentTimeMillis();
return unit.convert(diff, TimeUnit.MILLISECONDS);
}
@Override
public int compareTo(Delayed other) {
return Long.compare(this.startTime, ((DelayedTask) other).startTime);
}
@Override
public String toString() {
return "DelayedTask{name='" + name + "', startTime=" + startTime + "}";
}
}

Comparison and Usage Patterns

Performance Characteristics Comparison

import java.util.concurrent.*;
public class BlockingQueueComparison {
public static void main(String[] args) throws InterruptedException {
int capacity = 1000;
int elements = 10000;
// Test different implementations
testQueue(new ArrayBlockingQueue<>(capacity), "ArrayBlockingQueue", elements);
testQueue(new LinkedBlockingQueue<>(capacity), "LinkedBlockingQueue", elements);
testQueue(new PriorityBlockingQueue<>(capacity), "PriorityBlockingQueue", elements);
testQueue(new SynchronousQueue<>(), "SynchronousQueue", elements);
}
private static void testQueue(BlockingQueue<Integer> queue, String name, int elements) 
throws InterruptedException {
System.out.println("\nTesting " + name);
long startTime = System.currentTimeMillis();
// Producer thread
Thread producer = new Thread(() -> {
try {
for (int i = 0; i < elements; i++) {
queue.put(i);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
// Consumer thread
Thread consumer = new Thread(() -> {
try {
for (int i = 0; i < elements; i++) {
queue.take();
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
producer.start();
consumer.start();
producer.join();
consumer.join();
long endTime = System.currentTimeMillis();
System.out.println(name + " completed in " + (endTime - startTime) + "ms");
}
}

Producer-Consumer with Multiple Implementations

import java.util.concurrent.*;
import java.util.*;
public class MultiQueueProducerConsumer {
public static void main(String[] args) {
// Create different queue implementations
List<BlockingQueue<String>> queues = Arrays.asList(
new ArrayBlockingQueue<>(10),
new LinkedBlockingQueue<>(10),
new PriorityBlockingQueue<>(10)
);
// Create producers for each queue
List<Thread> producers = new ArrayList<>();
for (int i = 0; i < queues.size(); i++) {
BlockingQueue<String> queue = queues.get(i);
Thread producer = new Thread(new StringProducer(queue, "Producer-" + i));
producers.add(producer);
producer.start();
}
// Create consumers for each queue
List<Thread> consumers = new ArrayList<>();
for (int i = 0; i < queues.size(); i++) {
BlockingQueue<String> queue = queues.get(i);
Thread consumer = new Thread(new StringConsumer(queue, "Consumer-" + i));
consumers.add(consumer);
consumer.start();
}
// Let it run for a while
try {
Thread.sleep(10000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
// Interrupt all threads
producers.forEach(Thread::interrupt);
consumers.forEach(Thread::interrupt);
}
}
class StringProducer implements Runnable {
private final BlockingQueue<String> queue;
private final String name;
private int counter = 0;
public StringProducer(BlockingQueue<String> queue, String name) {
this.queue = queue;
this.name = name;
}
@Override
public void run() {
try {
while (!Thread.currentThread().isInterrupted()) {
String item = name + "-Item-" + (counter++);
if (queue.offer(item, 100, TimeUnit.MILLISECONDS)) {
System.out.println(name + " produced: " + item);
} else {
System.out.println(name + " failed to produce: " + item + " (queue full)");
}
Thread.sleep(200 + (int)(Math.random() * 300));
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
class StringConsumer implements Runnable {
private final BlockingQueue<String> queue;
private final String name;
public StringConsumer(BlockingQueue<String> queue, String name) {
this.queue = queue;
this.name = name;
}
@Override
public void run() {
try {
while (!Thread.currentThread().isInterrupted()) {
String item = queue.poll(500, TimeUnit.MILLISECONDS);
if (item != null) {
System.out.println(name + " consumed: " + item);
} else {
System.out.println(name + " found queue empty");
}
Thread.sleep(300 + (int)(Math.random() * 400));
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}

Advanced Usage: Batching with BlockingQueue

import java.util.concurrent.*;
import java.util.*;
public class BatchProcessor {
private final BlockingQueue<DataItem> queue;
private final int batchSize;
private final long timeoutMs;
private volatile boolean running = true;
public BatchProcessor(BlockingQueue<DataItem> queue, int batchSize, long timeoutMs) {
this.queue = queue;
this.batchSize = batchSize;
this.timeoutMs = timeoutMs;
}
public void startProcessing() {
Thread processor = new Thread(this::processBatches);
processor.start();
}
public void stopProcessing() {
running = false;
}
private void processBatches() {
List<DataItem> batch = new ArrayList<>(batchSize);
while (running || !queue.isEmpty()) {
try {
batch.clear();
// Wait for first element
DataItem firstItem = queue.poll(timeoutMs, TimeUnit.MILLISECONDS);
if (firstItem != null) {
batch.add(firstItem);
// Try to gather more items up to batchSize
while (batch.size() < batchSize && running) {
DataItem item = queue.poll(100, TimeUnit.MILLISECONDS);
if (item != null) {
batch.add(item);
} else {
break; // No more items immediately available
}
}
// Process the batch
processBatch(batch);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
break;
}
}
}
private void processBatch(List<DataItem> batch) {
System.out.println("Processing batch of " + batch.size() + " items: " + batch);
// Simulate processing time
try {
Thread.sleep(100);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
public void addItem(DataItem item) throws InterruptedException {
queue.put(item);
}
public static void main(String[] args) throws InterruptedException {
BlockingQueue<DataItem> queue = new LinkedBlockingQueue<>();
BatchProcessor processor = new BatchProcessor(queue, 5, 1000);
processor.startProcessing();
// Add items
for (int i = 0; i < 23; i++) {
processor.addItem(new DataItem("Item-" + i));
Thread.sleep(150); // Variable rate
}
Thread.sleep(2000); // Let processor finish
processor.stopProcessing();
}
}
class DataItem {
private final String data;
public DataItem(String data) {
this.data = data;
}
@Override
public String toString() {
return data;
}
}

Key Differences and When to Use Each

Best Practices

1. Choose the right implementation based on your ordering and bounding needs
2. Always handle InterruptedException properly
3. Use timeouts to avoid indefinite blocking
4. Consider capacity carefully for bounded queues
5. Monitor queue sizes in production applications
6. Use appropriate queue for the workload characteristics

BlockingQueue implementations are essential building blocks for concurrent applications in Java, providing robust thread-safe communication between producers and consumers.