CountDownLatch 并发编程中的同步利器

CountDownLatch 并发编程中的同步利器

​

​ 在多线程并发编程中，线程同步是一个永恒的话题。当我们需要等待多个线程完成某些操作后再继续执行，或者需要等待某个条件满足后才能开始执行，CountDownLatch 便成为了一把利器。它就像一个倒计时器，只有当计数归零时，等待的线程才能继续执行。

一、CountDownLatch 基础概念

1.1 什么是 CountDownLatch？

CountDownLatch 是 Java 并发包（java.util.concurrent）中的一个同步工具类，它允许一个或多个线程等待一系列指定操作的完成。CountDownLatch 的核心思想是维护一个计数器，每完成一个操作，计数器减一，当计数器归零时，等待的线程被释放继续执行。

这听起来有点抽象，我们可以通过一个简单的比喻来理解：想象你是一个项目经理，手下有5个开发人员，每人负责一个模块。只有当这5个模块都开发完成后，整个项目才能进入测试阶段。CountDownLatch 就相当于一个跟踪这5个模块开发进度的工具，每完成一个模块，计数器减一，当计数器变为0时，测试团队就可以开始工作了。

1.2 CountDownLatch 的核心方法

CountDownLatch 主要有两个核心方法：

• countDown()：递减计数器，表示一个操作已完成
• await()：等待计数器归零，如果计数器大于0，则当前线程进入等待状态，此外还有一些其他实用方法：
  • await(long timeout, TimeUnit unit)：等待计数器归零，但最多等待指定的时间
  • getCount()：获取当前计数器的值

1.3 基本使用示例

下面是一个简单的示例，演示 CountDownLatch 的基本使用：

