BlockingQueue & Producer-Consumer Pattern

// Bounded queue — blocks when full
BlockingQueue<Order> queue = new ArrayBlockingQueue<>(1000);

// Fair mode — FIFO ordering for waiting threads (slower but prevents starvation)
BlockingQueue<Order> fairQueue = new ArrayBlockingQueue<>(1000, true);

// Unbounded (dangerous in production!)
BlockingQueue<Task> unbounded = new LinkedBlockingQueue<>();

// Bounded — preferred for production
BlockingQueue<Task> bounded = new LinkedBlockingQueue<>(5000);

// Natural ordering (Comparable)
BlockingQueue<PriorityTask> pq = new PriorityBlockingQueue<>();

// Custom comparator — higher priority first
BlockingQueue<PriorityTask> pq = new PriorityBlockingQueue<>(100,
    Comparator.comparingInt(PriorityTask::getPriority).reversed()
);

class PriorityTask implements Comparable<PriorityTask> {
    private final int priority;
    private final String payload;

    public PriorityTask(int priority, String payload) {
        this.priority = priority;
        this.payload = payload;
    }

    @Override
    public int compareTo(PriorityTask other) {
        return Integer.compare(this.priority, other.priority); // lower = higher priority
    }
}

BlockingQueue<Runnable> handoff = new SynchronousQueue<>();

// This is exactly what Executors.newCachedThreadPool() uses!
ExecutorService cached = new ThreadPoolExecutor(
    0, Integer.MAX_VALUE,
    60L, TimeUnit.SECONDS,
    new SynchronousQueue<>()  // zero buffering — task goes directly to a thread
);

class ScheduledTask implements Delayed {
    private final String name;
    private final long executeAt; // absolute time in millis

    public ScheduledTask(String name, long delayMillis) {
        this.name = name;
        this.executeAt = System.currentTimeMillis() + delayMillis;
    }

    @Override
    public long getDelay(TimeUnit unit) {
        long remaining = executeAt - System.currentTimeMillis();
        return unit.convert(remaining, TimeUnit.MILLISECONDS);
    }

    @Override
    public int compareTo(Delayed other) {
        return Long.compare(this.getDelay(TimeUnit.MILLISECONDS),
                           other.getDelay(TimeUnit.MILLISECONDS));
    }
}

DelayQueue<ScheduledTask> delayQueue = new DelayQueue<>();
delayQueue.put(new ScheduledTask("Send reminder email", 5000));  // execute after 5s
delayQueue.put(new ScheduledTask("Expire session", 30000));       // execute after 30s

// Consumer — blocks until an element's delay expires
ScheduledTask task = delayQueue.take(); // waits for earliest expiry

TransferQueue<Event> queue = new LinkedTransferQueue<>();

// Regular put — non-blocking (enqueues like LinkedBlockingQueue)
queue.put(event);

// Transfer — blocks until a consumer takes this specific element
queue.transfer(event); // producer waits for consumer

// Try transfer — if a consumer is waiting, hand off directly; else return false
boolean immediate = queue.tryTransfer(event);

// Try transfer with timeout
boolean handed = queue.tryTransfer(event, 100, TimeUnit.MILLISECONDS);

import java.util.concurrent.*;

public class ProducerConsumerDemo {
    private static final BlockingQueue<String> queue = new ArrayBlockingQueue<>(10);
    private static final String POISON_PILL = "DONE";

