1. 简介
在本文中，我们将介绍 CountDownLatch 类，并通过几个实际示例展示其用法。
简单来说，使用 CountDownLatch 可以让一个线程阻塞，直到其他线程完成指定的任务。
2. 在并发编程中的用途
CountDownLatch 内部维护一个计数器字段，我们可以按需对其进行递减。调用 await() 方法的线程会被阻塞，直到计数器被减至零。
例如，在进行并行处理时，我们可以将 CountDownLatch 的初始计数值设置为工作线程的数量。每个工作线程完成任务后调用 countDown() 方法，这样依赖线程（调用 await() 的线程）就会一直阻塞，直到所有工作线程都完成任务。
3. 等待线程池中的所有线程完成
让我们通过创建一个 Worker 类并使用 CountDownLatch 字段来演示这种模式，当任务完成后发出信号：
public class Worker implements Runnable {
private List<String> outputScraper;
private CountDownLatch countDownLatch;
public Worker(List<String> outputScraper, CountDownLatch countDownLatch) {
this.outputScraper = outputScraper;
this.countDownLatch = countDownLatch;
}
@Override
public void run() {
doSomeWork();
outputScraper.add("Counted down");
countDownLatch.countDown();
}
}
接下来，我们编写一个测试，验证 CountDownLatch 能否确保主线程等待所有 Worker 实例完成：
@Test
public void whenParallelProcessing_thenMainThreadWillBlockUntilCompletion()
throws InterruptedException {
List<String> outputScraper = Collections.synchronizedList(new ArrayList<>());
CountDownLatch countDownLatch = new CountDownLatch(5);
List<Thread> workers = Stream
.generate(() -> new Thread(new Worker(outputScraper, countDownLatch)))
.limit(5)
.collect(toList());
workers.forEach(Thread::start);
countDownLatch.await();
outputScraper.add("Latch released");
assertThat(outputScraper)
.containsExactly(
"Counted down",
"Counted down",
"Counted down",
"Counted down",
"Counted down",
"Latch released"
);
}
显然，“Latch released” 总是最后输出——因为它依赖于 CountDownLatch 的释放。
注意：如果我们不调用 await()，就无法保证线程执行的顺序，测试可能会随机失败。
4. 让一组线程同时开始执行
假设我们把上一个例子中的线程数量从 5 个增加到数千个，很可能前面启动的线程在后面线程调用 start() 之前就已经完成了处理。这会使我们难以复现某些并发问题，因为我们无法确保所有线程真正并行运行。
为了解决这个问题，我们可以改变 CountDownLatch 的使用方式：不再让主线程等待子线程完成，而是让每个子线程在开始处理前先等待，直到所有线程都已启动。
修改 run() 方法，使其在执行前先阻塞：
public class WaitingWorker implements Runnable {
private List<String> outputScraper;
private CountDownLatch readyThreadCounter;
private CountDownLatch callingThreadBlocker;
private CountDownLatch completedThreadCounter;
public WaitingWorker(
List<String> outputScraper,
CountDownLatch readyThreadCounter,
CountDownLatch callingThreadBlocker,
CountDownLatch completedThreadCounter) {
this.outputScraper = outputScraper;
this.readyThreadCounter = readyThreadCounter;
this.callingThreadBlocker = callingThreadBlocker;
this.completedThreadCounter = completedThreadCounter;
}
@Override
public void run() {
readyThreadCounter.countDown();
try {
callingThreadBlocker.await();
doSomeWork();
outputScraper.add("Counted down");
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
completedThreadCounter.countDown();
}
}
}
现在，我们修改测试方法，使其先等待所有 Worker 启动，然后统一放行，再等待它们全部完成：
@Test
public void whenDoingLotsOfThreadsInParallel_thenStartThemAtTheSameTime()
throws InterruptedException {
List<String> outputScraper = Collections.synchronizedList(new ArrayList<>());
CountDownLatch readyThreadCounter = new CountDownLatch(5);
CountDownLatch callingThreadBlocker = new CountDownLatch(1);
CountDownLatch completedThreadCounter = new CountDownLatch(5);
List<Thread> workers = Stream
.generate(() -> new Thread(new WaitingWorker(
outputScraper, readyThreadCounter, callingThreadBlocker, completedThreadCounter)))
.limit(5)
.collect(toList());
workers.forEach(Thread::start);
readyThreadCounter.await();
outputScraper.add("Workers ready");
callingThreadBlocker.countDown();
completedThreadCounter.await();
outputScraper.add("Workers complete");
assertThat(outputScraper)
.containsExactly(
"Workers ready",
"Counted down",
"Counted down",
"Counted down",
"Counted down",
"Counted down",
"Workers complete"
);
}
这种模式在尝试复现并发 bug 时非常有用，因为它可以强制成千上万个线程同时执行某段逻辑。
5. 提前终止 CountDownLatch
有时，工作线程可能在调用 countDown() 之前因错误而提前终止。这会导致计数器永远无法归零，从而使 await() 永远阻塞：
@Override
public void run() {
if (true) {
throw new RuntimeException("Oh dear, I'm a BrokenWorker");
}
countDownLatch.countDown();
outputScraper.add("Counted down");
}
让我们修改之前的测试，使用 BrokenWorker 来演示 await() 如何无限期阻塞：
@Test
public void whenFailingToParallelProcess_thenMainThreadShouldGetNotGetStuck()
throws InterruptedException {
List<String> outputScraper = Collections.synchronizedList(new ArrayList<>());
CountDownLatch countDownLatch = new CountDownLatch(5);
List<Thread> workers = Stream
.generate(() -> new Thread(new BrokenWorker(outputScraper, countDownLatch)))
.limit(5)
.collect(toList());
workers.forEach(Thread::start);
countDownLatch.await();
}
显然，这不是我们期望的行为——与其无限阻塞，不如让程序继续执行更好。
为了解决这个问题，我们可以在调用 await() 时添加超时参数：
boolean completed = countDownLatch.await(3L, TimeUnit.SECONDS);
assertThat(completed).isFalse();
如上所示，测试最终会超时，await() 将返回 false。
6. 结论
在本快速指南中，我们演示了如何使用 CountDownLatch 阻塞一个线程，直到其他线程完成某些处理任务。
此外，我们还展示了它如何用于调试并发问题——通过确保多个线程真正并行执行。