Java CyclicBarrier 同步屏障完整指南：核心原理、await()機制、可重用性與對比 CountDownLatch

CyclicBarrier 核心定位

CyclicBarrier (循環屏障)：讓一組線程互相等待，直到所有線程都到達屏障點，然後同時繼續執行。關鍵特性：可重用 (循環使用)、支援屏障動作 (所有線程到達前執行)。

核心構造函數

public class CyclicBarrierDemo {

    // 1. 基本構造：指定攔截線程數
    CyclicBarrier barrier = new CyclicBarrier(3);

    // 2. 攜帶屏障動作：所有線程到達前，由其中一個線程執行
    CyclicBarrier barrierWithAction = new CyclicBarrier(3, () -> {
        System.out.println("=== 所有線程已就緒，執行屏障動作 ===");
        // 可用於匯總結果、初始化共享資源、發送啟動信號
    });
}

完整實戰範例：多線程分階段計算

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

public class CyclicBarrierExample {

    static final int WORKER_COUNT = 5;
    static final CyclicBarrier BARRIER = new CyclicBarrier(WORKER_COUNT, 
        () -> System.out.println(">>> 階段完成，進入下一階段 <<<"));

    static class Worker implements Runnable {
        private final int id;
        private final List results = Collections.synchronizedList(new ArrayList<>());

        Worker(int id) { this.id = id; }

        @Override
        public void run() {
            try {
                // 階段 1：數據準備
                System.out.printf("Worker-%d: 階段1 開始%n", id);
                Thread.sleep(new Random().nextInt(1000));
                int phase1Result = id * 10;
                results.add(phase1Result);
                System.out.printf("Worker-%d: 階段1 完成，等待屏障%n", id);
                BARRIER.await(); // 關鍵：阻塞直到所有線程到達

                // 階段 2：數據處理 (所有線程同步啟動)
                System.out.printf("Worker-%d: 階段2 開始%n", id);
                Thread.sleep(new Random().nextInt(1000));
                int phase2Result = phase1Result * 2;
                results.add(phase2Result);
                System.out.printf("Worker-%d: 階段2 完成，等待屏障%n", id);
                BARRIER.await();

                // 階段 3：結果匯總
                System.out.printf("Worker-%d: 階段3 開始%n", id);
                Thread.sleep(new Random().nextInt(1000));
                System.out.printf("Worker-%d: 完成，結果: %s%n", id, results);
            } catch (InterruptedException | BrokenBarrierException e) {
                Thread.currentThread().interrupt();
                System.err.println("Worker-" + id + " 異常: " + e.getMessage());
            }
        }
    }

    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(WORKER_COUNT);
        for (int i = 1; i <= WORKER_COUNT; i++) {
            executor.submit(new Worker(i));
        }
        executor.shutdown();
    }
}

await() 核心機制與異常處理

// 1. 基本等待
barrier.await(); // 阻塞直到計數歸零，拋 InterruptedException/BrokenBarrierException

// 2. 超時等待 (避免無限阻塞)
try {
    barrier.await(10, TimeUnit.SECONDS); // 超時拋 TimeoutException
} catch (TimeoutException e) {
    // 超時處理：記錄、重試、取消
}

// 3. 關鍵異常區別
// InterruptedException: 當前線程被中斷 (Thread.interrupt())
// BrokenBarrierException: 屏障被打破 (其他線程中斷/超時/重置)
// TimeoutException: 僅 await(timeout) 拋出，屏障狀態不變

reset() 重用機制

// 重置屏障：計數歸零、喚醒所有等待線程 (拋 BrokenBarrierException)
// 適用場景：重複的階段性任務、遊戲回合制、批次處理循環
public class ReusableBarrier {
    private final CyclicBarrier barrier = new CyclicBarrier(3);

    public void runPhase(int phase) {
        try {
            doWork(phase);
            barrier.await();
        } catch (BrokenBarrierException e) {
            // 屏障被重置/打破，通常意味著任務取消
            Thread.currentThread().interrupt();
        }
    }

    public void resetForNextRound() {
        barrier.reset(); // 所有等待線程收到 BrokenBarrierException
    }
}

CyclicBarrier vs CountDownLatch 核心對比

進階應用場景

1. MapReduce 風格並行聚合

public class ParallelAggregator {
    private final CyclicBarrier barrier;
    private final List partials = Collections.synchronizedList(new ArrayList<>());

    public ParallelAggregator(int workers) {
        this.barrier = new CyclicBarrier(workers, this::mergeResults);
    }

    private void mergeResults() {
        // 由最後到達的線程執行合併
        Result finalResult = partials.stream().reduce(Result::merge).orElse(Result.empty());
        System.out.println("最終結果: " + finalResult);
    }

    public void submitTask(Callable task) {
        executor.submit(() -> {
            PartialResult pr = task.call();
            partials.add(pr);
            barrier.await(); // 等待所有部分完成
        });
    }
}

2. 可控制的壓力測試同步啟動

public class LoadTestStarter {
    private final CyclicBarrier startBarrier;
    private final CyclicBarrier endBarrier;
    private final AtomicInteger successCount = new AtomicInteger();

    public LoadTestStarter(int concurrentUsers) {
        this.startBarrier = new CyclicBarrier(concurrentUsers);
        this.endBarrier = new CyclicBarrier(concurrentUsers, 
            () -> System.out.println("測試結束，成功: " + successCount.get()));
    }

    public void runUserSimulation(Runnable userAction) {
        executor.submit(() -> {
            try {
                startBarrier.await(); // 所有人同時起跑
                userAction.run();
                successCount.incrementAndGet();
            } catch (Exception e) {
                // 記錄失敗
            } finally {
                try { endBarrier.await(); } catch (Exception ignored) {}
            }
        });
    }
}

3. 動態調整參與者數量 (技巧)

// CyclicBarrier 構造後人數固定，需動態調整時：
// 方案：使用 Phaser (Java 7+) - 支援動態註冊/取消註冊
Phaser phaser = new Phaser(1); // 主線程註冊
// 子任務啟動時
phaser.register(); // 註冊參與者
// 子任務結束時
phaser.arriveAndDeregister(); // 到達並取消註冊
// 主線程等待
phaser.arriveAndAwaitAdvance(); // 相當於 await()

常見錯誤與最佳實踐

• 黨數不匹配：構造函數 N 必須等於實際調用 await() 的線程數，否則永久阻塞
• 忽略 BrokenBarrierException：必須處理，通常意味著任務取消，需清理資源
• 屏障動作耗時過長：由最後到達線程執行，會延遲其他線程繼續，保持輕量
• 未使用超時：生產環境建議 await(timeout, unit) 避免死鎖
• 混淆 CountDownLatch：主線程等待工作者用 CountDownLatch，工作者互相等待用 CyclicBarrier