Java线程协作工具：CountDownLatch 、CyclicBarrier、Phaser、Semaphore 、Exchanger

在高并发系统中，多个线程之间的协作是不可避免的。Java 从 JDK 1.5 开始引入了java.util.concurrent（JUC）包，其中包含了一系列强大的线程协作工具类，极大简化了并发编程的复杂性。

CountDownLatch（JDK 5+）：倒计时门闩

使用场景

CountDownLatch常用于“一个或多个线程等待其他线程完成任务后再继续执行”的场景，例如：

• 主线程等待所有子线程初始化完毕再启动服务。
• 并发测试中模拟 N 个请求同时发出后统计总耗时。
• 资源加载完成后统一触发后续逻辑。

核心 API

public CountDownLatch(int count); // 初始化计数器 public void await() throws InterruptedException; // 阻塞等待计数归零 public boolean await(long timeout, TimeUnit unit); // 带超时的等待 public void countDown(); // 计数减一

使用示例

public class CountDownLatchExample { public static void main(String[] args) throws InterruptedException { int threadCount = 5; CountDownLatch latch = new CountDownLatch(threadCount); for (int i = 0; i < threadCount; i++) { new Thread(() -> { try { System.out.println(Thread.currentThread().getName() + " 正在执行任务..."); Thread.sleep(1000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } finally { latch.countDown(); // 任务完成，计数减一 } }).start(); } latch.await(); // 主线程等待所有任务完成 System.out.println("所有任务已完成，主线程继续执行！"); } }

输出：

Thread-0 正在执行任务... Thread-1 正在执行任务... Thread-3 正在执行任务... Thread-2 正在执行任务... Thread-4 正在执行任务... 所有任务已完成，主线程继续执行！

应用场景示例

服务启动检查：

// 等待数据库、缓存、消息队列初始化完成 CountDownLatch startupLatch = new CountDownLatch(3); initDB(startupLatch); initCache(startupLatch); initMQ(startupLatch); startupLatch.await(); startHttpServer();

使用注意点

• 不可重用：一旦计数归零，无法再次使用（除非新建实例）。需要重复使用的场景应考虑CyclicBarrier，
• 若某个线程未调用countDown()，会导致主线程永久阻塞， 导致死锁，所以务必在 finally 块中调用。
• await()可被中断，需处理InterruptedException。

底层原理

CountDownLatch基于AQS（AbstractQueuedSynchronizer）实现：（AQS 原理详见Java并发编程：Lock原理详解）

• 内部状态state表示剩余计数值。
• await()对应 AQS 的acquireSharedInterruptibly(1)，尝试获取共享锁，若state != 0则入队阻塞。
• countDown()调用releaseShared(1)，将state减 1，当state == 0时唤醒所有等待线程。

CyclicBarrier（JDK 5+）：循环屏障

使用场景

CyclicBarrier适用于“多个线程互相等待，全部到达某一点后才能继续”的场景，例如：

• 多线程分阶段计算（每阶段结束后汇总结果）。
• 游戏开发中等待所有玩家准备就绪。
• 并行算法中的同步点。

核心 API

public CyclicBarrier(int parties); public CyclicBarrier(int parties, Runnable barrierAction); // 所有线程到达后执行的动作 public int await() throws InterruptedException, BrokenBarrierException; public int await(long timeout, TimeUnit unit) throws ...; public void reset(); // 重置屏障（可能中断当前等待）

使用示例

public class CyclicBarrierExample { public static void main(String[] args) { CyclicBarrier barrier = new CyclicBarrier(3, () -> { System.out.println("所有线程已到达屏障，执行汇总操作！"); }); for (int i = 0; i < 3; i++) { new Thread(() -> { try { System.out.println(Thread.currentThread().getName() + " 第一阶段完成"); barrier.await(); // 等待其他线程 System.out.println(Thread.currentThread().getName() + " 第二阶段开始"); } catch (Exception e) { e.printStackTrace(); } }).start(); } } }

输出：

Thread-1 第一阶段完成 Thread-2 第一阶段完成 Thread-0 第一阶段完成 所有线程已到达屏障，执行汇总操作！ Thread-0 第二阶段开始 Thread-1 第二阶段开始 Thread-2 第二阶段开始

应用场景示例

多轮并行计算：

CyclicBarrier barrier = new CyclicBarrier(4, this::aggregateResults); for (int i = 0; i < 4; i++) { executor.submit(() -> { while (round < MAX_ROUNDS) { computePartialResult(); barrier.await(); // 同步进入下一轮 round++; } }); }

底层原理

CyclicBarrier不基于 AQS，而是使用ReentrantLock+Condition实现：（Condition 原理详见Java并发编程：Lock原理详解）

• 每次调用await()，线程进入条件队列等待。
• 当最后一个线程到达时，唤醒所有等待线程，并执行barrierAction（如有）。
• 完成后自动重置计数器，支持重复使用。

若某个线程在等待期间被中断或超时，屏障将进入“损坏”（broken）状态，后续调用会抛出BrokenBarrierException。

CyclicBarrier 与 CountDownLatch 的区别

Phaser（JDK 7+）：灵活的阶段同步器

Phaser新增于 JDK 7，是对CyclicBarrier和CountDownLatch的强大扩展，支持动态注册参与者、多阶段同步和分层结构，适用于更复杂的并发协调场景。

使用场景

• 多阶段并行任务（如 Map-Reduce 的 map → shuffle → reduce）。
• 动态增减工作线程（例如任务中途加入新 worker）。
• 大规模并行计算中需要分组同步（通过分层 Phaser 实现）。
• 替代多个CyclicBarrier或CountDownLatch的组合逻辑。

核心 API

// 构造方法 public Phaser(); // 无初始参与者 public Phaser(int parties); // 指定初始参与者数量 public Phaser(Phaser parent); // 构建父子分层结构 // 核心方法 public int register(); // 注册一个新参与者，返回当前阶段号 public int arrive(); // 到达屏障，不等待 public int arriveAndAwaitAdvance(); // 到达并等待其他参与者 public int arriveAndDeregister(); // 到达并注销自己（退出后续阶段） public boolean isTerminated(); // 是否已终止（当参与者数为0时终止）

使用示例

动态多阶段任务：

public class PhaserExample { public static void main(String[] args) { Phaser phaser = new Phaser(0); // 初始无参与者 // 启动3个任务线程 for (int i = 0; i < 3; i++) { phaser.register(); // 注册参与者 new Thread(() -> { try { for (int phase = 0; phase < 3; phase++) { System.out.println(Thread.currentThread().getName() + " 完成阶段 " + phase); // 到达屏障并等待其他线程 phaser.arriveAndAwaitAdvance(); } // 任务结束，注销自己 phaser.arriveAndDeregister(); } catch (Exception e) { e.printStackTrace(); } }).start(); } // 主线程等待所有阶段完成 while (!phaser.isTerminated()) { Thread.yield(); } System.out.println("所有阶段已完成！"); } }

输出：

Thread-1 完成阶段 0 Thread-2 完成阶段 0 Thread-0 完成阶段 0 Thread-0 完成阶段 1 Thread-2 完成阶段 1 Thread-1 完成阶段 1 Thread-2 完成阶段 2 Thread-1 完成阶段 2 Thread-0 完成阶段 2 所有阶段已完成！

高级特性：分层 Phaser

当参与者数量极大（如数千线程）时，单一Phaser的同步开销会很高。此时可构建树状分层结构：

Phaser root = new Phaser(); Phaser group1 = new Phaser(root); // 子 Phaser，父为 root Phaser group2 = new Phaser(root); // group1 内部线程只与同组同步，最终由 root 汇总 group1.register(); new Thread(() -> { group1.arriveAndAwaitAdvance(); // 仅与 group1 同步 }).start();

这种结构能显著减少锁竞争，提升大规模并发性能。、

应用场景示例

分阶段并行处理：

// 模拟三阶段数据处理：加载 → 转换 → 输出 Phaser phaser = new Phaser(0); List<Thread> workers = IntStream.range(0, 4) .mapToObj(i -> { phaser.register(); return new Thread(() -> { load(); phaser.arriveAndAwaitAdvance(); transform(); phaser.arriveAndAwaitAdvance(); output(); phaser.arriveAndAwaitAdvance(); phaser.arriveAndDeregister(); }); }) .peek(Thread::start) .collect(Collectors.toList()); // 等待全部完成 while (!phaser.isTerminated()) { Thread.yield(); }

底层原理

不同于CountDownLatch和Semaphore，Phaser未基于 AQS，而是自行实现同步机制，它的实现融合了自旋锁 + 队列管理：

• 状态字段：使用一个long类型的state字段，高位存阶段号（phase），低位存参与者数量（parties）。
• 自适应等待策略：
  • 参与者少时使用自旋等待（低延迟）。
  • 参与者多或等待时间长时，自动切换到阻塞队列（节省 CPU）。

Phaser 与 CyclicBarrier 的对比

使用建议：

• 若参与者数量固定且较少，用CyclicBarrier即可。
• 若需动态调整、多阶段、大规模，则优先选择Phaser。

Semaphore（JDK 5+）：信号量

使用场景

Semaphore用于控制同时访问特定资源的线程数量，比如：

• 数据库连接池（最多 N 个连接）。
• 接口限流（如每秒最多处理 100 个请求）。
• 模拟有限资源的分配（如停车场车位）。
• new Semaphore(1)实际上等价于synchronized或ReentrantLock。

核心 API

public Semaphore(int permits); public Semaphore(int permits, boolean fair); // 公平/非公平模式 public void acquire() throws InterruptedException; public void release(); public boolean tryAcquire(); // 非阻塞尝试获取

使用示例

public class SemaphoreExample { private static final Semaphore semaphore = new Semaphore(2); // 最多2个线程并发 public static void main(String[] args) { for (int i = 0; i < 5; i++) { new Thread(() -> { try { semaphore.acquire(); System.out.println(Thread.currentThread().getName() + " 获取许可，正在执行..."); Thread.sleep(2000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } finally { semaphore.release(); System.out.println(Thread.currentThread().getName() + " 释放许可"); } }).start(); } } }

输出：

Thread-0 获取许可，正在执行... Thread-1 获取许可，正在执行... Thread-0 释放许可 Thread-2 获取许可，正在执行... Thread-1 释放许可 Thread-3 获取许可，正在执行... Thread-2 释放许可 Thread-3 释放许可 Thread-4 获取许可，正在执行... Thread-4 释放许可

应用场景示例

API 限流：

private final Semaphore rateLimiter = new Semaphore(100); // 每秒最多100请求 public void handleRequest() { if (rateLimiter.tryAcquire()) { try { process(); } finally { rateLimiter.release(); } } else { throw new RuntimeException("请求过于频繁"); } }

底层原理

同样基于AQS：（AQS 原理详见Java并发编程：Lock原理详解）

• state表示可用许可数量。
• acquire()尝试减少state，若不足则入队等待。
• release()增加state并唤醒等待线程。
• 支持公平模式（按请求顺序分配许可）。

Exchanger（JDK 5+）：数据交换器

使用场景

Exchanger专为两个线程之间交换数据而设计，典型场景如：

• 双缓冲机制（一个线程写入缓冲区 A，另一个写入 B，满后交换）。
• 生产者与消费者高效传递中间结果。
• 日志收集器中交换日志缓冲区。

核心 API

public V exchange(V x) throws InterruptedException; public V exchange(V x, long timeout, TimeUnit unit) throws ...;

示例代码

public class ExchangerExample { public static void main(String[] args) { Exchanger<String> exchanger = new Exchanger<>(); new Thread(() -> { try { String data = "Thread-A 的数据"; System.out.println("A 准备交换: " + data); String received = exchanger.exchange(data); System.out.println("A 收到: " + received); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } }).start(); new Thread(() -> { try { Thread.sleep(1000); // 模拟延迟 String data = "Thread-B 的数据"; System.out.println("B 准备交换: " + data); String received = exchanger.exchange(data); System.out.println("B 收到: " + received); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } }).start(); } }

输出：

A 准备交换: Thread-A 的数据 B 准备交换: Thread-B 的数据 B 收到: Thread-A 的数据 A 收到: Thread-B 的数据

应用场景示例

双缓冲日志：

Exchanger<List<LogEntry>> exchanger = new Exchanger<>(); List<LogEntry> bufferA = new ArrayList<>(1000); List<LogEntry> bufferB = new ArrayList<>(1000); // 写线程不断填充 bufferA // 写满后与读线程交换，读线程异步刷盘

底层原理

Exchanger内部使用Slot 或 Node 结构暂存数据，通过CAS + 自旋实现高效交换：

• 第一个调用exchange()的线程将数据放入槽位并等待。
• 第二个线程到来后，取出对方数据，放入自己的数据，完成交换。
• 若超时或中断，会清理槽位并抛出异常。

注意：仅支持两个线程配对交换，第三个线程调用exchange()会一直阻塞直到有新的配对。

工具类对比总结