Java并发24：CountDownLatch与CyclicBarrier - 线程协作工具

CountDownLatch latch = new CountDownLatch(3);  // 初始化计数为3

// 工作线程
new Thread(() -> {
    doWork();
    latch.countDown();  // 计数-1
}).start();

// 主线程等待
latch.await();  // 阻塞，直到计数为0
System.out.println("所有任务完成");

public class ApplicationStarter {
    private final CountDownLatch latch = new CountDownLatch(3);

    public void start() throws InterruptedException {
        // 启动数据库
        new Thread(() -> {
            initDatabase();
            latch.countDown();
        }).start();

        // 启动缓存
        new Thread(() -> {
            initCache();
            latch.countDown();
        }).start();

        // 启动MQ
        new Thread(() -> {
            initMQ();
            latch.countDown();
        }).start();

        // 等待所有服务启动完成
        latch.await();
        System.out.println("应用启动完成");
    }
}

public class ParallelTask {
    public List<Result> executeParallel(List<Task> tasks) throws InterruptedException {
        CountDownLatch latch = new CountDownLatch(tasks.size());
        List<Result> results = new CopyOnWriteArrayList<>();

        for (Task task : tasks) {
            executor.submit(() -> {
                try {
                    Result result = task.execute();
                    results.add(result);
                } finally {
                    latch.countDown();
                }
            });
        }

        latch.await();  // 等待所有任务完成
        return results;
    }
}

public class LoadTest {
    public void test(int threadCount) throws InterruptedException {
        CountDownLatch startGate = new CountDownLatch(1);  // 起跑门
        CountDownLatch endGate = new CountDownLatch(threadCount);  // 终点门

        for (int i = 0; i < threadCount; i++) {
            new Thread(() -> {
                try {
                    startGate.await();  // 等待起跑信号
                    doTest();
                } finally {
                    endGate.countDown();
                }
            }).start();
        }

        long start = System.currentTimeMillis();
        startGate.countDown();  // 发出起跑信号，所有线程同时开始
        endGate.await();  // 等待所有线程完成
        long time = System.currentTimeMillis() - start;

        System.out.println("测试完成，耗时: " + time + "ms");
    }
}

CyclicBarrier barrier = new CyclicBarrier(3, () -> {
    System.out.println("所有线程到达屏障，开始下一阶段");
});

// 工作线程
new Thread(() -> {
    doWork1();
    barrier.await();  // 等待其他线程
    doWork2();
}).start();

public class ParallelCompute {
    public void compute() throws Exception {
        int threadCount = 4;
        CyclicBarrier barrier = new CyclicBarrier(threadCount, () -> {
            System.out.println("所有线程计算完成，合并结果");
            mergeResults();
        });

        for (int i = 0; i < threadCount; i++) {
            int part = i;
            new Thread(() -> {
                try {
                    computePart(part);
                    barrier.await();  // 等待其他线程
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }).start();
        }
    }
}

public class BatchProcessor {
    public void process(List<Data> dataList) throws Exception {
        int batchSize = 100;
        CyclicBarrier barrier = new CyclicBarrier(3);  // 可重用

        for (int i = 0; i < dataList.size(); i += batchSize) {
            List<Data> batch = dataList.subList(i, Math.min(i + batchSize, dataList.size()));

            executor.submit(() -> {
                processBatch(batch);
                barrier.await();  // 等待本批次处理完成
                return null;
            });
        }
    }
}

// CountDownLatch：主线程等待工作线程
CountDownLatch latch = new CountDownLatch(3);
for (int i = 0; i < 3; i++) {
    new Thread(() -> {
        doWork();
        latch.countDown();  // 工作线程完成，计数-1
    }).start();
}
latch.await();  // 主线程等待

// CyclicBarrier：工作线程互相等待
CyclicBarrier barrier = new CyclicBarrier(3);
for (int i = 0; i < 3; i++) {
    new Thread(() -> {
        doWork();
        barrier.await();  // 工作线程互相等待
    }).start();
}

// 基于AQS的共享模式
public class CountDownLatch {
    private final Sync sync;

    private static final class Sync extends AbstractQueuedSynchronizer {
        Sync(int count) {
            setState(count);  // 设置初始计数
        }

        // countDown()
        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;  // 计数为0时返回true
            }
        }

        // await()
        protected int tryAcquireShared(int acquires) {
            return (getState() == 0) ? 1 : -1;  // 计数为0时获取成功
        }
    }
}

// 基于ReentrantLock + Condition
public class CyclicBarrier {
    private final ReentrantLock lock = new ReentrantLock();
    private final Condition trip = lock.newCondition();
    private final int parties;
    private int count;  // 当前等待线程数

    public int await() throws InterruptedException, BrokenBarrierException {
        lock.lock();
        try {
            int index = --count;
            if (index == 0) {
                // 最后一个到达，唤醒所有线程
                trip.signalAll();
                // 重置count，可重用
                count = parties;
            } else {
                // 等待其他线程
                trip.await();
            }
            return index;
        } finally {
            lock.unlock();
        }
    }
}

// ❌ 不能重置，必须创建新实例
latch.await();
// latch无法重置，需要：
latch = new CountDownLatch(3);

// ✅ 可以reset
barrier.reset();  // 重置屏障，唤醒所有等待线程（抛BrokenBarrierException）

// CountDownLatch
if (latch.await(10, TimeUnit.SECONDS)) {
    // 在10秒内完成
} else {
    // 超时
}

// CyclicBarrier
try {
    barrier.await(10, TimeUnit.SECONDS);
} catch (TimeoutException e) {
    // 超时
}