CountDownLatch vs CyclicBarrier in Java — Deep Dive with Real Examples

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * Simulates a service that must wait for three dependent sub-services
 * (database, cache, and message broker) to finish initialising before
 * it opens its own HTTP listener.
 */
public class ServiceStartupCoordinator {

    // Three sub-services must signal readiness before the main service starts.
    private static final int DEPENDENCY_COUNT = 3;

    public static void main(String[] args) throws InterruptedException {

        CountDownLatch readinessLatch = new CountDownLatch(DEPENDENCY_COUNT);
        ExecutorService startupPool = Executors.newFixedThreadPool(DEPENDENCY_COUNT);

        // Each Runnable simulates a sub-service initialising and then
        // decrementing the latch to signal it is ready.
        startupPool.submit(new SubServiceInitialiser("DatabasePool",   1200, readinessLatch));
        startupPool.submit(new SubServiceInitialiser("RedisCache",      400, readinessLatch));
        startupPool.submit(new SubServiceInitialiser("MessageBroker",   800, readinessLatch));

        System.out.println("[MainThread] Waiting for all dependencies to become ready...");

        // await() parks the main thread inside AQS until the internal state reaches 0.
        // Using a timeout is critical in production — never block forever.
        boolean allReady = readinessLatch.await(5, TimeUnit.SECONDS);

        if (allReady) {
            System.out.println("[MainThread] All dependencies ready. Opening HTTP listener on :8080");
        } else {
            // This branch fires if one sub-service hangs past the timeout.
            System.err.println("[MainThread] Startup timeout! Shutting down safely.");
        }

        startupPool.shutdown();
    }

    static class SubServiceInitialiser implements Runnable {
        private final String serviceName;
        private final long   initDelayMs;    // Simulates different boot times
        private final CountDownLatch latch;

        SubServiceInitialiser(String serviceName, long initDelayMs, CountDownLatch latch) {
            this.serviceName  = serviceName;
            this.initDelayMs  = initDelayMs;
            this.latch        = latch;
        }

        @Override
        public void run() {
            try {
                System.out.printf("[%s] Initialising...%n", serviceName);
                Thread.sleep(initDelayMs);  // Simulate IO-bound startup work
                System.out.printf("[%s] Ready. Counting down.%n", serviceName);

                // countDown() is atomic — safe to call from multiple threads simultaneously.
                // It NEVER throws; even calling it when count is already 0 is a no-op.
                latch.countDown();

            } catch (InterruptedException e) {
                // Restore the interrupt flag — never swallow InterruptedException silently.
                Thread.currentThread().interrupt();
                System.err.printf("[%s] Interrupted during initialisation.%n", serviceName);
                // Still count down so the main thread isn't left waiting forever.
                latch.countDown();
            }
        }
    }
}

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.Arrays;

/**
 * Demonstrates a two-phase parallel computation.
 * Phase 1: Each worker thread computes the row-sum for its assigned matrix row.
 * Phase 2: After all row-sums are ready, each worker uses the global total
 *          to normalise its own row-sum.
 *
 * The CyclicBarrier ensures Phase 2 never starts until Phase 1 is 100% done.
 */
public class ParallelMatrixRowProcessor {

    private static final int ROW_COUNT    = 4;  // One worker thread per row
    private static final int COLUMN_COUNT = 5;

    // Shared result arrays — workers write here, the barrier action reads here.
    private static final int[]    rowSums     = new int[ROW_COUNT];
    private static       int      globalTotal = 0;  // Set by barrier action

    // Sample matrix — in practice this would be loaded from a data source.
    private static final int[][] matrix = {
        {1,  2,  3,  4,  5},
        {6,  7,  8,  9, 10},
        {11, 12, 13, 14, 15},
        {16, 17, 18, 19, 20}
    };

