Java CyclicBarrier Deep Dive: Batch Synchronization Patterns & Design Improvements

Java's CyclicBarrier coordinates a fixed number of threads so they all wait at a barrier point until every thread arrives, then proceed together. This enables phased parallel execution: threads work independently within a phase, synchronize at the barrier, then move to the next phase.

Example 1: API Stress Testing

The first example simulates 10 concurrent users hitting a product-detail endpoint simultaneously. A CyclicBarrier initialized with 10 parties ensures all threads finish preparation (random sleep up to 100 ms) before any thread calls the API.

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

/**
 * Scenario: simulate 10 concurrent users requesting product detail API,
 * testing response time under high concurrency.
 * Core: all threads prepare, then fire requests at the same instant
 * to avoid staggered start times.
 */
public class ProductApiStressTest {
    private static final int CONCURRENCY = 10;

    public static void main(String[] args) throws Exception {
        ExecutorService executor = Executors.newFixedThreadPool(CONCURRENCY);
        CyclicBarrier startGate = new CyclicBarrier(CONCURRENCY);

        for (int i = 0; i < CONCURRENCY; i++) {
            executor.submit(() -> {
                try {
                    // Simulate thread initialization, connection setup
                    Thread.sleep((long) (Math.random() * 100));

                    // All threads wait here until 10 have arrived
                    startGate.await();

                    // Simultaneously invoke API and measure latency
                    long start = System.currentTimeMillis();
                    callProductDetailApi();
                    long cost = System.currentTimeMillis() - start;

                    System.out.println(Thread.currentThread().getName()
                            + " | Response time: " + cost + "ms");
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
            });
        }
        executor.shutdown();
    }

    static void callProductDetailApi() throws InterruptedException {
        // Simulate API call latency 50-200 ms
        Thread.sleep(50 + (long)(Math.random() * 150));
    }
}

Each worker thread calls await(); the last arrival releases all waiting threads so they invoke callProductDetailApi() concurrently. This demonstrates single-batch coordination.

Example 2: Batch Importing 1 Million Excel Rows

The second example processes a large Excel file in batches of 5,000 rows. Four worker threads parse and validate each partition in parallel; a new CyclicBarrier is created per batch with THREAD_COUNT + 1 parties (4 workers + main thread). The barrier's runnable action performs the bulk database insert after all partitions are validated.

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Scenario: stream-read Excel data, 5000 rows per batch,
 * multi-threaded validation followed by bulk insert.
 * Core: CyclicBarrier defined inside while loop, each batch gets
 * its own independent synchronization cycle.
 */
public class BatchDataImport {
    private static final int BATCH_SIZE = 5000;
    private static final int THREAD_COUNT = 4;

    public static void main(String[] args) throws Exception {
        ExecutorService executor = Executors.newFixedThreadPool(THREAD_COUNT);
        List<String> batch;
        int batchNo = 0;

        while (!(batch = readNextBatch(BATCH_SIZE)).isEmpty()) {
            batchNo++;
            System.out.println("=== Start processing batch " + batchNo
                    + ", " + batch.size() + " rows ===");

            // Key: barrier created inside loop, each batch a fresh cycle
            // parties = THREAD_COUNT + 1 because main thread also waits
            CyclicBarrier barrier = new CyclicBarrier(THREAD_COUNT + 1, () -> {
                // Executed by the last arriving thread (worker or main)
                System.out.println("  [Barrier triggered] Batch validation done, executing bulk INSERT...");
                // batchInsert(currentBatchData);
            });

            // Split batch evenly into THREAD_COUNT partitions
            List<List<String>> partitions = partition(batch, THREAD_COUNT);
            for (List<String> subTask : partitions) {
                executor.submit(() -> {
                    try {
                        parseAndValidate(subTask); // parallel validation
                        barrier.await();           // wait at barrier
                    } catch (Exception e) {
                        throw new RuntimeException(e);
                    }
                });
            }

            // Main thread explicitly awaits as the (THREAD_COUNT+1)th participant
            // Guarantees next loop iteration starts only after this batch persists
            barrier.await();

            System.out.println("=== Batch " + batchNo + " completed, safe to proceed ===
");
        }
        executor.shutdown();
        System.out.println("All data imported!");
    }

    static List<String> readNextBatch(int size) {
        /* Stream-read from Excel */
        return new ArrayList<>();
    }

    static void parseAndValidate(List<String> data) {
        /* Parse and validate */
    }

    static <T> List<List<T>> partition(List<T> list, int n) {
        /* Evenly split */
        return new ArrayList<>();
    }
}

Design Issue: Main Thread Participating in the Barrier

The code uses new CyclicBarrier(THREAD_COUNT + 1) because the main thread must also wait for the batch to finish before reading the next batch. However, this mixes two responsibilities: the barrier coordinates worker threads and the main thread. The thread that executes the barrier action (bulk insert) could be either a worker or the main thread, leading to ambiguous ownership.

Improved Design: Separate Coordination with CountDownLatch

A cleaner separation uses CountDownLatch for the main-thread wait and keeps CyclicBarrier solely for worker synchronization. The barrier's party count becomes THREAD_COUNT (workers only). The last worker to arrive performs the bulk insert and then counts down the latch, releasing the main thread.

CyclicBarrier barrier = new CyclicBarrier(THREAD_COUNT, () -> {
    // Executed by the last arriving worker thread
    System.out.println("  [Barrier triggered] Batch validation done, executing bulk INSERT...");
    // batchInsert(currentBatchData);

    // After insert, signal main thread that this batch is finished
    batchLatch.countDown();
});

In the loop, create CountDownLatch batchLatch = new CountDownLatch(1); before submitting tasks, and replace the main thread's barrier.await() with batchLatch.await(). This makes roles explicit: CyclicBarrier synchronizes workers; CountDownLatch signals batch completion to the driver thread.

Understanding these patterns helps you apply CyclicBarrier correctly for phased parallel tasks while avoiding tangled coordination logic.