[{"_1":2,"_3":-5,"_4":-5},"loaderData",{"_5":6},"actionData","errors","routes/blog-tag",{"_7":8,"_9":10},"tag",{"_31":95,"_73":96,"_35":97},"articles",{"_11":12,"_13":14,"_15":16,"_17":16,"_18":19,"_20":21,"_22":23,"_24":23,"_25":16,"_26":27},"content",[30],"pageable",{"_28":21,"_29":19},"totalElements",1,"totalPages","size",10,"number",0,"first",true,"last","numberOfElements","empty",false,"page_number","page_size",{"_31":32,"_33":34,"_35":36,"_37":38,"_11":39,"_40":-5,"_41":23,"_42":27,"_43":44,"_45":46,"_47":48,"_49":50,"_51":52,"_53":52,"_54":55,"_56":57,"_58":59,"_60":52,"_61":62,"_63":64,"_65":66,"_67":68,"_69":70,"_71":72},"id",117,"title","Blocking Queues","slug","java-collections-part-21-blocking-queues","excerpt","Foundation of concurrent applications: BlockingQueue family, producer-consumer patterns, backpressure handling, and thread pool internal mechanisms.","# Java Collections Compendium Part 21: Blocking Queues & Producer-Consumer\n\nMaster the foundation of concurrent Java applications. Learn BlockingQueue implementations for producer-consumer patterns, backpressure handling, and thread coordination without explicit synchronization.\n\n## 📋 At a Glance\n\n| Aspect | Details |\n|--------|---------|\n| **Topic** | BlockingQueue, ArrayBlockingQueue, LinkedBlockingQueue, DelayQueue |\n| **Complexity** | Advanced |\n| **Prerequisites** | Part 16 (Queue/Deque), Part 18 (ConcurrentHashMap) |\n| **Time to Master** | 4-5 hours |\n| **Interview Frequency** | Very High (producer-consumer, thread pools) |\n\n## 🎯 What You'll Learn\n\nAfter completing this article, you will be able to:\n\n1. Implement producer-consumer patterns with BlockingQueue\n2. Choose the right BlockingQueue implementation for your use case\n3. Handle backpressure with bounded queues\n4. Use DelayQueue for scheduled task execution\n5. Understand how thread pools use BlockingQueues internally\n\n### Interactive Visualizer\n\nSee the producer-consumer pattern in action - watch how producers block when queue is full and consumers block when empty:\n\n```visualizer-blockingqueue\ncontrols: true\n```\n\n---\n\n## Production Story: The Memory Leak That Wasn't\n\n### The Incident\n\nOur message processing service was running out of memory. Heap dumps showed millions of pending messages, but our \"unbounded\" queue was supposed to handle any load:\n\n```java\n// MessageProcessor.java - The problematic code\npublic class MessageProcessor {\n private final Queue queue = new ConcurrentLinkedQueue<>();\n private final ExecutorService executor = Executors.newFixedThreadPool(10);\n\n public void submit(Message message) {\n queue.offer(message); // Never blocks, never rejects!\n executor.submit(this::processNext);\n }\n\n private void processNext() {\n Message msg = queue.poll();\n if (msg != null) {\n process(msg); // Takes 100ms average\n }\n }\n}\n```\n\n### The Problem\n\n```text\nMessage arrival rate: 1000/second\nProcessing rate: 10 threads × 10/second = 100/second\n\nQueue growth per second: 1000 - 100 = 900 messages\nAfter 1 hour: 3,240,000 messages in queue!\nMemory: ~3.2 GB just for messages\n\nTimeline:\n├─ 0 min: Queue = 0, memory OK\n├─ 10 min: Queue = 540,000, memory warning\n├─ 30 min: Queue = 1,620,000, GC struggling\n├─ 45 min: OutOfMemoryError!\n```\n\n### The BlockingQueue Solution\n\n```java\n// MessageProcessor.java - The backpressure-aware solution\npublic class MessageProcessor {\n // Bounded queue provides backpressure\n private final BlockingQueue queue = new ArrayBlockingQueue<>(10000);\n private final ExecutorService executor = Executors.newFixedThreadPool(10);\n\n public MessageProcessor() {\n // Start consumer threads\n for (int i = 0; i < 10; i++) {\n executor.submit(this::consumerLoop);\n }\n }\n\n public boolean submit(Message message) {\n // Option 1: Block until space available\n try {\n queue.put(message); // Blocks if queue full\n return true;\n } catch (InterruptedException e) {\n Thread.currentThread().interrupt();\n return false;\n }\n\n // Option 2: Reject if full (fail fast)\n // return queue.offer(message);\n\n // Option 3: Wait with timeout\n // return queue.offer(message, 1, TimeUnit.SECONDS);\n }\n\n private void consumerLoop() {\n while (!Thread.currentThread().isInterrupted()) {\n try {\n Message msg = queue.take(); // Blocks if empty\n process(msg);\n } catch (InterruptedException e) {\n Thread.currentThread().interrupt();\n break;\n }\n }\n }\n}\n```\n\n### The Difference\n\n```text\nWith ArrayBlockingQueue(10000):\n\nProducer too fast:\n├─ Queue fills to 10,000\n├─ put() BLOCKS producer\n├─ Producer slows down to consumer rate\n├─ Memory stays bounded!\n\nBackpressure propagates:\n├─ HTTP server → MessageProcessor.submit() blocks\n├─ HTTP thread pool exhausted\n├─ Server returns 503 Service Unavailable\n├─ Load balancer routes to other instances\n├─ System stays healthy!\n```\n\n---\n\n## Mental Model: The Factory Assembly Line\n\n```text\n┌─────────────────────────────────────────────────────────────────────┐\n│ FACTORY ASSEMBLY LINE │\n├─────────────────────────────────────────────────────────────────────┤\n│ │\n│ ConcurrentLinkedQueue = Infinite Conveyor Belt │\n│ ┌─────────────────────────────────────────────────────────────┐ │\n│ │ Producer → [___][___][___][___][___]... → Consumer │ │\n│ │ Items pile up infinitely │ │\n│ │ No signal to slow down │ │\n│ │ Eventually: CRASH! 💥 │ │\n│ └─────────────────────────────────────────────────────────────┘ │\n│ │\n│ BlockingQueue = Limited Conveyor Belt │\n│ ┌─────────────────────────────────────────────────────────────┐ │\n│ │ Producer → [■][■][■][■][■] → Consumer │ │\n│ │ ↑ ↑ │ │\n│ │ │ │ │ │\n│ │ put() blocks take() blocks │ │\n│ │ when FULL when EMPTY │ │\n│ │ │ │\n│ │ Producer: \"Belt is full, I'll wait\" │ │\n│ │ Consumer: \"Belt is empty, I'll wait\" │ │\n│ │ Natural flow control! │ │\n│ └─────────────────────────────────────────────────────────────┘ │\n│ │\n│ SynchronousQueue = Direct Handoff (no belt) │\n│ ┌─────────────────────────────────────────────────────────────┐ │\n│ │ Producer ─────🤝───── Consumer │ │\n│ │ Direct handoff │ │\n│ │ Both must be present │ │\n│ │ Zero capacity queue │ │\n│ └─────────────────────────────────────────────────────────────┘ │\n│ │\n└─────────────────────────────────────────────────────────────────────┘\n```\n\n---\n\n## Deep Dive: BlockingQueue Interface\n\n### Method Categories\n\n```java\npublic interface BlockingQueue extends Queue {\n // ═══════════════════════════════════════════════════════════════\n // BLOCKING OPERATIONS (wait indefinitely)\n // ═══════════════════════════════════════════════════════════════\n void put(E e) throws InterruptedException;\n // Blocks until space available\n\n E take() throws InterruptedException;\n // Blocks until element available\n\n // ═══════════════════════════════════════════════════════════════\n // TIMED OPERATIONS (wait with timeout)\n // ═══════════════════════════════════════════════════════════════\n boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException;\n // Returns false if timeout expires\n\n E poll(long timeout, TimeUnit unit) throws InterruptedException;\n // Returns null if timeout expires\n\n // ═══════════════════════════════════════════════════════════════\n // NON-BLOCKING OPERATIONS (return immediately)\n // ═══════════════════════════════════════════════════════════════\n boolean offer(E e); // Returns false if full\n E poll(); // Returns null if empty\n E peek(); // Returns null if empty\n\n // ═══════════════════════════════════════════════════════════════\n // THROWING OPERATIONS (exception if impossible)\n // ═══════════════════════════════════════════════════════════════\n boolean add(E e); // Throws IllegalStateException if full\n E remove(); // Throws NoSuchElementException if empty\n E element(); // Throws NoSuchElementException if empty\n\n // ═══════════════════════════════════════════════════════════════\n // BULK & UTILITY\n // ═══════════════════════════════════════════════════════════════\n int remainingCapacity(); // Approximate space left\n int drainTo(Collection c); // Remove all and add to collection\n int drainTo(Collection c, int maxElements);\n}\n```\n\n### Operation Summary Table\n\n```text\n┌─────────────┬─────────────┬─────────────┬─────────────┬─────────────┐\n│ Operation │ Throws │ Special │ Blocks │ Times Out │\n├─────────────┼─────────────┼─────────────┼─────────────┼─────────────┤\n│ Insert │ add(e) │ offer(e) │ put(e) │ offer(e,t,u)│\n├─────────────┼─────────────┼─────────────┼─────────────┼─────────────┤\n│ Remove │ remove() │ poll() │ take() │ poll(t,u) │\n├─────────────┼─────────────┼─────────────┼─────────────┼─────────────┤\n│ Examine │ element() │ peek() │ N/A │ N/A │\n└─────────────┴─────────────┴─────────────┴─────────────┴─────────────┘\n```\n\n---\n\n## Deep Dive: ArrayBlockingQueue\n\n### Structure and Properties\n\n```java\n// Bounded, array-based, FIFO queue\nArrayBlockingQueue queue = new ArrayBlockingQueue<>(100);\n\n// Key properties:\n// - Fixed capacity (set at construction)\n// - FIFO ordering\n// - Single lock for all operations\n// - Optional fairness (FIFO thread ordering)\n```\n\n### Internal Implementation\n\n```java\npublic class ArrayBlockingQueue extends AbstractQueue\n implements BlockingQueue {\n\n final Object[] items; // Circular buffer\n int takeIndex; // Next position for take\n int putIndex; // Next position for put\n int count; // Number of elements\n\n final ReentrantLock lock; // Single lock for all operations\n private final Condition notEmpty; // Signals when not empty\n private final Condition notFull; // Signals when not full\n\n public void put(E e) throws InterruptedException {\n Objects.requireNonNull(e);\n final ReentrantLock lock = this.lock;\n lock.lockInterruptibly();\n try {\n while (count == items.length) {\n notFull.await(); // Wait until not full\n }\n enqueue(e);\n } finally {\n lock.unlock();\n }\n }\n\n public E take() throws InterruptedException {\n final ReentrantLock lock = this.lock;\n lock.lockInterruptibly();\n try {\n while (count == 0) {\n notEmpty.await(); // Wait until not empty\n }\n return dequeue();\n } finally {\n lock.unlock();\n }\n }\n\n private void enqueue(E x) {\n items[putIndex] = x;\n if (++putIndex == items.length) putIndex = 0; // Wrap around\n count++;\n notEmpty.signal(); // Wake up waiting consumers\n }\n\n private E dequeue() {\n E x = (E) items[takeIndex];\n items[takeIndex] = null;\n if (++takeIndex == items.length) takeIndex = 0; // Wrap around\n count--;\n notFull.signal(); // Wake up waiting producers\n return x;\n }\n}\n```\n\n### Fairness Option\n\n```java\n// Non-fair (default): Better throughput\nArrayBlockingQueue nonFair = new ArrayBlockingQueue<>(100);\n\n// Fair: FIFO ordering for waiting threads\nArrayBlockingQueue fair = new ArrayBlockingQueue<>(100, true);\n\n// Fair mode:\n// - Threads acquire lock in arrival order\n// - Prevents starvation\n// - Lower throughput (more overhead)\n// - Use when fairness matters more than speed\n```\n\n---\n\n## Deep Dive: LinkedBlockingQueue\n\n### Structure and Properties\n\n```java\n// Optionally bounded, linked-node based, FIFO queue\nLinkedBlockingQueue unbounded = new LinkedBlockingQueue<>(); // Integer.MAX_VALUE\nLinkedBlockingQueue bounded = new LinkedBlockingQueue<>(1000);\n\n// Key properties:\n// - Optional capacity (unbounded by default)\n// - Two separate locks (put and take)\n// - Better concurrency than ArrayBlockingQueue\n// - Slightly higher per-element overhead\n```\n\n### Two-Lock Design\n\n```java\npublic class LinkedBlockingQueue extends AbstractQueue\n implements BlockingQueue {\n\n private final int capacity;\n private final AtomicInteger count = new AtomicInteger();\n\n // Two separate locks!\n private final ReentrantLock takeLock = new ReentrantLock();\n private final Condition notEmpty = takeLock.newCondition();\n\n private final ReentrantLock putLock = new ReentrantLock();\n private final Condition notFull = putLock.newCondition();\n\n // Producers and consumers can operate simultaneously!\n // Only need coordination at boundary conditions (empty/full)\n}\n```\n\n### ArrayBlockingQueue vs LinkedBlockingQueue\n\n```text\n┌─────────────────────────────────────────────────────────────────────┐\n│ ArrayBlockingQueue vs LinkedBlockingQueue │\n├───────────────────────┬─────────────────────┬───────────────────────┤\n│ Aspect │ ArrayBlockingQueue │ LinkedBlockingQueue │\n├───────────────────────┼─────────────────────┼───────────────────────┤\n│ Capacity │ Fixed (required) │ Optional (unbounded) │\n│ Locking │ Single lock │ Two locks (put/take) │\n│ Concurrency │ Lower │ Higher │\n│ Memory │ Fixed array │ Per-node allocation │\n│ GC impact │ Minimal │ Node churn │\n│ Cache locality │ Better │ Worse │\n│ Fairness option │ Yes │ No │\n└───────────────────────┴─────────────────────┴───────────────────────┘\n\nUse ArrayBlockingQueue when:\n- Fixed capacity required\n- Memory predictability important\n- Fairness needed\n\nUse LinkedBlockingQueue when:\n- Higher throughput needed\n- Capacity flexibility desired\n- Put/take rate very different\n```\n\n---\n\n## Deep Dive: Specialized BlockingQueues\n\n### PriorityBlockingQueue\n\n```java\n// Unbounded, priority-ordered blocking queue\nPriorityBlockingQueue taskQueue = new PriorityBlockingQueue<>();\n\n// Or with comparator\nPriorityBlockingQueue prioritized = new PriorityBlockingQueue<>(\n 100, // Initial capacity\n Comparator.comparingInt(Task::getPriority).reversed()\n);\n\n// Properties:\n// - Unbounded (put never blocks!)\n// - Ordering by Comparable or Comparator\n// - No FIFO guarantee for equal-priority elements\n// - Backed by binary heap\n\n// Usage pattern: Priority task processing\nwhile (!shutdown) {\n Task task = taskQueue.take(); // Blocks if empty, returns highest priority\n process(task);\n}\n```\n\n### DelayQueue\n\n```java\n// Unbounded queue where elements become available after delay\nDelayQueue delayQueue = new DelayQueue<>();\n\nclass DelayedTask implements Delayed {\n private final long executeAt;\n private final Runnable task;\n\n public DelayedTask(Runnable task, long delayMs) {\n this.task = task;\n this.executeAt = System.currentTimeMillis() + delayMs;\n }\n\n @Override\n public long getDelay(TimeUnit unit) {\n long remaining = executeAt - System.currentTimeMillis();\n return unit.convert(remaining, TimeUnit.MILLISECONDS);\n }\n\n @Override\n public int compareTo(Delayed other) {\n return Long.compare(\n this.getDelay(TimeUnit.MILLISECONDS),\n other.getDelay(TimeUnit.MILLISECONDS)\n );\n }\n}\n\n// Usage: Scheduled execution\ndelayQueue.put(new DelayedTask(() -> sendReminder(), 3600_000)); // 1 hour delay\ndelayQueue.put(new DelayedTask(() -> cleanup(), 60_000)); // 1 minute delay\n\n// Consumer waits for element to become ready\nwhile (!shutdown) {\n DelayedTask task = delayQueue.take(); // Blocks until delay expires\n task.execute();\n}\n```\n\n### SynchronousQueue\n\n```java\n// Zero-capacity queue - direct handoff between threads\nSynchronousQueue handoff = new SynchronousQueue<>();\n\n// Producer blocks until consumer takes\n// Consumer blocks until producer puts\n// No buffering!\n\n// Use case: Thread pool handoff\nExecutorService cachedPool = new ThreadPoolExecutor(\n 0, Integer.MAX_VALUE,\n 60L, TimeUnit.SECONDS,\n new SynchronousQueue<>() // Tasks handed directly to threads\n);\n\n// Fair mode available\nSynchronousQueue fair = new SynchronousQueue<>(true);\n```\n\n### LinkedTransferQueue\n\n```java\n// Combination of ConcurrentLinkedQueue and SynchronousQueue\nLinkedTransferQueue queue = new LinkedTransferQueue<>();\n\n// Standard offer (non-blocking, always succeeds)\nqueue.offer(\"item\");\n\n// Transfer (blocks until consumer takes THIS specific item)\nqueue.transfer(\"urgent\"); // Blocks until someone takes \"urgent\"\n\n// tryTransfer (non-blocking attempt at direct handoff)\nboolean transferred = queue.tryTransfer(\"item\");\n\n// tryTransfer with timeout\nboolean done = queue.tryTransfer(\"item\", 1, TimeUnit.SECONDS);\n\n// Use case: Prefer direct handoff, fall back to queuing\nif (!queue.tryTransfer(item)) {\n queue.put(item); // Queue if no immediate consumer\n}\n```\n\n---\n\n## Deep Dive: Producer-Consumer Patterns\n\n### Basic Pattern\n\n```java\npublic class ProducerConsumer {\n private final BlockingQueue queue;\n private final int numConsumers;\n private final ExecutorService executor;\n private volatile boolean running = true;\n\n public ProducerConsumer(int capacity, int numConsumers) {\n this.queue = new ArrayBlockingQueue<>(capacity);\n this.numConsumers = numConsumers;\n this.executor = Executors.newFixedThreadPool(numConsumers);\n startConsumers();\n }\n\n private void startConsumers() {\n for (int i = 0; i < numConsumers; i++) {\n executor.submit(this::consumerLoop);\n }\n }\n\n private void consumerLoop() {\n while (running || !queue.isEmpty()) {\n try {\n T item = queue.poll(100, TimeUnit.MILLISECONDS);\n if (item != null) {\n process(item);\n }\n } catch (InterruptedException e) {\n Thread.currentThread().interrupt();\n break;\n }\n }\n }\n\n public void produce(T item) throws InterruptedException {\n queue.put(item);\n }\n\n public void shutdown() {\n running = false;\n executor.shutdown();\n }\n\n protected void process(T item) {\n // Override to implement processing\n }\n}\n```\n\n### Poison Pill Pattern\n\n```java\npublic class PoisonPillExample {\n private final BlockingQueue queue = new LinkedBlockingQueue<>();\n private final T poisonPill;\n\n public PoisonPillExample(T poisonPill) {\n this.poisonPill = poisonPill;\n }\n\n // Consumer checks for poison pill\n public void consumerLoop() {\n while (true) {\n try {\n T item = queue.take();\n if (item == poisonPill) {\n // Put it back for other consumers, then exit\n queue.put(poisonPill);\n break;\n }\n process(item);\n } catch (InterruptedException e) {\n Thread.currentThread().interrupt();\n break;\n }\n }\n }\n\n // Producer sends poison pill to signal shutdown\n public void shutdown() throws InterruptedException {\n queue.put(poisonPill);\n }\n}\n```\n\n### Batch Processing Pattern\n\n```java\npublic class BatchProcessor {\n private final BlockingQueue queue;\n private final int batchSize;\n private final long maxWaitMs;\n\n public void processorLoop() {\n List batch = new ArrayList<>(batchSize);\n\n while (!Thread.currentThread().isInterrupted()) {\n try {\n // Wait for first item\n T first = queue.poll(maxWaitMs, TimeUnit.MILLISECONDS);\n if (first == null) continue;\n\n batch.add(first);\n\n // Drain more items if available (non-blocking)\n queue.drainTo(batch, batchSize - 1);\n\n processBatch(batch);\n batch.clear();\n } catch (InterruptedException e) {\n Thread.currentThread().interrupt();\n break;\n }\n }\n }\n\n private void processBatch(List batch) {\n // Process all items in batch\n }\n}\n```\n\n---\n\n## Deep Dive: Thread Pool Integration\n\n### How Executors Use BlockingQueues\n\n```java\n// Fixed thread pool: LinkedBlockingQueue (unbounded)\nExecutorService fixed = Executors.newFixedThreadPool(10);\n// Actually: new ThreadPoolExecutor(10, 10, 0L, MILLISECONDS,\n// new LinkedBlockingQueue<>())\n\n// Cached thread pool: SynchronousQueue (direct handoff)\nExecutorService cached = Executors.newCachedThreadPool();\n// Actually: new ThreadPoolExecutor(0, MAX_VALUE, 60L, SECONDS,\n// new SynchronousQueue<>())\n\n// Custom pool with bounded queue\nExecutorService bounded = new ThreadPoolExecutor(\n 5, // Core pool size\n 20, // Maximum pool size\n 60L, TimeUnit.SECONDS, // Keep-alive time\n new ArrayBlockingQueue<>(1000), // Bounded work queue\n new ThreadPoolExecutor.CallerRunsPolicy() // Rejection policy\n);\n```\n\n### Rejection Policies\n\n```java\n// When queue is full AND max threads reached:\n\n// AbortPolicy (default): Throw RejectedExecutionException\nnew ThreadPoolExecutor.AbortPolicy()\n\n// CallerRunsPolicy: Run task in caller's thread (backpressure)\nnew ThreadPoolExecutor.CallerRunsPolicy()\n\n// DiscardPolicy: Silently discard task\nnew ThreadPoolExecutor.DiscardPolicy()\n\n// DiscardOldestPolicy: Discard oldest queued task\nnew ThreadPoolExecutor.DiscardOldestPolicy()\n\n// Custom policy\nRejectedExecutionHandler custom = (runnable, executor) -> {\n // Log, retry, save to disk, etc.\n logger.warn(\"Task rejected: {}\", runnable);\n};\n```\n\n---\n\n## ⚠️ Common Mistakes\n\n### Mistake 1: Using Unbounded Queue with Fixed Resources\n\n```java\n// ❌ WRONG: Unbounded queue can cause OOM\nExecutorService executor = new ThreadPoolExecutor(\n 10, 10, 0L, TimeUnit.MILLISECONDS,\n new LinkedBlockingQueue<>() // Unbounded!\n);\n\n// If producers outpace consumers, queue grows forever\nfor (int i = 0; i < 1_000_000; i++) {\n executor.submit(slowTask); // Queue fills up → OOM\n}\n\n// ✅ CORRECT: Use bounded queue with rejection policy\nExecutorService executor = new ThreadPoolExecutor(\n 10, 10, 0L, TimeUnit.MILLISECONDS,\n new ArrayBlockingQueue<>(1000),\n new ThreadPoolExecutor.CallerRunsPolicy()\n);\n```\n\n### Mistake 2: Not Handling InterruptedException\n\n```java\n// ❌ WRONG: Swallowing InterruptedException\nwhile (true) {\n try {\n Item item = queue.take();\n process(item);\n } catch (InterruptedException e) {\n // Silently ignored - thread won't stop!\n }\n}\n\n// ✅ CORRECT: Restore interrupt status and exit\nwhile (!Thread.currentThread().isInterrupted()) {\n try {\n Item item = queue.take();\n process(item);\n } catch (InterruptedException e) {\n Thread.currentThread().interrupt(); // Restore flag\n break; // Exit loop\n }\n}\n```\n\n### Mistake 3: Blocking in Consumer Without Timeout\n\n```java\n// ❌ WRONG: take() blocks forever, can't check shutdown flag\nwhile (running) {\n Item item = queue.take(); // Blocks even after running=false!\n process(item);\n}\n\n// ✅ CORRECT: Use poll with timeout\nwhile (running || !queue.isEmpty()) {\n try {\n Item item = queue.poll(100, TimeUnit.MILLISECONDS);\n if (item != null) {\n process(item);\n }\n } catch (InterruptedException e) {\n Thread.currentThread().interrupt();\n break;\n }\n}\n```\n\n### Mistake 4: Wrong Queue for Use Case\n\n```java\n// ❌ WRONG: PriorityBlockingQueue for FIFO\n// Elements with same priority may not maintain order\nPriorityBlockingQueue queue = new PriorityBlockingQueue<>();\n\n// ❌ WRONG: SynchronousQueue when buffering needed\n// Producers block until consumer available\nSynchronousQueue queue = new SynchronousQueue<>();\nfor (Message msg : batch) {\n queue.put(msg); // Each put blocks until taken!\n}\n\n// ✅ CORRECT: Match queue to requirements\n// FIFO with backpressure: ArrayBlockingQueue\n// High throughput: LinkedBlockingQueue\n// Priority ordering: PriorityBlockingQueue\n// Direct handoff: SynchronousQueue\n```\n\n### Mistake 5: Forgetting remainingCapacity() is Approximate\n\n```java\n// ❌ WRONG: Checking capacity then adding\nif (queue.remainingCapacity() > 0) {\n queue.offer(item); // Might still fail!\n}\n\n// ✅ CORRECT: Use the operation's return value\nif (!queue.offer(item)) {\n handleQueueFull(item);\n}\n\n// Or use blocking with timeout\nif (!queue.offer(item, 1, TimeUnit.SECONDS)) {\n handleTimeout(item);\n}\n```\n\n---\n\n## 🐛 Debug This\n\n### Challenge 1: The Stuck Consumers\n\n```java\nBlockingQueue queue = new ArrayBlockingQueue<>(10);\nExecutorService executor = Executors.newFixedThreadPool(5);\nAtomicBoolean running = new AtomicBoolean(true);\n\n// Start consumers\nfor (int i = 0; i < 5; i++) {\n executor.submit(() -> {\n while (running.get()) {\n try {\n String item = queue.take();\n process(item);\n } catch (InterruptedException e) {\n // Handle interrupt\n }\n }\n });\n}\n\n// Shutdown\nrunning.set(false);\nexecutor.shutdown();\nexecutor.awaitTermination(10, TimeUnit.SECONDS);\n// Why doesn't it terminate?\n```\n\n**✅ Answer:**\n\nConsumers are stuck in `queue.take()` blocking call. Setting `running=false` doesn't wake them up.\n\n**Fix:**\n```java\n// Option 1: Use poll with timeout\nString item = queue.poll(100, TimeUnit.MILLISECONDS);\n\n// Option 2: Interrupt threads\nexecutor.shutdownNow(); // Sends interrupt to all threads\n\n// Option 3: Poison pill\nqueue.put(POISON); // Each consumer exits when it gets this\n```\n\n### Challenge 2: The Lost Messages\n\n```java\nBlockingQueue queue = new LinkedBlockingQueue<>(100);\n\n// Producer\nfor (int i = 0; i < 1000; i++) {\n queue.offer(new Message(i)); // What's wrong?\n}\n\n// Consumer\nwhile (true) {\n Message msg = queue.take();\n process(msg);\n}\n```\n\n**✅ Answer:**\n\n`offer()` returns `false` when queue is full, but we ignore it. 900 messages are silently lost!\n\n**Fix:**\n```java\n// Use put() to block when full\nqueue.put(new Message(i));\n\n// Or handle rejection\nif (!queue.offer(message)) {\n handleRejection(message);\n}\n```\n\n### Challenge 3: The Priority Confusion\n\n```java\nPriorityBlockingQueue queue = new PriorityBlockingQueue<>();\n\nqueue.offer(new Task(\"A\", 1));\nqueue.offer(new Task(\"B\", 1));\nqueue.offer(new Task(\"C\", 1));\n\n// All same priority\nwhile (!queue.isEmpty()) {\n System.out.println(queue.poll().getName());\n}\n// What order?\n```\n\n**✅ Answer:**\n\nOrder is **unpredictable** for equal-priority elements! PriorityBlockingQueue doesn't guarantee FIFO for same priority.\n\n**Fix for FIFO within priority:**\n```java\nclass Task implements Comparable {\n private static final AtomicLong seq = new AtomicLong();\n private final long sequenceNumber = seq.getAndIncrement();\n private final int priority;\n\n public int compareTo(Task other) {\n int cmp = Integer.compare(this.priority, other.priority);\n if (cmp != 0) return cmp;\n return Long.compare(this.sequenceNumber, other.sequenceNumber);\n }\n}\n```\n\n---\n\n## 💻 Exercises\n\n### Exercise 1: Rate-Limited Queue\n\nImplement a queue that limits consumption rate:\n\n```java\npublic class RateLimitedQueue {\n public RateLimitedQueue(int capacity, int maxPerSecond);\n public void put(T item) throws InterruptedException;\n public T take() throws InterruptedException; // Rate limited\n}\n```\n\n**✅ Solution:**\n\n```java\npublic class RateLimitedQueue {\n private final BlockingQueue queue;\n private final RateLimiter rateLimiter;\n\n public RateLimitedQueue(int capacity, int maxPerSecond) {\n this.queue = new ArrayBlockingQueue<>(capacity);\n this.rateLimiter = RateLimiter.create(maxPerSecond);\n }\n\n public void put(T item) throws InterruptedException {\n queue.put(item);\n }\n\n public T take() throws InterruptedException {\n rateLimiter.acquire(); // Wait for permit\n return queue.take();\n }\n\n // Using Guava's RateLimiter, or implement simple version:\n static class RateLimiter {\n private final long intervalNanos;\n private long nextAllowedNanos = 0;\n\n static RateLimiter create(double permitsPerSecond) {\n return new RateLimiter((long) (1e9 / permitsPerSecond));\n }\n\n RateLimiter(long intervalNanos) {\n this.intervalNanos = intervalNanos;\n }\n\n synchronized void acquire() throws InterruptedException {\n long now = System.nanoTime();\n if (nextAllowedNanos > now) {\n Thread.sleep((nextAllowedNanos - now) / 1_000_000);\n }\n nextAllowedNanos = Math.max(now, nextAllowedNanos) + intervalNanos;\n }\n }\n}\n```\n\n### Exercise 2: Work Stealing Queue\n\nImplement basic work stealing between consumer threads:\n\n```java\npublic class WorkStealingPool {\n public WorkStealingPool(int numWorkers);\n public void submit(T work);\n public void shutdown();\n}\n```\n\n**✅ Solution:**\n\n```java\npublic class WorkStealingPool {\n private final Deque[] queues;\n private final Thread[] workers;\n private volatile boolean running = true;\n\n @SuppressWarnings(\"unchecked\")\n public WorkStealingPool(int numWorkers, Consumer processor) {\n queues = new LinkedBlockingDeque[numWorkers];\n workers = new Thread[numWorkers];\n\n for (int i = 0; i < numWorkers; i++) {\n queues[i] = new LinkedBlockingDeque<>();\n final int workerId = i;\n workers[i] = new Thread(() -> workerLoop(workerId, processor));\n workers[i].start();\n }\n }\n\n public void submit(T work) {\n // Round-robin assignment\n int idx = ThreadLocalRandom.current().nextInt(queues.length);\n queues[idx].addLast(work);\n }\n\n private void workerLoop(int id, Consumer processor) {\n Deque myQueue = queues[id];\n\n while (running) {\n T work = myQueue.pollFirst();\n\n if (work == null) {\n // Try to steal from others\n work = trySteal(id);\n }\n\n if (work != null) {\n processor.accept(work);\n } else {\n Thread.yield(); // Nothing to do\n }\n }\n }\n\n private T trySteal(int myId) {\n for (int i = 0; i < queues.length; i++) {\n if (i != myId) {\n T stolen = queues[i].pollLast(); // Steal from back\n if (stolen != null) return stolen;\n }\n }\n return null;\n }\n\n public void shutdown() {\n running = false;\n for (Thread worker : workers) {\n worker.interrupt();\n }\n }\n}\n```\n\n### Exercise 3: Timeout Queue\n\nImplement a queue where items expire if not consumed in time:\n\n```java\npublic class TimeoutQueue {\n public TimeoutQueue(Duration itemTimeout);\n public void put(T item);\n public T poll(); // Returns null if empty or all items expired\n public int size(); // Only non-expired items\n}\n```\n\n**✅ Solution:**\n\n```java\npublic class TimeoutQueue {\n private final BlockingQueue> queue = new LinkedBlockingQueue<>();\n private final long timeoutMs;\n\n public TimeoutQueue(Duration itemTimeout) {\n this.timeoutMs = itemTimeout.toMillis();\n }\n\n public void put(T item) {\n queue.offer(new TimestampedItem<>(item, System.currentTimeMillis()));\n }\n\n public T poll() {\n while (true) {\n TimestampedItem item = queue.poll();\n if (item == null) return null;\n\n if (!item.isExpired(timeoutMs)) {\n return item.value;\n }\n // Item expired, try next\n }\n }\n\n public int size() {\n long now = System.currentTimeMillis();\n return (int) queue.stream()\n .filter(item -> !item.isExpired(timeoutMs, now))\n .count();\n }\n\n private static class TimestampedItem {\n final T value;\n final long timestamp;\n\n TimestampedItem(T value, long timestamp) {\n this.value = value;\n this.timestamp = timestamp;\n }\n\n boolean isExpired(long timeoutMs) {\n return isExpired(timeoutMs, System.currentTimeMillis());\n }\n\n boolean isExpired(long timeoutMs, long now) {\n return now - timestamp > timeoutMs;\n }\n }\n}\n```\n\n---\n\n## 🎤 Senior-Level Interview Questions\n\n### Question 1: put() vs offer()\n\n**Q: When would you use put() vs offer() vs offer(timeout)?**\n\n**A:**\n\n| Method | Behavior | Use When |\n|--------|----------|----------|\n| `put()` | Blocks indefinitely | Must not lose data, can wait |\n| `offer()` | Returns false immediately | Can't afford to wait, handle rejection |\n| `offer(timeout)` | Blocks up to timeout | Balance between waiting and responsiveness |\n\n```java\n// put(): Critical data, producer can wait\nqueue.put(criticalEvent); // Will wait forever if needed\n\n// offer(): High-throughput, handle rejection\nif (!queue.offer(logEntry)) {\n writeToOverflow(logEntry);\n}\n\n// offer(timeout): Balanced approach\nif (!queue.offer(request, 100, TimeUnit.MILLISECONDS)) {\n return new Response(503, \"Service busy\");\n}\n```\n\n### Question 2: ArrayBlockingQueue vs LinkedBlockingQueue\n\n**Q: Why does LinkedBlockingQueue often have higher throughput than ArrayBlockingQueue?**\n\n**A:**\n\nLinkedBlockingQueue uses **two separate locks**:\n- `putLock` for producers\n- `takeLock` for consumers\n\n```java\n// ArrayBlockingQueue: Single lock\n// Producer and consumer compete for same lock\nput() → lock → enqueue → unlock\ntake() → lock → dequeue → unlock\n\n// LinkedBlockingQueue: Two locks\n// Producer and consumer can work simultaneously\nput() → putLock → enqueue → putUnlock → signal\ntake() → takeLock → dequeue → takeUnlock → signal\n\n// Result: Producer and consumer don't block each other\n// Higher concurrency = higher throughput\n```\n\n### Question 3: SynchronousQueue Use Cases\n\n**Q: When would you use SynchronousQueue over other blocking queues?**\n\n**A:**\n\nSynchronousQueue for direct handoff scenarios:\n\n1. **Thread pool (cached)**: No queueing, spawn thread immediately\n2. **Real-time processing**: Data must be processed NOW\n3. **Backpressure needed**: Producer waits for consumer\n\n```java\n// Cached thread pool uses SynchronousQueue\nExecutorService cached = new ThreadPoolExecutor(\n 0, Integer.MAX_VALUE, 60L, SECONDS,\n new SynchronousQueue<>() // No buffering\n);\n// Task immediately gets a thread or producer blocks\n\n// When NOT to use:\n// - Bursty producers (would block frequently)\n// - Slow consumers (producer throughput limited)\n// - Need buffering for resilience\n```\n\n### Question 4: DelayQueue Implementation\n\n**Q: How does DelayQueue know when an element is ready?**\n\n**A:**\n\nDelayQueue uses `PriorityQueue` internally, ordered by delay:\n\n```java\n// Elements must implement Delayed\ninterface Delayed extends Comparable {\n long getDelay(TimeUnit unit);\n}\n\n// take() implementation:\npublic E take() throws InterruptedException {\n lock.lockInterruptibly();\n try {\n for (;;) {\n E first = pq.peek();\n if (first == null) {\n available.await(); // Wait for any element\n } else {\n long delay = first.getDelay(NANOSECONDS);\n if (delay <= 0) {\n return pq.poll(); // Ready!\n }\n // Wait for delay to expire\n available.awaitNanos(delay);\n }\n }\n } finally {\n lock.unlock();\n }\n}\n```\n\n### Question 5: drainTo() Benefits\n\n**Q: What are the benefits of drainTo() over repeated poll()?**\n\n**A:**\n\n`drainTo()` is more efficient:\n\n1. **Single lock acquisition**: Locks once, drains all\n2. **Bulk operation**: Atomic from queue's perspective\n3. **Reduced contention**: Less lock traffic\n\n```java\n// Inefficient: Lock per poll\nList batch = new ArrayList<>();\nItem item;\nwhile ((item = queue.poll()) != null) { // Lock for each poll!\n batch.add(item);\n}\n\n// Efficient: Single lock for all\nList batch = new ArrayList<>();\nqueue.drainTo(batch); // One lock, drains all available\n\n// Even better: With limit\nqueue.drainTo(batch, 100); // Drain up to 100\n```\n\n---\n\n## 📝 Summary & Key Takeaways\n\n### BlockingQueue Methods\n- **Blocking**: `put()`, `take()` - wait indefinitely\n- **Timed**: `offer(timeout)`, `poll(timeout)` - wait with limit\n- **Non-blocking**: `offer()`, `poll()` - return immediately\n- **Throwing**: `add()`, `remove()` - exception on failure\n\n### Implementation Choices\n- **ArrayBlockingQueue**: Fixed capacity, single lock, fairness option\n- **LinkedBlockingQueue**: Optional capacity, two locks, higher throughput\n- **PriorityBlockingQueue**: Unbounded, priority ordered\n- **DelayQueue**: Time-delayed elements\n- **SynchronousQueue**: Direct handoff, zero capacity\n\n### Producer-Consumer Patterns\n- Use bounded queues for backpressure\n- Handle InterruptedException properly\n- Consider poison pill for graceful shutdown\n- Match queue type to requirements\n\n### Thread Pool Integration\n- Unbounded queue → OOM risk\n- Bounded queue → need rejection policy\n- CallerRunsPolicy provides natural backpressure\n\n---\n\n## 🏁 Conclusion\n\nBlockingQueues are the backbone of concurrent Java applications. They provide thread coordination, backpressure, and clean separation between producers and consumers. Understanding their internals helps you choose the right implementation and avoid subtle concurrency bugs.\n\nKey takeaways:\n1. **Use bounded queues** for backpressure and memory safety\n2. **Choose implementation based on requirements** - throughput, fairness, priority\n3. **Handle InterruptedException correctly** - restore flag and exit\n4. **Thread pools depend on queue choice** - understand the implications\n5. **drainTo() for efficient batching** - single lock acquisition\n\nIn the next article, we'll explore Collections and Streams integration - how to efficiently convert between collections and leverage the Stream API for powerful data transformations.\n\n---\n\n## 📅 Review Schedule\n\nTo solidify your understanding, review this material:\n- **Tomorrow**: Implement basic producer-consumer\n- **In 3 days**: Compare ArrayBlockingQueue vs LinkedBlockingQueue\n- **In 1 week**: Practice graceful shutdown patterns\n- **In 2 weeks**: Review all BlockingQueue implementations\n\n---\n\n**Next in Series**: [Part 22: Immutable Collections](/articles/java-collections-part-22-immutable-collections)\n\n**Series Navigation**: [Part 0: How to Use This Series](/articles/java-collections-part-0-how-to-use-this-series) | [Full Index](/articles/java-collections-compendium)\n","featuredImage","isPublished","isFeatured","subscriptionType","BASIC","publishedAt","2026-06-10T07:19:26.112485Z","createdAt","2026-06-10T07:19:26.113281Z","updatedAt","2026-06-16T21:26:56.086200Z","readingTimeMinutes",19,"estimatedReadingTime","authorId",3,"authorUsername","tomasz_main","categoryName","Java","readingTime","seriesId",5,"seriesOrder",21,"author",{"_31":55,"_100":57,"_101":102,"_103":104,"_105":-5,"_106":-5,"_107":-5,"_108":-5,"_109":-5},"category",{"_31":62,"_73":59,"_35":91,"_76":98,"_99":-5},"tags",[79,80,81,82,83],"series",{"_31":62,"_73":74,"_35":75,"_76":77,"_78":23},"name","Java Collections Compendium: From Fundamentals to Production Mastery","java-collections-compendium","description","A comprehensive 30-part series covering everything about Java Collections Framework. From foundational concepts like generics and equals/hashCode contract, through deep dives into ArrayList, LinkedList, HashMap, TreeMap, ConcurrentHashMap, to modern features like immutable collections, Java 21 sequenced collections, streams & collectors, and production patterns with JPA, Jackson & Spring. Includes real production stories, mental models, exercises, and 50+ interview questions.","isActive",{"_31":95,"_73":96,"_35":97},{"_31":92,"_73":93,"_35":94},{"_31":90,"_73":59,"_35":91},{"_31":87,"_73":88,"_35":89},{"_31":84,"_73":85,"_35":86},287,"Producer-consumer","producer-consumer",359,"Linkedblockingqueue","linkedblockingqueue",14,"java",286,"Blockingqueue","blockingqueue",360,"Arrayblockingqueue","arrayblockingqueue","Category for Java","colorHex","username","firstName","Tomasz","lastName","Jarosz","profilePicture","bio","website","githubUrl","linkedinUrl"]