How to Implement Multiple Blocking Queues with a Single Consumer in Java?

Question

How can I manage multiple blocking queues with a single consumer in a Java application?

import java.util.concurrent.*;

public class MultipleBlockingQueues {
    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<Integer> queue1 = new LinkedBlockingQueue<>();
        BlockingQueue<Integer> queue2 = new LinkedBlockingQueue<>();

        // Producer for queue1
        new Thread(() -> {
            for (int i = 0; i < 10; i++) {
                try {
                    queue1.put(i);
                    System.out.println("Produced " + i + " in Queue1");
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        }).start();

        // Producer for queue2
        new Thread(() -> {
            for (int i = 10; i < 20; i++) {
                try {
                    queue2.put(i);
                    System.out.println("Produced " + i + " in Queue2");
                    Thread.sleep(150);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        }).start();

        // Consumer
        new Thread(() -> {
            try {
                while (true) {
                    Integer item = null;
                    // Poll items from queues
                    if (!queue1.isEmpty()) {
                        item = queue1.take();
                    } else if (!queue2.isEmpty()) {
                        item = queue2.take();
                    }
                    if (item != null) {
                        System.out.println("Consumed " + item);
                    }
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }).start();
    }
}

Answer

Managing multiple blocking queues with a single consumer can enhance performance and lead to efficient resource utilization in a multi-threaded Java application. The process involves using blocking queues to store producer-generated data while ensuring that a single consumer effectively processes items from the queues in a coordinated manner.

BlockingQueue<Integer> queue = new LinkedBlockingQueue<>();
try {
    Integer element = queue.take();  // Waits for an element to become available.
} catch (InterruptedException e) {
    Thread.currentThread().interrupt();
}

Causes

Thread safety concerns when modifying shared data structures.
Inefficient task scheduling due to improper queue management.
Potential blocking if the consumer cannot keep up with the producers.

Solutions

Use a priority mechanism to choose which queue to consume from first.
Implement timeouts for the consumer when trying to pull from empty queues.
Utilize multiple consumers if the workload from the queues becomes too high.

Common Mistakes

Mistake: Assuming all queues will have items at all times, leading to resource contention.

Solution: Implement periodic checks or timeouts when attempting to poll items from the queues to prevent blocking.

Mistake: Not handling InterruptedException properly, which could lead to thread leaks.

Solution: Always handle Interruptions by checking the interrupt flag and breaking out of the loop gracefully.

Helpers

Java blocking queue
multiple queues single consumer
Java concurrency
BlockingQueue example
Java executor services