    public static void main(String[] args) throws InterruptedException {

        // The barrier action runs in the LAST thread to arrive.
        // It aggregates all row-sums into a global total before Phase 2 starts.
        Runnable aggregateRowSums = () -> {
            globalTotal = Arrays.stream(rowSums).sum();
            System.out.printf("%n[BarrierAction] All row sums computed. Global total = %d. Releasing Phase 2.%n%n", globalTotal);
        };

        // A single CyclicBarrier instance coordinates BOTH phases.
        // After it trips once (end of Phase 1), it resets automatically for Phase 2.
        CyclicBarrier phaseBarrier = new CyclicBarrier(ROW_COUNT, aggregateRowSums);

        ExecutorService workerPool = Executors.newFixedThreadPool(ROW_COUNT);

        for (int rowIndex = 0; rowIndex < ROW_COUNT; rowIndex++) {
            workerPool.submit(new RowProcessor(rowIndex, phaseBarrier));
        }

        workerPool.shutdown();
    }

    static class RowProcessor implements Runnable {
        private final int           rowIndex;
        private final CyclicBarrier phaseBarrier;

        RowProcessor(int rowIndex, CyclicBarrier phaseBarrier) {
            this.rowIndex     = rowIndex;
            this.phaseBarrier = phaseBarrier;
        }

        @Override
        public void run() {
            try {
                // ── PHASE 1: Compute this row's sum ──────────────────────────────────
                int sum = 0;
                for (int col = 0; col < COLUMN_COUNT; col++) {
                    sum += matrix[rowIndex][col];
                }
                rowSums[rowIndex] = sum;  // Write result to shared array
                System.out.printf("[Row-%d] Phase 1 done. Row sum = %d. Waiting at barrier.%n", rowIndex, sum);

                // await() decrements the internal count. When the last thread
                // arrives, the barrier action fires, THEN all threads are released.
                phaseBarrier.await();

                // ── PHASE 2: Normalise using the global total ─────────────────────────
                // At this point, globalTotal is guaranteed to be fully populated
                // because the barrier action completed before this line runs.
                double normalisedShare = (double) rowSums[rowIndex] / globalTotal * 100.0;
                System.out.printf("[Row-%d] Phase 2 done. Share of total = %.2f%%%n", rowIndex, normalisedShare);

            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                System.err.printf("[Row-%d] Interrupted.%n", rowIndex);
            } catch (BrokenBarrierException e) {
                // Another thread was interrupted or timed out — the barrier is broken.
                // Log and exit cleanly; do NOT proceed with partial data.
                System.err.printf("[Row-%d] Barrier broken — aborting phase processing.%n", rowIndex);
            }
        }
    }
}

import java.util.concurrent.Phaser;

/**
 * Quick illustration of Phaser as the flexible upgrade path.
 * Unlike CyclicBarrier, Phaser supports dynamic participant registration
 * and per-phase arrival tracking — useful when the number of workers
 * isn't known at construction time.
 *
 * This is NOT a replacement example — it's a pointer for when you've
 * outgrown both CountDownLatch and CyclicBarrier.
 */
public class PhaserMigrationHint {

    public static void main(String[] args) throws InterruptedException {

        // Phaser starts with 1 participant — the main thread (the 'overseer').
        Phaser overseerPhaser = new Phaser(1);

        for (int workerIndex = 0; workerIndex < 3; workerIndex++) {
            final int id = workerIndex;

            // Each worker registers itself dynamically — no fixed count at construction.
            overseerPhaser.register();

            Thread worker = new Thread(() -> {
                System.out.printf("[Worker-%d] Arriving at phase %d%n", id, overseerPhaser.getPhase());

                // arriveAndAwaitAdvance is the CyclicBarrier.await() equivalent.
                // Returns the phase number AFTER the barrier trips.
                overseerPhaser.arriveAndAwaitAdvance();

                System.out.printf("[Worker-%d] Phase advanced. Continuing.%n", id);

                // Deregister when done — reduces participant count for future phases.
                overseerPhaser.arriveAndDeregister();
            });
            worker.start();
        }

        // Main thread arrives — this is the last arrival that trips the phase.
        overseerPhaser.arriveAndDeregister();

        System.out.println("[Main] All workers released. Phaser terminated: " + overseerPhaser.isTerminated());
    }
}