Master Java Concurrency: Fork/Join, CountDownLatch, CyclicBarrier, Semaphore & Exchanger Deep Dive

Java Concurrency Overview

The article introduces a series of Java concurrency utilities, focusing on their design goals, underlying AbstractQueuedSynchronizer (AQS) mechanisms, and practical usage in high‑performance applications.

Fork/Join Framework

Fork/Join, introduced in Java 7, implements a parallel computation model based on the "divide‑and‑conquer" principle. It automatically splits large tasks into subtasks, executes them in a work‑stealing thread pool ( ForkJoinPool), and merges results.

Work‑Stealing Algorithm

Each worker thread maintains a double‑ended queue ( Deque) where the head processes its own tasks (LIFO) and the tail is used by other threads to steal work (FIFO), maximizing CPU utilization.

Task Splitting Flow

Large tasks are recursively divided until a configurable THRESHOLD is reached, after which a simple sort is performed and results are merged.

public class ParallelMergeSort extends RecursiveAction {
    private final int[] array;
    private final int start;
    private final int end;
    private static final int THRESHOLD = 1_000_000;

    @Override
    protected void compute() {
        if (end - start <= THRESHOLD) {
            Arrays.sort(array, start, end);
            return;
        }
        int mid = (start + end) >>> 1;
        invokeAll(new ParallelMergeSort(array, start, mid),
                  new ParallelMergeSort(array, mid, end));
        merge(array, start, mid, end);
    }
    // merge implementation omitted for brevity
}

ForkJoinPool pool = new ForkJoinPool(Runtime.getRuntime().availableProcessors());
int[] data = loadHugeData();
pool.invoke(new ParallelMergeSort(data, 0, data.length));

Performance Optimizations

Fine‑tune THRESHOLD based on benchmark results.

Avoid creating temporary arrays inside recursion.

Use ThreadLocalRandom for thread‑safe random number generation.

CountDownLatch

CountDownLatch

allows one or more threads to wait until a set of operations completes. It is built on AQS via a static inner class Sync that extends AbstractQueuedSynchronizer.

private static final class Sync extends AbstractQueuedSynchronizer {
    Sync(int count) { setState(count); }
    protected int tryAcquireShared(int acquires) {
        return (getState() == 0) ? 1 : -1;
    }
    protected boolean tryReleaseShared(int releases) {
        for (;;) {
            int c = getState();
            if (c == 0) return false;
            int nextc = c - 1;
            if (compareAndSetState(c, nextc))
                return nextc == 0;
        }
    }
}
public void await() throws InterruptedException {
    sync.acquireSharedInterruptibly(1);
}
public void countDown() {
    sync.releaseShared(1);
}

The await() method invokes acquireSharedInterruptibly, which repeatedly calls tryAcquireShared until the count reaches zero, then releases waiting threads.

CyclicBarrier

CyclicBarrier

is a reusable synchronization point for a fixed number of threads. It differs from CountDownLatch by allowing the barrier to be reset and reused.

public CyclicBarrier(int parties) { this(parties, null); }
public CyclicBarrier(int parties, Runnable barrierAction) { ... }
public int await() throws InterruptedException, BrokenBarrierException {
    return dowait(false, 0L);
}
private int dowait(boolean timed, long nanos) throws ... {
    final Generation g = generation;
    int index = --count;
    if (index == 0) {
        // run barrierAction, create new generation, release all
    }
    // otherwise, thread waits on Condition until generation changes
}

The implementation uses a ReentrantLock and a Condition to block threads until the required number arrives.

Semaphore

Semaphore

controls the number of concurrent accesses to a resource. It supports fair and non‑fair modes, both implemented via AQS subclasses FairSync and NonfairSync.

public void acquire() throws InterruptedException {
    sync.acquireSharedInterruptibly(1);
}
public void release() {
    sync.releaseShared(1);
}
abstract static class Sync extends AbstractQueuedSynchronizer {
    Sync(int permits) { setState(permits); }
    protected final int tryAcquireShared(int acquires) { ... }
    protected final boolean tryReleaseShared(int releases) { ... }
}
static final class FairSync extends Sync { ... }
static final class NonfairSync extends Sync { ... }

Fair mode checks the waiting queue before acquiring permits, ensuring FIFO ordering.

Exchanger

Exchanger

enables two threads to swap objects at a synchronization point. Each thread holds a thread‑local Node (via ThreadLocal) to avoid contention.

public V exchange(V x) throws InterruptedException {
    Node q = participant.get();
    // try to match with a waiting node; if none, park current thread
    // upon match, set node.match = x and unpark partner
}
static final class Node {
    volatile Object item;   // data offered by this thread
    volatile Object match; // data received from partner
    volatile Thread parked; // thread waiting for partner
}

The exchange proceeds by CAS‑setting a shared slot, swapping item and match, and unparking the partner thread.

Key Takeaways

All utilities share a common foundation in AQS, which provides a template method pattern for acquiring and releasing synchronization state.

Understanding the internal tryAcquireShared and tryReleaseShared implementations helps diagnose performance bottlenecks and choose the right tool for a given concurrency scenario.

Proper configuration (e.g., thresholds, permit counts, barrier parties) and awareness of fairness vs. non‑fairness trade‑offs are essential for production‑grade systems.