Concurrency Patterns & Advanced Threading

import java.util.concurrent.*;

public class ProducerConsumerDemo {

    public static void main(String[] args) {
        BlockingQueue<String> queue = new ArrayBlockingQueue<>(10);

        // Producer
        Thread producer = new Thread(() -> {
            try {
                for (int i = 0; i < 20; i++) {
                    String item = "Order-" + i;
                    queue.put(item); // blocks if queue is full
                    System.out.println("Produced: " + item);
                }
                queue.put("POISON_PILL"); // signal to stop
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        // Consumer
        Thread consumer = new Thread(() -> {
            try {
                while (true) {
                    String item = queue.take(); // blocks if queue is empty
                    if ("POISON_PILL".equals(item)) break;
                    System.out.println("Consumed: " + item);
                    Thread.sleep(100); // simulate processing
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

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

import java.util.concurrent.locks.*;
import java.util.*;

public class ThreadSafeCache<K, V> {

    private final Map<K, V> cache = new HashMap<>();
    private final ReadWriteLock lock = new ReentrantReadWriteLock();
    private final Lock readLock = lock.readLock();
    private final Lock writeLock = lock.writeLock();

    public V get(K key) {
        readLock.lock(); // multiple threads can hold this simultaneously
        try {
            return cache.get(key);
        } finally {
            readLock.unlock();
        }
    }

    public void put(K key, V value) {
        writeLock.lock(); // exclusive -- no readers or writers allowed
        try {
            cache.put(key, value);
        } finally {
            writeLock.unlock();
        }
    }

    public List<K> allKeys() {
        readLock.lock();
        try {
            return new ArrayList<>(cache.keySet());
        } finally {
            readLock.unlock();
        }
    }
}

import java.util.concurrent.locks.StampedLock;

public class Point {
    private double x, y;
    private final StampedLock lock = new StampedLock();

    public void move(double deltaX, double deltaY) {
        long stamp = lock.writeLock();
        try {
            x += deltaX;
            y += deltaY;
        } finally {
            lock.unlockWrite(stamp);
        }
    }

    public double distanceFromOrigin() {
        // Optimistic read -- no locking overhead
        long stamp = lock.tryOptimisticRead();
        double currentX = x, currentY = y;

        // Validate that no write occurred during our read
        if (!lock.validate(stamp)) {
            // Fallback to full read lock
            stamp = lock.readLock();
            try {
                currentX = x;
                currentY = y;
            } finally {
                lock.unlockRead(stamp);
            }
        }
        return Math.sqrt(currentX * currentX + currentY * currentY);
    }
}

import java.util.concurrent.CountDownLatch;

public class ServiceStartup {

    public static void main(String[] args) throws InterruptedException {
        int serviceCount = 3;
        CountDownLatch latch = new CountDownLatch(serviceCount);

        for (String service : new String[]{"Database", "Cache", "MessageQueue"}) {
            new Thread(() -> {
                try {
                    System.out.println(service + " starting...");
                    Thread.sleep((long) (Math.random() * 2000));
                    System.out.println(service + " ready!");
                    latch.countDown();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }).start();
        }

        latch.await(); // blocks until count reaches 0
        System.out.println("All services ready -- starting application!");
    }
}

import java.util.concurrent.CyclicBarrier;

public class ParallelMatrixComputation {

    public static void main(String[] args) {
        int workers = 4;
        CyclicBarrier barrier = new CyclicBarrier(workers, () -> {
            // Runs after all threads arrive at the barrier
            System.out.println("--- Phase complete, merging results ---");
        });

        for (int i = 0; i < workers; i++) {
            final int workerId = i;
            new Thread(() -> {
                try {
                    for (int phase = 1; phase <= 3; phase++) {
                        System.out.println("Worker " + workerId + " computing phase " + phase);
                        Thread.sleep((long) (Math.random() * 1000));
                        barrier.await(); // wait for all workers
                    }
                } catch (Exception e) {
                    Thread.currentThread().interrupt();
                }
            }).start();
        }
    }
}

import java.util.concurrent.Phaser;

public class DynamicPhaserDemo {

    public static void main(String[] args) {
        Phaser phaser = new Phaser(1); // register self (main thread)

        for (int i = 0; i < 3; i++) {
            phaser.register(); // dynamically add participant
            final int id = i;
            new Thread(() -> {
                System.out.println("Worker " + id + " phase 0");
                phaser.arriveAndAwaitAdvance();

                System.out.println("Worker " + id + " phase 1");
                phaser.arriveAndDeregister(); // leave after phase 1
            }).start();
        }

        phaser.arriveAndAwaitAdvance(); // main waits for phase 0
        phaser.arriveAndDeregister();   // main deregisters
    }
}

public class DatabaseConnectionPool {
    private final Semaphore semaphore;
    private final BlockingQueue<Connection> pool;

    public DatabaseConnectionPool(int maxConnections) {
        this.semaphore = new Semaphore(maxConnections, true); // fair
        this.pool = new ArrayBlockingQueue<>(maxConnections);
        for (int i = 0; i < maxConnections; i++) {
            pool.offer(createConnection());
        }
    }

    public Connection acquire() throws InterruptedException {
        semaphore.acquire();  // blocks if all connections in use
        return pool.poll();
    }

    public Connection acquire(long timeout, TimeUnit unit) throws InterruptedException {
        if (semaphore.tryAcquire(timeout, unit)) {
            return pool.poll();
        }
        throw new TimeoutException("No connection available within " + timeout + " " + unit);
    }

    public void release(Connection conn) {
        pool.offer(conn);
        semaphore.release();  // return permit — unblocks waiting thread
    }

    public int availableConnections() {
        return semaphore.availablePermits();
    }
}

Semaphore mutex = new Semaphore(1);

mutex.acquire();
try {
    // critical section — only one thread at a time
} finally {
    mutex.release();
}

public class ApiRateLimiter {
    private final Semaphore semaphore;
    private final ScheduledExecutorService scheduler;

    public ApiRateLimiter(int requestsPerSecond) {
        this.semaphore = new Semaphore(requestsPerSecond);
        this.scheduler = Executors.newSingleThreadScheduledExecutor(r -> {
            Thread t = new Thread(r, "rate-limiter-refiller");
            t.setDaemon(true);
            return t;
        });

        // Refill permits every second
        scheduler.scheduleAtFixedRate(() -> {
            int deficit = requestsPerSecond - semaphore.availablePermits();
            if (deficit > 0) semaphore.release(deficit);
        }, 1, 1, TimeUnit.SECONDS);
    }

    public boolean tryRequest() {
        return semaphore.tryAcquire();
    }

    public void request() throws InterruptedException {
        semaphore.acquire();
    }

    public boolean request(long timeout, TimeUnit unit) throws InterruptedException {
        return semaphore.tryAcquire(timeout, unit);
    }
}

// Only allow 5 concurrent file uploads
Semaphore uploadThrottle = new Semaphore(5);

public CompletableFuture<Void> uploadFile(Path file) {
    return CompletableFuture.runAsync(() -> {
        try {
            uploadThrottle.acquire();
            try {
                performUpload(file);  // only 5 at a time
            } finally {
                uploadThrottle.release();
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }, uploadExecutor);
}

// Large queries consume 3 permits, small queries consume 1
Semaphore resourceSemaphore = new Semaphore(10);

public ResultSet executeLargeQuery(String sql) throws InterruptedException {
    resourceSemaphore.acquire(3);  // acquire 3 permits
    try {
        return database.execute(sql);
    } finally {
        resourceSemaphore.release(3);
    }
}

public ResultSet executeSmallQuery(String sql) throws InterruptedException {
    resourceSemaphore.acquire(1);
    try {
        return database.execute(sql);
    } finally {
        resourceSemaphore.release(1);
    }
}

// UNFAIR (default) — higher throughput, possible starvation
Semaphore unfair = new Semaphore(5);

// FAIR — guarantees FIFO ordering, lower throughput
Semaphore fair = new Semaphore(5, true);

public class DoubleBufferPipeline {
    private final Exchanger<List<LogEvent>> exchanger = new Exchanger<>();
    private final int batchSize;

    public DoubleBufferPipeline(int batchSize) {
        this.batchSize = batchSize;
    }

    // Producer: collects events, swaps when batch is full
    public void startProducer() {
        new Thread(() -> {
            List<LogEvent> buffer = new ArrayList<>(batchSize);
            try {
                while (!Thread.currentThread().isInterrupted()) {
                    LogEvent event = pollNextEvent();
                    buffer.add(event);
                    if (buffer.size() >= batchSize) {
                        // Swap full buffer for an empty one
                        buffer = exchanger.exchange(buffer);
                        buffer.clear();
                    }
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }, "log-producer").start();
    }

    // Consumer: processes the full buffer, swaps back the empty one
    public void startConsumer() {
        new Thread(() -> {
            List<LogEvent> buffer = new ArrayList<>(batchSize);
            try {
                while (!Thread.currentThread().isInterrupted()) {
                    buffer = exchanger.exchange(buffer);  // get full buffer
                    writeBatch(buffer);  // process
                    buffer.clear();      // prepare for next swap
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }, "log-consumer").start();
    }
}

try {
    List<Order> received = exchanger.exchange(myBuffer, 5, TimeUnit.SECONDS);
} catch (TimeoutException e) {
    // Partner didn't show up within 5 seconds — handle gracefully
    processPartialBatch(myBuffer);
}

ExecutorService executor = Executors.newFixedThreadPool(10);
CompletionService<PriceQuote> completionService = 
    new ExecutorCompletionService<>(executor);

// Submit requests to multiple pricing services
for (String vendor : vendors) {
    completionService.submit(() -> fetchPrice(vendor, item));
}

// Process results as they arrive (fastest first)
for (int i = 0; i < vendors.size(); i++) {
    Future<PriceQuote> future = completionService.take();  // blocks for next result
    try {
        PriceQuote quote = future.get();
        if (quote.getPrice() < bestPrice) {
            bestPrice = quote.getPrice();
        }
    } catch (ExecutionException e) {
        log.warn("Vendor failed: {}", e.getCause().getMessage());
    }
}

// BAD: processes in submission order — blocks on slow tasks
List<Future<String>> futures = executor.invokeAll(tasks);
for (Future<String> f : futures) {
    f.get();  // if task 0 is slow, we wait even though task 3 finished first
}

// GOOD: processes in completion order — fastest results first
for (int i = 0; i < tasks.size(); i++) {
    Future<String> f = completionService.take();  // next completed result
    process(f.get());  // always processes the fastest available
}

// Send request to 5 replicas, return as soon as 3 respond (quorum)
public List<Response> quorumRead(Request req, List<Replica> replicas, int quorum) 
        throws InterruptedException {
    CompletionService<Response> cs = new ExecutorCompletionService<>(executor);

    for (Replica r : replicas) {
        cs.submit(() -> r.read(req));
    }

    List<Response> responses = new ArrayList<>();
    for (int i = 0; i < quorum; i++) {
        responses.add(cs.take().get());
    }
    return responses;  // return after quorum reached — don't wait for stragglers
}

public class LockFreeStack<T> {
    private final AtomicReference<Node<T>> top = new AtomicReference<>(null);

    private static class Node<T> {
        final T value;
        final Node<T> next;
        Node(T value, Node<T> next) {
            this.value = value;
            this.next = next;
        }
    }

    public void push(T value) {
        Node<T> newNode = new Node<>(value, null);
        Node<T> current;
        do {
            current = top.get();
            newNode = new Node<>(value, current);
        } while (!top.compareAndSet(current, newNode));
    }

    public T pop() {
        Node<T> current;
        Node<T> next;
        do {
            current = top.get();
            if (current == null) return null;
            next = current.next;
        } while (!top.compareAndSet(current, next));
        return current.value;
    }
}

// Conceptual Disruptor usage (using LMAX library)
Disruptor<OrderEvent> disruptor = new Disruptor<>(
    OrderEvent::new,          // event factory
    1024,                     // ring buffer size (power of 2)
    DaemonThreadFactory.INSTANCE,
    ProducerType.MULTI,
    new BusySpinWaitStrategy() // lowest latency, highest CPU
);

disruptor.handleEventsWith(journaler, replicator)
    .then(businessLogic);     // pipeline: journal+replicate first, then process

disruptor.start();

// Publish event
RingBuffer<OrderEvent> rb = disruptor.getRingBuffer();
long sequence = rb.next();
try {
    OrderEvent event = rb.get(sequence);
    event.setOrderId(orderId);
    event.setPrice(price);
} finally {
    rb.publish(sequence);
}

import java.util.concurrent.*;

public class ParallelSum extends RecursiveTask<Long> {

    private static final int THRESHOLD = 10_000;
    private final long[] array;
    private final int start, end;

    public ParallelSum(long[] array, int start, int end) {
        this.array = array;
        this.start = start;
        this.end = end;
    }

    @Override
    protected Long compute() {
        if (end - start <= THRESHOLD) {
            // Base case: compute directly
            long sum = 0;
            for (int i = start; i < end; i++) sum += array[i];
            return sum;
        }

        int mid = (start + end) / 2;
        ParallelSum left = new ParallelSum(array, start, mid);
        ParallelSum right = new ParallelSum(array, mid, end);

        left.fork();         // submit left to the pool
        long rightResult = right.compute(); // compute right in current thread
        long leftResult = left.join();      // wait for left

        return leftResult + rightResult;
    }

    public static void main(String[] args) {
        long[] data = new long[1_000_000];
        for (int i = 0; i < data.length; i++) data[i] = i + 1;

        ForkJoinPool pool = ForkJoinPool.commonPool();
        long result = pool.invoke(new ParallelSum(data, 0, data.length));
        System.out.println("Sum: " + result);
    }
}

import java.util.concurrent.RecursiveAction;
import java.util.Arrays;

public class ParallelMergeSort extends RecursiveAction {

    private static final int THRESHOLD = 4096;
    private final int[] array;
    private final int start, end;

    public ParallelMergeSort(int[] array, int start, int end) {
        this.array = array;
        this.start = start;
        this.end = end;
    }

    @Override
    protected void compute() {
        if (end - start <= THRESHOLD) {
            Arrays.sort(array, start, end); // small enough, sort sequentially
            return;
        }
        int mid = (start + end) / 2;
        ParallelMergeSort left = new ParallelMergeSort(array, start, mid);
        ParallelMergeSort right = new ParallelMergeSort(array, mid, end);

        invokeAll(left, right); // fork both, wait for both
        merge(array, start, mid, end);
    }

    private void merge(int[] arr, int lo, int mid, int hi) {
        int[] temp = Arrays.copyOfRange(arr, lo, mid);
        int i = 0, j = mid, k = lo;
        while (i < temp.length && j < hi) {
            arr[k++] = (temp[i] <= arr[j]) ? temp[i++] : arr[j++];
        }
        while (i < temp.length) arr[k++] = temp[i++];
    }
}

public class Singleton {
    // volatile prevents instruction reordering
    private static volatile Singleton INSTANCE;

    private Singleton() {}

    public static Singleton getInstance() {
        if (INSTANCE == null) {               // 1st check (no lock)
            synchronized (Singleton.class) {
                if (INSTANCE == null) {       // 2nd check (with lock)
                    INSTANCE = new Singleton();
                }
            }
        }
        return INSTANCE;
    }
}

public enum DatabaseConnection {
    INSTANCE;

    private final Connection conn;

    DatabaseConnection() {
        conn = createConnection();
    }

    public Connection getConnection() {
        return conn;
    }

    private Connection createConnection() {
        // ... initialize real connection
        return null;
    }
}

// Usage
DatabaseConnection.INSTANCE.getConnection();

public class Singleton {
    private Singleton() {}

    private static class Holder {
        // Inner class is not loaded until getInstance() is called
        static final Singleton INSTANCE = new Singleton();
    }

    public static Singleton getInstance() {
        return Holder.INSTANCE;
    }
}

import java.util.concurrent.*;

public class ObjectPool<T> {

    private final BlockingQueue<T> pool;
    private final java.util.function.Supplier<T> factory;

    public ObjectPool(int size, java.util.function.Supplier<T> factory) {
        this.pool = new ArrayBlockingQueue<>(size);
        this.factory = factory;
        for (int i = 0; i < size; i++) {
            pool.offer(factory.get());
        }
    }

    public T borrow() throws InterruptedException {
        return pool.take(); // blocks if pool is empty
    }

    public T borrowWithTimeout(long ms) throws InterruptedException {
        T obj = pool.poll(ms, TimeUnit.MILLISECONDS);
        if (obj == null) throw new RuntimeException("Pool exhausted, timeout after " + ms + "ms");
        return obj;
    }

    public void release(T obj) {
        pool.offer(obj); // return to pool
    }

    // Usage
    public static void main(String[] args) throws Exception {
        ObjectPool<StringBuilder> pool = new ObjectPool<>(5, StringBuilder::new);

        StringBuilder sb = pool.borrow();
        try {
            sb.setLength(0); // reset before use
            sb.append("Hello from pooled object");
            System.out.println(sb);
        } finally {
            pool.release(sb); // always return to pool
        }
    }
}

import java.util.concurrent.*;

public class ActiveObject {

    private final ExecutorService executor = Executors.newSingleThreadExecutor();
    private double balance = 0.0;

    public Future<Double> deposit(double amount) {
        return executor.submit(() -> {
            balance += amount;
            System.out.println("Deposited " + amount + ", balance = " + balance);
            return balance;
        });
    }

    public Future<Double> withdraw(double amount) {
        return executor.submit(() -> {
            if (balance >= amount) {
                balance -= amount;
                System.out.println("Withdrew " + amount + ", balance = " + balance);
            } else {
                System.out.println("Insufficient funds for " + amount);
            }
            return balance;
        });
    }

    public Future<Double> getBalance() {
        return executor.submit(() -> balance);
    }

    public void shutdown() {
        executor.shutdown();
    }

    public static void main(String[] args) throws Exception {
        ActiveObject account = new ActiveObject();

        // Multiple threads can call these safely -- all serialized internally
        account.deposit(1000);
        account.withdraw(200);
        Future<Double> balance = account.getBalance();
        System.out.println("Final balance: " + balance.get());

        account.shutdown();
    }
}

import java.util.concurrent.*;

public class HalfSyncHalfAsync {

    private final BlockingQueue<String> requestQueue = new LinkedBlockingQueue<>();
    private final ExecutorService syncWorkers = Executors.newFixedThreadPool(4);

    // ASYNC layer: non-blocking event loop receives requests
    public void onRequestReceived(String request) {
        requestQueue.offer(request); // non-blocking enqueue
    }

    // SYNC layer: blocking workers process from the queue
    public void startSyncLayer() {
        for (int i = 0; i < 4; i++) {
            syncWorkers.submit(() -> {
                while (!Thread.currentThread().isInterrupted()) {
                    try {
                        String request = requestQueue.take(); // blocking
                        processRequest(request);              // synchronous business logic
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        break;
                    }
                }
            });
        }
    }

    private void processRequest(String request) {
        System.out.println(Thread.currentThread().getName() + " processing: " + request);
    }
}

// Thread-per-message with virtual threads (Java 21+)
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
    serverSocket.accept(); // for each connection:
    executor.submit(() -> handleClient(clientSocket));
}

import java.nio.channels.*;
import java.nio.*;
import java.net.*;

public class ReactorDemo {

    public void start() throws Exception {
        Selector selector = Selector.open();
        ServerSocketChannel server = ServerSocketChannel.open();
        server.bind(new InetSocketAddress(8080));
        server.configureBlocking(false);
        server.register(selector, SelectionKey.OP_ACCEPT);

        while (true) {
            selector.select(); // blocks until events arrive
            var keys = selector.selectedKeys().iterator();
            while (keys.hasNext()) {
                SelectionKey key = keys.next();
                keys.remove();

                if (key.isAcceptable()) {
                    SocketChannel client = server.accept();
                    client.configureBlocking(false);
                    client.register(selector, SelectionKey.OP_READ);
                } else if (key.isReadable()) {
                    SocketChannel client = (SocketChannel) key.channel();
                    ByteBuffer buf = ByteBuffer.allocate(1024);
                    client.read(buf);
                    // process data...
                }
            }
        }
    }
}

public class LazyInitResource {

    private volatile boolean initialized = false;
    private Resource resource;

    public synchronized void initialize() {
        if (initialized) return; // BALK -- already initialized

        resource = new Resource();
        resource.connect();
        initialized = true;
    }

    public void useResource() {
        if (!initialized) {
            throw new IllegalStateException("Resource not initialized -- call initialize() first");
        }
        resource.execute();
    }
}

import java.util.LinkedList;
import java.util.Queue;

public class GuardedQueue<T> {

    private final Queue<T> queue = new LinkedList<>();
    private final int capacity;

    public GuardedQueue(int capacity) {
        this.capacity = capacity;
    }

    public synchronized void put(T item) throws InterruptedException {
        while (queue.size() >= capacity) {
            wait(); // guard: wait until space is available
        }
        queue.add(item);
        notifyAll(); // wake up consumers
    }

    public synchronized T take() throws InterruptedException {
        while (queue.isEmpty()) {
            wait(); // guard: wait until item is available
        }
        T item = queue.poll();
        notifyAll(); // wake up producers
        return item;
    }
}