import java.util.concurrent.CountDownLatch;public class CountDownLatchDemo {public static void main(String[] args) throws InterruptedException {// 创建一个计数器为5的CountDownLatchCountDownLatch latch = new CountDownLatch(5);System.out.println("主线程开始等待所有工作线程完成...");// 创建并启动5个工作线程for (int i = 0; i < 5; i++) {final int workerId = i + 1;new Thread(() -> {try {// 模拟工作线程执行任务System.out.println("工作线程" + workerId + "开始执行任务");Thread.sleep((long) (Math.random() * 1000));System.out.println("工作线程" + workerId + "完成任务");// 任务完成，计数器减一latch.countDown();} catch (InterruptedException e) {Thread.currentThread().interrupt();}}).start();}// 主线程等待所有工作线程完成latch.await();System.out.println("所有工作线程已完成，主线程继续执行");}
}

在这个例子中，主线程创建了5个工作线程，然后调用latch.await()等待所有工作线程完成。每个工作线程完成任务后调用latch.countDown()将计数器减一。当所有5个工作线程都完成任务后，计数器归零，主线程从 await() 方法返回并继续执行。

二、CountDownLatch 实战应用

2.1 应用场景一：并行任务协调

CountDownLatch 最常见的应用场景是协调并行任务的执行。当一个复杂任务可以分解为多个独立的子任务并行执行时，我们可以使用 CountDownLatch 来等待所有子任务完成后再进行下一步操作。

下面是一个模拟并行加载系统模块的例子：

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;public class SystemInitializer {public static void main(String[] args) throws InterruptedException {// 需要初始化的模块列表List<String> modules = Arrays.asList("数据库连接模块", "缓存服务模块", "消息队列模块", "用户认证模块", "日志服务模块");int moduleCount = modules.size();// 创建与模块数量相同的计数器CountDownLatch latch = new CountDownLatch(moduleCount);System.out.println("系统启动中，等待所有模块初始化...");// 创建固定大小的线程池ExecutorService executor = Executors.newFixedThreadPool(Math.min(moduleCount, 3));// 启动时间long startTime = System.currentTimeMillis();// 为每个模块创建初始化任务for (String module : modules) {executor.submit(() -> {try {initModule(module);} finally {// 模块初始化完成，计数器减一latch.countDown();System.out.println("模块 [" + module + "] 初始化完成，还剩 " + latch.getCount() + " 个模块");}});}// 等待所有模块初始化完成latch.await();// 计算总耗时long endTime = System.currentTimeMillis();System.out.println("所有模块初始化完成，系统启动成功！总耗时: " + (endTime - startTime) + "ms");// 关闭线程池executor.shutdown();}private static void initModule(String moduleName) {System.out.println("开始初始化模块: " + moduleName);try {// 模拟模块初始化耗时long sleepTime = (long) (Math.random() * 2000 + 1000);Thread.sleep(sleepTime);System.out.println("模块 [" + moduleName + "] 初始化耗时: " + sleepTime + "ms");} catch (InterruptedException e) {Thread.currentThread().interrupt();System.out.println("模块 [" + moduleName + "] 初始化被中断");}}
}

在这个例子中，我们使用 CountDownLatch 来协调系统各个模块的初始化过程。系统启动时需要初始化多个模块，每个模块的初始化可以并行进行，但只有当所有模块都初始化完成后，系统才能正常运行。

2.2 应用场景二：多阶段并发处理

有时我们需要将一个大型任务分为多个阶段，每个阶段都需要多个线程协同完成，且下一阶段必须等待上一阶段完全完成后才能开始。这种情况下，我们可以为每个阶段创建一个 CountDownLatch。

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;public class MultiPhaseTask {public static void main(String[] args) throws InterruptedException {int threadCount = 3;// 创建线程池ExecutorService executor = Executors.newFixedThreadPool(threadCount);// 第一阶段的计数器CountDownLatch phase1Latch = new CountDownLatch(threadCount);// 第二阶段的计数器CountDownLatch phase2Latch = new CountDownLatch(threadCount);// 第三阶段的计数器CountDownLatch phase3Latch = new CountDownLatch(threadCount);System.out.println("开始执行多阶段任务...");// 启动所有工作线程for (int i = 0; i < threadCount; i++) {final int workerId = i + 1;executor.submit(() -> {try {// 第一阶段：数据准备System.out.println("工作线程" + workerId + "开始第一阶段：数据准备");Thread.sleep((long) (Math.random() * 1000));System.out.println("工作线程" + workerId + "完成第一阶段");phase1Latch.countDown();// 等待所有线程完成第一阶段phase1Latch.await();System.out.println("所有线程完成第一阶段，工作线程" + workerId + "开始第二阶段");// 第二阶段：数据处理Thread.sleep((long) (Math.random() * 1500));System.out.println("工作线程" + workerId + "完成第二阶段");phase2Latch.countDown();// 等待所有线程完成第二阶段phase2Latch.await();System.out.println("所有线程完成第二阶段，工作线程" + workerId + "开始第三阶段");// 第三阶段：结果汇总Thread.sleep((long) (Math.random() * 800));System.out.println("工作线程" + workerId + "完成第三阶段");phase3Latch.countDown();} catch (InterruptedException e) {Thread.currentThread().interrupt();}});}// 等待所有阶段完成phase3Latch.await();System.out.println("所有阶段都已完成，任务结束");// 关闭线程池executor.shutdown();}
}

在这个例子中，我们将任务分为三个阶段：数据准备、数据处理和结果汇总。每个阶段都需要多个线程共同完成，且只有当所有线程都完成当前阶段后，才能进入下一阶段。

2.3 应用场景三：并发测试

CountDownLatch 在性能测试中也有重要应用，特别是需要模拟大量并发请求的场景。下面是一个模拟并发请求的例子：

import java.util.concurrent.CountDownLatch;startLatch.await();// 记录请求开始时间long startTime = System.currentTimeMillis();System.out.println("请求" + requestId + "开始执行");// 模拟发送HTTP请求boolean success = sendHttpRequest(requestId);// 记录请求结束时间并计算耗时long endTime = System.currentTimeMillis();long duration = endTime - startTime;// 根据请求结果更新计数器if (success) {successCount.incrementAndGet();System.out.println("请求" + requestId + "成功，耗时: " + duration + "ms");} else {failureCount.incrementAndGet();System.out.println("请求" + requestId + "失败，耗时: " + duration + "ms");}} catch (InterruptedException e) {Thread.currentThread().interrupt();} finally {// 请求完成，计数器减一endLatch.countDown();}});}// 记录测试开始时间long testStartTime = System.currentTimeMillis();// 发出开始信号，所有线程同时开始执行System.out.println("所有请求准备就绪，开始并发测试...");startLatch.countDown();// 等待所有请求完成boolean allCompleted = endLatch.await(30, TimeUnit.SECONDS);// 记录测试结束时间long testEndTime = System.currentTimeMillis();long totalDuration = testEndTime - testStartTime;// 输出测试结果System.out.println("\n并发测试完成！");if (!allCompleted) {System.out.println("警告：有些请求在超时时间内未完成");}System.out.println("总耗时: " + totalDuration + "ms");System.out.println("成功请求数: " + successCount.get());System.out.println("失败请求数: " + failureCount.get());System.out.println("平均响应时间: " + (totalDuration / concurrentRequests) + "ms");// 关闭线程池executor.shutdown();}// 模拟发送HTTP请求private static boolean sendHttpRequest(int requestId) {try {// 模拟请求处理时间，随机100-500msThread.sleep((long) (Math.random() * 400 + 100));// 模拟偶尔的请求失败，约5%的失败率return Math.random() > 0.05;} catch (InterruptedException e) {Thread.currentThread().interrupt();return false;}}
}

在这个例子中，我们使用两个 CountDownLatch：

• startLatch 用于控制所有线程同时开始，模拟真实的并发请求场景。
• endLatch 用于等待所有请求完成，以便统计整体测试结果。

这种方式可以准确测量系统在高并发情况下的性能表现，是性能测试的常用手段。

三、源码解析

了解 CountDownLatch 的内部实现原理，有助于我们更好地使用它。CountDownLatch 的实现基于 AQS（AbstractQueuedSynchronizer）框架，这也是 Java 并发包中许多同步器的基础。

3.1 CountDownLatch 的构造与字段

先看 CountDownLatch 的构造函数：

public CountDownLatch(int count) {if (count < 0) throw new IllegalArgumentException("count < 0");this.sync = new Sync(count);
}

CountDownLatch 内部维护了一个 Sync 实例，它是 AQS 的子类：

private static final class Sync extends AbstractQueuedSynchronizer {private static final long serialVersionUID = 4982264981922014374L;Sync(int count) {setState(count);}int getCount() {return getState();}protected int tryAcquireShared(int acquires) {return (getState() == 0) ? 1 : -1;}protected boolean tryReleaseShared(int releases) {// Decrement count; signal when transition to zerofor (;;) {int c = getState();if (c == 0)return false;int nextc = c-1;if (compareAndSetState(c, nextc))return nextc == 0;}}
}

Sync 类直接使用 AQS 的 state 变量作为计数器。构造函数中，将 state 设置为指定的计数值。

3.2 核心方法实现

3.2.1 await 方法

public void await() throws InterruptedException {sync.acquireSharedInterruptibly(1);
}

await() 方法调用 AQS 的 acquireSharedInterruptibly()，这个方法会调用 Sync 中重写的 tryAcquireShared() 方法。只有当计数器（state）为 0 时，tryAcquireShared() 才会返回正值，表示可以获取共享锁，否则线程将被阻塞。

3.2.2 countDown 方法

public void countDown() {sync.releaseShared(1);
}

countDown() 方法调用 AQS 的 releaseShared()，这个方法会调用 Sync 中重写的 tryReleaseShared() 方法。tryReleaseShared() 通过 CAS 操作减少计数器的值，当计数器变为 0 时，会返回 true，此时 AQS 会释放所有等待的线程。

3.3 工作流程图解

CountDownLatch 的工作流程可以简化为以下几个步骤：

1. 创建 CountDownLatch 对象，初始化计数器

2. 等待线程调用 await() 方法，如果计数器大于 0，线程将被阻塞

3. 工作线程完成任务后调用 countDown() 方法，计数器减一

4. 当计数器减至 0 时，AQS 会释放所有等待的线程，它们从 await() 方法返回继续执行

四、CountDownLatch 的高级用法

4.1 带超时的等待

有时我们不希望无限期地等待所有操作完成，这时可以使用带超时参数的 await() 方法：

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;public class CountDownLatchWithTimeout {public static void main(String[] args) {// 创建计数器为3的CountDownLatchCountDownLatch latch = new CountDownLatch(3);System.out.println("主线程开始等待工作线程完成...");// 启动工作线程for (int i = 0; i < 3; i++) {final int workerId = i + 1;new Thread(() -> {try {System.out.println("工作线程" + workerId + "开始执行");// 模拟工作线程耗时，第三个线程会故意执行很长时间long sleepTime = (workerId == 3) ? 5000 : 1000;Thread.sleep(sleepTime);System.out.println("工作线程" + workerId + "完成执行");latch.countDown();} catch (InterruptedException e) {Thread.currentThread().interrupt();}}).start();}try {// 等待，但最多等待2秒boolean completed = latch.await(2, TimeUnit.SECONDS);if (completed) {System.out.println("所有工作线程在超时前完成了任务");} else {System.out.println("等待超时，但仍有" + latch.getCount() + "个任务未完成");System.out.println("主线程将继续执行，不再等待未完成的任务");}} catch (InterruptedException e) {Thread.currentThread().interrupt();System.out.println("主线程等待被中断");}System.out.println("主线程继续执行其他操作");}
}

在这个例子中，我们给 await() 方法设置了2秒的超时时间。由于第三个工作线程需要5秒才能完成，因此主线程会在等待2秒后继续执行，即使并非所有工作线程都已完成。

4.2 结合 CyclicBarrier 实现复杂同步

在更复杂的场景中，我们可能需要结合使用 CountDownLatch 和 CyclicBarrier 来实现多阶段、多方向的同步。例如，在一个分布式计算任务中，既需要等待所有工作节点准备就绪后才开始计算，又需要等待所有计算结果返回后才能进行汇总。

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;public class ComplexSynchronization {public static void main(String[] args) throws InterruptedException {int workerCount = 5;// 用于等待所有工作线程完成初始化CountDownLatch initLatch = new CountDownLatch(workerCount);// 用于等待所有工作线程完成计算CountDownLatch computeLatch = new CountDownLatch(workerCount);// 用于同步所有工作线程开始计算的时刻CyclicBarrier computeBarrier = new CyclicBarrier(workerCount, () -> {System.out.println("所有工作线程已就绪，开始并行计算");});ExecutorService executor = Executors.newFixedThreadPool(workerCount);System.out.println("主线程开始初始化工作线程...");for (int i = 0; i < workerCount; i++) {final int workerId = i + 1;executor.submit(() -> {try {// 初始化阶段System.out.println("工作线程" + workerId + "开始初始化");Thread.sleep((long) (Math.random() * 1000));System.out.println("工作线程" + workerId + "初始化完成");// 初始化完成，计数器减一initLatch.countDown();// 等待所有线程初始化完成，并同步开始计算computeBarrier.await();// 计算阶段System.out.println("工作线程" + workerId + "开始计算");Thread.sleep((long) (Math.random() * 2000));System.out.println("工作线程" + workerId + "计算完成");// 计算完成，计数器减一computeLatch.countDown();} catch (Exception e) {e.printStackTrace();}});}// 等待所有工作线程初始化完成System.out.println("主线程等待所有工作线程初始化完成...");initLatch.await();System.out.println("所有工作线程初始化完成，等待计算结果...");// 等待所有工作线程计算完成computeLatch.await();System.out.println("所有工作线程计算完成，开始汇总结果");// 模拟结果汇总Thread.sleep(500);System.out.println("结果汇总完成，任务结束");executor.shutdown();}
}

在这个例子中，我们使用了：

• initLatch（CountDownLatch）等待所有工作线程完成初始化。
• computeBarrier（CyclicBarrier）同步所有工作线程开始计算的时刻。
• computeLatch（CountDownLatch）等待所有工作线程完成计算。

这种组合使用可以实现更复杂的多阶段同步场景。

4.3 注意事项与最佳实践

在使用 CountDownLatch 时，有一些注意事项和最佳实践：

1. 计数器不可重置：CountDownLatch 的计数器一旦归零就不能重新设置，如果需要重复使用，应考虑 CyclicBarrier。

2. 防止计数器错误：确保 countDown() 方法被正确调用，特别是在有异常的情况下，通常应该在finally块中调用。

3. 避免死锁：如果某些线程未能调用 countDown()，等待的线程将永远阻塞，应考虑使用带超时的await()方法。

4. 资源释放：在所有线程都完成后，及时释放资源，如关闭线程池。

5. 与线程池结合：通常应将 CountDownLatch 与线程池结合使用，而不是直接创建大量线程。

// 错误的使用方式（没有在finally中调用countDown）
public void wrongWay() {CountDownLatch latch = new CountDownLatch(1);executor.submit(() -> {try {// 如果这里抛出异常，countDown不会被调用doSomething();latch.countDown();} catch (Exception e) {e.printStackTrace();}});
}// 正确的使用方式
public void rightWay() {CountDownLatch latch = new CountDownLatch(1);executor.submit(() -> {try {doSomething();} catch (Exception e) {e.printStackTrace();} finally {// 确保即使出现异常，countDown也会被调用latch.countDown();}});
}

五、CountDownLatch 与其他同步工具的对比

为了更全面地理解 CountDownLatch，我们将它与其他常见的同步工具进行对比：

5.1 CountDownLatch vs CyclicBarrier

• 重用性：CountDownLatch 的计数器减到0后就不能再用，而 CyclicBarrier 可以通过 reset() 方法重置，可以重复使用。
• 触发机制：CountDownLatch 是一个线程等待多个线程，或多个线程等待一个线程；CyclicBarrier 是多个线程互相等待，直到所有线程都到达屏障点。
• 计数方式：CountDownLatch 的计数器只能减少，CyclicBarrier 的计数器可以减少也可以重置。

5.2 CountDownLatch vs Semaphore

• 作用范围：CountDownLatch 主要用于等待事件；Semaphore 用于控制对资源的并发访问数量。
• 计数方向：CountDownLatch 计数器只能递减直至归零；Semaphore 的计数器可以递增递减，只要不超过设定的最大值。
• 使用场景：CountDownLatch 适用于一次性等待场景；Semaphore 适用于限制并发访问资源的场景。

5.3 CountDownLatch vs Join

• 灵活性：CountDownLatch 可以在任何时候调用 countDown()，而不必等待线程结束；Thread.join() 必须等待线程执行完成。
• 粒度：CountDownLatch 可以实现更细粒度的控制，一个线程可以多次 countDown()；join() 只能等待整个线程执行完毕。
• 中断处理：两者都支持中断，但 CountDownLatch 可以设置等待超时。

下面是一个简单的对比示例：

import java.util.concurrent.*;}private static void semaphoreExample() throws InterruptedException {// 创建只允许3个线程同时访问的信号量Semaphore semaphore = new Semaphore(3);ExecutorService executor = Executors.newFixedThreadPool(10);for (int i = 0; i < 10; i++) {final int id = i;executor.submit(() -> {try {System.out.println("线程" + id + "等待获取许可");semaphore.acquire();System.out.println("线程" + id + "获得许可，开始执行");Thread.sleep(1000);System.out.println("线程" + id + "执行完毕，释放许可");semaphore.release();} catch (InterruptedException e) {Thread.currentThread().interrupt();}});}// 给足够时间让任务执行Thread.sleep(5000);executor.shutdown();}private static void joinExample() throws InterruptedException {Thread[] threads = new Thread[3];for (int i = 0; i < threads.length; i++) {final int id = i;threads[i] = new Thread(() -> {try {System.out.println("线程" + id + "开始执行");Thread.sleep(1000 + id * 500);System.out.println("线程" + id + "执行完毕");} catch (InterruptedException e) {Thread.currentThread().interrupt();}});threads[i].start();}System.out.println("等待所有线程完成...");for (Thread thread : threads) {thread.join();}System.out.println("所有线程已完成！");}
}

这个例子展示了CountDownLatch、CyclicBarrier、Semaphore 和Thread.join() 的不同使用方式和特点，有助于理解它们各自的应用场景。

六、面试中的 CountDownLatch 问题解析

CountDownLatch 是Java并发编程面试中的高频话题。下面列出一些常见面试问题及其答案：

6.1 基础概念题

问题1：什么是CountDownLatch？它的主要用途是什么？

答：CountDownLatch是Java并发包中的同步工具类，允许一个或多个线程等待一系列指定操作的完成。它主要用于以下场景：

• 让主线程等待子线程完成再继续执行
• 实现多个线程之间的同步
• 控制并发测试中的并发量
• 多阶段并发任务的协调

它通过维护一个计数器来工作，每完成一个操作就减一，当计数器归零时释放所有等待的线程。

问题2：CountDownLatch与CyclicBarrier的区别？

答：主要区别有：

1. 重用性：CountDownLatch是一次性的，计数器归零后不能重置；CyclicBarrier可以通过reset()方法重置，可重复使用。

2. 计数方向：CountDownLatch计数器只能减少；CyclicBarrier的计数器是先减少，后自动重置。

3. 触发方式：CountDownLatch是调用countDown()方法；CyclicBarrier是调用await()方法。

4. 使用场景：CountDownLatch适用于一个或多个线程等待其他操作完成；CyclicBarrier适用于多个线程互相等待至某个状态，然后一起继续运行。

6.2 代码实现题

问题：实现一个高并发限流器，限制最大并发请求数为100，当请求处理完成后允许新的请求进入。

答：可以使用Semaphore实现最基本的限流功能，但结合CountDownLatch可以实现更加灵活的控制：

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicInteger;public class RequestLimiter {// 最大并发数限制private final Semaphore semaphore;// 已完成的请求计数private final AtomicInteger completedRequests = new AtomicInteger(0);// 用于等待所有请求处理完成private CountDownLatch completionLatch;public RequestLimiter(int maxConcurrent) {this.semaphore = new Semaphore(maxConcurrent);}// 开始一轮请求处理public void startBatch(int requestCount) {this.completionLatch = new CountDownLatch(requestCount);this.completedRequests.set(0);}// 处理请求public void processRequest(Runnable task) throws InterruptedException {// 获取许可semaphore.acquire();try {// 执行请求处理task.run();} finally {// 释放许可semaphore.release();// 完成一个请求completedRequests.incrementAndGet();// 计数器减一completionLatch.countDown();}}// 等待所有请求处理完成public void awaitCompletion() throws InterruptedException {completionLatch.await();}// 获取已完成的请求数public int getCompletedRequestCount() {return completedRequests.get();}public static void main(String[] args) throws InterruptedException {RequestLimiter limiter = new RequestLimiter(10);int totalRequests = 100;// 开始处理100个请求limiter.startBatch(totalRequests);// 提交请求for (int i = 0; i < totalRequests; i++) {final int requestId = i;new Thread(() -> {try {System.out.println("请求 " + requestId + " 等待处理");limiter.processRequest(() -> {try {// 模拟请求处理System.out.println("处理请求 " + requestId);Thread.sleep((long) (Math.random() * 500));} catch (InterruptedException e) {Thread.currentThread().interrupt();}});System.out.println("请求 " + requestId + " 处理完成");} catch (InterruptedException e) {Thread.currentThread().interrupt();}}).start();}// 等待所有请求处理完成limiter.awaitCompletion();System.out.println("所有请求处理完成，共处理: " + limiter.getCompletedRequestCount() + " 个请求");}
}

6.3 原理分析题

问题：CountDownLatch的内部实现原理是什么？它是如何实现线程等待和唤醒的？

答：CountDownLatch内部基于AQS（AbstractQueuedSynchronizer）框架实现：

1. 状态管理：使用AQS的state变量作为计数器

2. 等待机制：调用await()时，如果计数器不为0，则当前线程会被加入到AQS的等待队列中

3. 唤醒机制：调用countDown()时，计数器减1，当减至0时，AQS会释放所有等待的线程

4. 线程安全性：通过CAS（Compare And Swap）操作保证计数器更新的原子性

具体实现上，CountDownLatch包含一个继承自AQS的内部类Sync，它重写了tryAcquireShared和tryReleaseShared方法：

• tryAcquireShared：只有当计数器为0时才返回1，表示获取成功。
• tryReleaseShared：使用CAS操作安全地减少计数器，并在计数器变为0时返回true