Implementing an Empty Queue Using SynchronousQueue

SynchronousQueue is a specialized blocking queue provided in the Java concurrency package (java.util.concurrent). Unlike a standard BlockingQueue, its most defining characteristic is that it has no internal capacity. Every put operation must wait for a corresponding take operation, and vice versa. You can think of it as a zero-buffer conduit where producers and consumers must pair up directly on a one-to-one basis to transfer data.

• Zero Capacity: The size() method always returns 0, and isEmpty() always evaluates to true.
• Synchronous Handoff: The put method blocks until another thread invokes take, and the take method blocks until another thread invokes put.
• Dual Operating Modes:
  • Fair Mode (FIFO): Instantiated via new SynchronousQueue<>(true), ensuring threads queue up in the order of their requests.
  • Unfair Mode (Default): Instantiated via new SynchronousQueue<>(), utilizing a competitive, greedy handoff strategy that yields higher throughput.

Why Use SynchronousQueue to Implement an "Empty Queue"

In certain architectural scenarios, we require a queue where data is only permitted to flow when a producer and a consumer are directly paired. Examples include:

• Thread Pool Hand-off Mechanism: When ThreadPoolExecutor utilizes a SynchronousQueue as its work queue, any incoming task forces the creation of a new thread if all existing worker threads are busy (provided the maximum pool size has not been reached); otherwise, it is handed off directly.
• Task Schedulers: Tasks should only be submitted when an idle worker thread is immediately available, preventing task accumulation and mitigating memory pressure.
• Producer-Consumer Synchronization Points: Ensuring that every produced element is handled instantaneously by a consumer, avoiding any residual data lingering in memory.

An "empty queue" realized through SynchronousQueue satisfies these requirements naturally because it does not store any elements at all.

Basic Usage Example

The following code illustrates the most common usage pattern of SynchronousQueue, covering both fair and unfair modes.

import java.util.concurrent.*;

public class SynchronousQueueDemo {

    public static void main(String[] args) throws InterruptedException {
        // Unfair mode (default)
        BlockingQueue<String> queue = new SynchronousQueue<>();

        // Start the consumer thread
        Thread consumer = new Thread(() -> {
            try {
                System.out.println("Consumer is ready to take an element...");
                String value = queue.take(); // Blocks until a producer arrives
                System.out.println("Consumer received: " + value);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
        consumer.start();

        // Main thread acts as the producer
        TimeUnit.SECONDS.sleep(1); // Simulate preparation time
        System.out.println("Producer is ready to put an element...");
        queue.put("Hello SynchronousQueue"); // Returns immediately upon pairing with the consumer
        System.out.println("Producer has successfully put the element");

        consumer.join();
    }
}

Execution Output:

Consumer is ready to take an element...
Producer is ready to put an element...
Consumer received: Hello SynchronousQueue
Producer has successfully put the element

As observed:

1. The consumer thread blocks at take() until the producer invokes put.
2. The put call also blocks until the consumer finishes take, guaranteeing synchronous delivery.

Fair Mode Example

BlockingQueue<Integer> fairQueue = new SynchronousQueue<>(true);

In fair mode, when multiple producers and consumers compete concurrently, the SynchronousQueue pairs them up in strict FIFO order, preventing thread starvation.

Common Patterns for Using SynchronousQueue in Production

1. Thread Pool Hand-off Design

ExecutorService executor = new ThreadPoolExecutor(
        0,                     // corePoolSize
        10,                    // maximumPoolSize
        60L, TimeUnit.SECONDS, // keepAliveTime
        new SynchronousQueue<>()); // Work queue

• When all worker threads are busy, incoming tasks do not enter a queue; instead, they attempt a direct hand-off to an idle thread, should one exist.
• If the thread limit has been reached and no idle threads are available, the ThreadPoolExecutor triggers its RejectedExecutionHandler (with AbortPolicy throwing an exception by default).

2. Implementing a Task Synchronization Point

Suppose a batch of tasks must be executed strictly in pairs: every produced task requires an immediate worker thread to process it, otherwise the task must wait.

    private final SynchronousQueue<Runnable> handoff = new SynchronousQueue<>();

    public SyncTaskProcessor(int workerCount) {
        for (int i = 0; i < workerCount; i++) {
            new Thread(this::workerLoop, "worker-" + i).start();
        }
    }

    private void workerLoop() {
        try {
            while (true) {
                Runnable task = handoff.take(); // Wait for task delivery
                task.run();
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    public void submit(Runnable task) throws InterruptedException {
        handoff.put(task); // Deliver directly to an idle worker
    }
}

• The submit method only returns when an idle worker thread becomes available, ensuring that tasks are never queued internally.
• Ideal for scenarios demanding ultra-low latency under constrained system resources.

3. Zero-Buffer Producer-Consumer Streams

In streaming data pipelines (such as real-time log ingestion), utilizing a SynchronousQueue prevents data from piling up in memory.

class LogCollector {
    private final SynchronousQueue<String> queue = new SynchronousQueue<>();

    public void start() {
        // Consumer thread: writes to disk or network sink
        new Thread(() -> {
            try {
                while (true) {
                    String log = queue.take(); // Write immediately
                    writeToSink(log);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }, "log-writer").start();
    }

    public void log(String message) throws InterruptedException {
        queue.put(message); // Producer must wait until the write completes
    }

    private void writeToSink(String log) {
        // Actual write implementation (File, Kafka, HTTP, etc.)
    }
}

• When the consumer's write speed falls behind the log generation rate, the log method blocks, naturally establishing backpressure and preventing Out-Of-Memory (OOM) errors.

Pitfalls and Best Practices

Summary

• SynchronousQueue is a zero-capacity, synchronously delivered blocking queue that naturally embodies the concept of an "empty queue."
• Through its strict put and take pairing mechanism, it delivers zero-buffering, backpressure, and strict synchronization across producer-consumer pipelines, thread pool hand-offs, and real-time stream processing.
• When adopting it, developers should select the appropriate constructor based on business requirements for fairness, throughput, and resource utilization, while maintaining rigorous exception handling and thread lifecycle management.

Mastering the mechanics of SynchronousQueue allows you to design highly reliable, resource-friendly concurrency models for high-throughput systems that demand strict, synchronous delivery.