    public static void main(String[] args) {
        Thread producer = new Thread(() -> {
            try {
                for (int i = 0; i < 50; i++) {
                    String item = "Order-" + i;
                    queue.put(item); // blocks if queue is full (backpressure!)
                    System.out.println("Produced: " + item);
                }
                queue.put(POISON_PILL); // signal termination
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        Thread consumer = new Thread(() -> {
            try {
                while (true) {
                    String item = queue.take(); // blocks if queue is empty
                    if (POISON_PILL.equals(item)) break;
                    process(item);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        producer.start();
        consumer.start();
    }

    private static void process(String item) {
        System.out.println("Consumed: " + item);
    }
}

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class MultiProducerConsumer {
    private static final int QUEUE_CAPACITY = 1000;
    private static final int NUM_PRODUCERS = 3;
    private static final int NUM_CONSUMERS = 5;
    private static final String POISON_PILL = "__STOP__";

    private static final BlockingQueue<String> queue =
        new ArrayBlockingQueue<>(QUEUE_CAPACITY);

    public static void main(String[] args) throws InterruptedException {
        ExecutorService producerPool = Executors.newFixedThreadPool(NUM_PRODUCERS);
        ExecutorService consumerPool = Executors.newFixedThreadPool(NUM_CONSUMERS);
        AtomicInteger producedCount = new AtomicInteger(0);

        // Start producers
        for (int p = 0; p < NUM_PRODUCERS; p++) {
            final int producerId = p;
            producerPool.submit(() -> {
                try {
                    for (int i = 0; i < 100; i++) {
                        String item = "P" + producerId + "-Item" + i;
                        queue.put(item);
                        producedCount.incrementAndGet();
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
        }

        // Start consumers
        for (int c = 0; c < NUM_CONSUMERS; c++) {
            final int consumerId = c;
            consumerPool.submit(() -> {
                try {
                    while (true) {
                        String item = queue.take();
                        if (POISON_PILL.equals(item)) {
                            queue.put(POISON_PILL); // propagate to other consumers
                            break;
                        }
                        processItem(consumerId, item);
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
        }

        // Shutdown producers, then signal consumers
        producerPool.shutdown();
        producerPool.awaitTermination(1, TimeUnit.MINUTES);

        // All producers done — send poison pill
        queue.put(POISON_PILL);

        consumerPool.shutdown();
        consumerPool.awaitTermination(1, TimeUnit.MINUTES);

        System.out.println("Total produced: " + producedCount.get());
    }

    private static void processItem(int consumerId, String item) {
        // Simulate work
        System.out.printf("Consumer-%d processed: %s%n", consumerId, item);
    }
}

// Production backpressure pattern with metrics
public class BackpressureProducer {
    private final BlockingQueue<Event> queue;
    private final MeterRegistry metrics;

    public void produce(Event event) {
        boolean accepted = queue.offer(event, 500, TimeUnit.MILLISECONDS);
        if (!accepted) {
            metrics.counter("queue.rejected").increment();
            // Fallback: write to dead-letter queue, retry later, or drop
            deadLetterQueue.add(event);
        } else {
            metrics.gauge("queue.size", queue.size());
        }
    }
}

// Fixed thread pool uses LinkedBlockingQueue (unbounded — dangerous!)
ExecutorService fixed = Executors.newFixedThreadPool(10);
// Internally: new LinkedBlockingQueue<>() — can OOM under load

// Recommended: explicit bounded queue with rejection policy
ExecutorService safe = new ThreadPoolExecutor(
    10, 20,                              // core/max threads
    60L, TimeUnit.SECONDS,               // idle timeout
    new ArrayBlockingQueue<>(500),       // bounded work queue
    new ThreadPoolExecutor.CallerRunsPolicy() // backpressure
);

// Multi-stage pipeline: each stage is a producer-consumer
BlockingQueue<RawEvent> ingest = new LinkedBlockingQueue<>(10000);
BlockingQueue<EnrichedEvent> enriched = new ArrayBlockingQueue<>(5000);
BlockingQueue<ValidatedEvent> validated = new ArrayBlockingQueue<>(5000);

// Stage 1: Ingest → Enrich
executor.submit(() -> {
    while (!Thread.currentThread().isInterrupted()) {
        RawEvent raw = ingest.take();
        EnrichedEvent e = enrichService.enrich(raw);
        enriched.put(e);
    }
});

// Stage 2: Enrich → Validate
executor.submit(() -> {
    while (!Thread.currentThread().isInterrupted()) {
        EnrichedEvent e = enriched.take();
        ValidatedEvent v = validator.validate(e);
        validated.put(v);
    }
});

// Stage 3: Validate → Persist
executor.submit(() -> {
    while (!Thread.currentThread().isInterrupted()) {
        ValidatedEvent v = validated.take();
        repository.save(v);
    }
});

public class TokenBucketRateLimiter {
    private final DelayQueue<Token> tokens = new DelayQueue<>();
    private final int maxTokens;
    private final long refillIntervalMs;

    public TokenBucketRateLimiter(int maxTokens, long refillIntervalMs) {
        this.maxTokens = maxTokens;
        this.refillIntervalMs = refillIntervalMs;
        // Pre-fill tokens
        for (int i = 0; i < maxTokens; i++) {
            tokens.put(new Token(0)); // immediately available
        }
    }

    public boolean tryAcquire(long timeoutMs) throws InterruptedException {
        Token token = tokens.poll(timeoutMs, TimeUnit.MILLISECONDS);
        if (token != null) {
            // Schedule a replacement token after refill interval
            tokens.put(new Token(refillIntervalMs));
            return true;
        }
        return false; // rate limited
    }

    private static class Token implements Delayed {
        private final long availableAt;

        Token(long delayMs) {
            this.availableAt = System.currentTimeMillis() + delayMs;
        }

        @Override
        public long getDelay(TimeUnit unit) {
            return unit.convert(availableAt - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        }

        @Override
        public int compareTo(Delayed o) {
            return Long.compare(this.getDelay(TimeUnit.MILLISECONDS),
                               o.getDelay(TimeUnit.MILLISECONDS));
        }
    }
}

class StablePriorityTask implements Comparable<StablePriorityTask> {
    private static final AtomicLong SEQ = new AtomicLong(0);
    private final int priority;
    private final long sequenceNumber = SEQ.incrementAndGet();

    @Override
    public int compareTo(StablePriorityTask other) {
        int cmp = Integer.compare(this.priority, other.priority);
        if (cmp != 0) return cmp;
        return Long.compare(this.sequenceNumber, other.sequenceNumber); // FIFO tiebreaker
    }
}