16. CountDownLatch：等待一批 PDF 全部处理完成
发布于 • 阅读量 0
上一节我用 Semaphore + join 写了一个 PDF 并发处理版本。
Semaphore 负责控制同一时间最多几个任务进入处理逻辑，join 负责让 main 等所有子线程结束。
这种写法能跑，但如果线程很多，用 join() 一个个等，代码会显得有点笨。
这时候就可以用 CountDownLatch。
我对它的理解比较简单：
CountDownLatch 就是一个倒计时计数器。
一开始设置有多少个任务要完成；
每完成一个任务，就 countDown 一次；
主线程 await 等着；
等计数变成 0，主线程继续执行。
它很适合表达这种需求：
等这一批 PDF 全部处理完成以后，再继续往下走。
先写一个最简单的例子
新建类：
com.succos.thread.CountDownLatchDemo
代码如下：
package com.succos.thread;
import java.util.concurrent.CountDownLatch;
public class CountDownLatchDemo {
public static void main(String[] args) throws InterruptedException {
CountDownLatch latch = new CountDownLatch(5);
for (int i = 1; i <= 5; i++) {
int taskId = i;
new Thread(() -> {
try {
System.out.println("任务 " + taskId
+ " 开始执行，线程："
+ Thread.currentThread().getName());
Thread.sleep(3000);
System.out.println("任务 " + taskId
+ " 执行完成，线程："
+ Thread.currentThread().getName());
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
latch.countDown();
System.out.println("任务 " + taskId
+ " countDown，剩余数量："
+ latch.getCount());
}
}, "task-thread-" + taskId).start();
}
System.out.println("main 等待所有任务完成");
latch.await();
System.out.println("所有任务都完成了，main 继续执行");
}
}
这里一开始写的是：
CountDownLatch latch = new CountDownLatch(5);
意思是：
我现在有 5 个任务需要等待。
每个任务结束时调用：
latch.countDown();
计数就减 1。
main 线程调用：
latch.await();
会在这里等待，直到计数变成 0。
执行流程
这段代码的执行流程大概是这样：
main 创建 CountDownLatch，初始值是 5；
main 启动 5 个子线程；
main 执行 latch.await()，开始等待；
任务1执行完，countDown，计数变成 4；
任务2执行完，countDown，计数变成 3；
任务3执行完，countDown，计数变成 2；
任务4执行完，countDown，计数变成 1；
任务5执行完，countDown，计数变成 0；
计数为 0 后，main 从 await() 继续往下执行。
这个过程和 join() 有点像，但表达方式不一样。
join() 是等某几个具体线程结束。
CountDownLatch 是等某个数量的任务完成。
我觉得 CountDownLatch 更适合批量任务，因为它关心的是“任务完成数量”，不是每个线程对象本身。
countDown 一定要放在 finally 里
这点和前面 Semaphore.release() 很像。
我一般会把：
latch.countDown();
放在 finally 里。
原因很简单：不管任务成功还是失败，都应该告诉主线程“这个任务已经结束了”。
如果只在成功时 countDown()，一旦某个任务中间抛异常，计数就不会减到 0，main 线程可能会一直卡在：
latch.await();
比如这种写法就有风险：
try {
// 处理 PDF
latch.countDown();
} catch (Exception e) {
e.printStackTrace();
}
如果处理 PDF 时抛异常，countDown() 根本不会执行。
更稳一点的写法是：
try {
// 处理 PDF
} catch (Exception e) {
// 记录失败
} finally {
latch.countDown();
}
这样至少能保证主线程不会一直等。
用 CountDownLatch 等待 PDF 任务
现在把它放到 PDF 场景里。
目标是：
input 目录下有多少个 PDF，CountDownLatch 初始值就是多少；
每个 PDF 处理完成后 countDown；
main 线程 await；
等所有 PDF 都结束后，再打印总耗时。
新建类：
com.succos.thread.CountDownLatchPdfDemo
代码如下：
package com.succos.thread;
import java.io.File;
import java.util.concurrent.CountDownLatch;
public class CountDownLatchPdfDemo {
public static void main(String[] args) throws InterruptedException {
File inputDir = new File("input");
File[] files = inputDir.listFiles(file ->
file.isFile() && file.getName().toLowerCase().endsWith(".pdf")
);
if (files == null || files.length == 0) {
System.out.println("input 目录下没有 PDF 文件");
return;
}
CountDownLatch latch = new CountDownLatch(files.length);
long start = System.currentTimeMillis();
for (File file : files) {
Thread thread = new Thread(() -> {
try {
System.out.println(Thread.currentThread().getName()
+ " 开始处理："
+ file.getName());
Thread.sleep(3000);
System.out.println(Thread.currentThread().getName()
+ " 处理完成："
+ file.getName());
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
latch.countDown();
System.out.println(Thread.currentThread().getName()
+ " 完成计数，剩余任务："
+ latch.getCount());
}
}, "pdf-thread-" + file.getName());
thread.start();
}
System.out.println("main 等待所有 PDF 处理完成");
latch.await();
long end = System.currentTimeMillis();
System.out.println("--------------------------------");
System.out.println("全部 PDF 处理完成");
System.out.println("总文件数：" + files.length);
System.out.println("总耗时：" + (end - start) + " ms");
}
}
这段代码里没有手动保存线程对象，也没有逐个 join()。
主线程只需要等：
latch.await();
等所有任务都 countDown() 完，主线程就继续。
CountDownLatch 和 join 的区别
这个地方我觉得可以稍微对比一下。
join() 是这样：
thread1.join();
thread2.join();
thread3.join();
它等的是线程对象。
如果我要等很多线程，就要保存这些线程，然后一个个调用 join()。
CountDownLatch 是这样：
CountDownLatch latch = new CountDownLatch(taskCount);
latch.await();
它等的是任务数量。
每个任务结束时自己调用：
latch.countDown();
所以用 CountDownLatch 时，main 不一定关心具体有哪些线程。
它只关心：
这一批任务有没有全部完成。
在批量任务场景里，这种表达更自然。
CountDownLatch 不能重复使用
CountDownLatch 有一个特点：它是一次性的。
比如：
CountDownLatch latch = new CountDownLatch(3);
当它的计数从 3 减到 0 以后，这个 latch 就不能重置回 3 再用一次。
如果下一批任务还要等，就要重新创建一个新的 CountDownLatch。
这一点和后面要学的 CyclicBarrier 不一样。
CyclicBarrier 是可以循环使用的。
但 PDF 批量处理这种场景，一般就是一批任务创建一个 latch，用完就结束，所以 CountDownLatch 很合适。
和 Semaphore 配合使用
上一节我用 Semaphore 限制同时处理的 PDF 数量。
这一节用 CountDownLatch 等所有任务结束。
这两个可以配合起来用。
比如：
Semaphore：同一时间最多 3 个 PDF 正在处理；
CountDownLatch：main 等所有 PDF 都处理完。
新建类：
com.succos.thread.SemaphoreCountDownLatchPdfDemo
代码如下：
package com.succos.thread;
import java.io.File;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Semaphore;
public class SemaphoreCountDownLatchPdfDemo {
private static final Semaphore semaphore = new Semaphore(3);
public static void main(String[] args) throws InterruptedException {
File inputDir = new File("input");
File[] files = inputDir.listFiles(file ->
file.isFile() && file.getName().toLowerCase().endsWith(".pdf")
);
if (files == null || files.length == 0) {
System.out.println("input 目录下没有 PDF 文件");
return;
}
CountDownLatch latch = new CountDownLatch(files.length);
long start = System.currentTimeMillis();
for (File file : files) {
Thread thread = new Thread(() -> {
boolean acquired = false;
try {
System.out.println(file.getName()
+ " 等待许可证，线程："
+ Thread.currentThread().getName());
semaphore.acquire();
acquired = true;
System.out.println(file.getName()
+ " 开始处理，线程："
+ Thread.currentThread().getName());
Thread.sleep(3000);
System.out.println(file.getName()
+ " 处理完成，线程："
+ Thread.currentThread().getName());
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
if (acquired) {
semaphore.release();
System.out.println(file.getName()
+ " 释放许可证，线程："
+ Thread.currentThread().getName());
}
latch.countDown();
System.out.println(file.getName()
+ " 完成计数，剩余任务："
+ latch.getCount());
}
}, "pdf-thread-" + file.getName());
thread.start();
}
System.out.println("main 等待所有 PDF 任务完成");
latch.await();
long end = System.currentTimeMillis();
System.out.println("--------------------------------");
System.out.println("全部 PDF 处理完成");
System.out.println("总文件数：" + files.length);
System.out.println("总耗时：" + (end - start) + " ms");
}
}
这个版本比上一节更完整一点。
因为它同时解决了两个问题：
并发数量限制；
主线程等待所有任务结束。
不过它仍然有一个老问题：每个 PDF 还是创建了一个线程。
所以它还不是最终方案。
后面用线程池以后，这个问题会解决得更好。
为什么 CountDownLatch 不负责控制并发数量
这里也要分清楚。
CountDownLatch 不会限制同一时间有多少个线程执行。
它只负责等待。
比如：
CountDownLatch latch = new CountDownLatch(files.length);
这只是告诉主线程：
你要等 files.length 个任务结束。
它不会控制任务怎么执行，也不会控制同时执行几个。
如果我不加 Semaphore，也不用线程池，直接每个 PDF 一个线程，那么这些线程还是会同时启动。
所以：
CountDownLatch 管等待；
Semaphore 管同时进入数量；
ThreadPoolExecutor 管线程和任务队列。
这几个工具解决的问题不一样。
不要混在一起。
真实 PDF 水印处理里怎么用
如果要接入真实 PDF 水印方法，可以在任务内部把 Thread.sleep(3000) 替换掉。
大概是这样：
try {
semaphore.acquire();
acquired = true;
File outputDir = new File("output");
if (!outputDir.exists()) {
outputDir.mkdirs();
}
String targetPath = outputDir.getAbsolutePath()
+ File.separator
+ file.getName().replace(".pdf", "-watermark.pdf");
FileItemContext fileItemContext = new FileItemContext();
fileItemContext.setSourcePath(file.getAbsolutePath());
fileItemContext.setTargetPath(targetPath);
fileItemContext.setWaterMakeText("上下文网");
PdfWatermarkService service = new PdfWatermarkService();
service.addWaterMakerOfPDF(fileItemContext);
System.out.println(file.getName() + " 处理完成，输出路径：" + targetPath);
} catch (Exception e) {
System.out.println(file.getName() + " 处理失败，原因：" + e.getMessage());
} finally {
if (acquired) {
semaphore.release();
}
latch.countDown();
}
需要加上：
import com.succos.dto.FileItemContext;
import com.succos.service.PdfWatermarkService;
这里还是那句话：latch.countDown() 要放在 finally。
因为不管 PDF 成功还是失败，这个任务都算结束了。
主线程等待的是“任务结束”，不是“任务成功”。
成功和失败可以另外统计。
CountDownLatch 适合什么场景
我觉得 CountDownLatch 很适合这种“一批任务”场景。
比如：
等 10 个 PDF 全部处理完；
等 5 个接口都返回；
等多个初始化任务完成后再启动服务；
等多个线程准备好以后再继续主流程。
它的表达很直接：
还有 N 件事没完成；
每完成一件，减一；
减到 0，继续。
所以它比手动管理一堆 join() 更适合批量任务。
这一节小结
这一节我主要记住几点：
1. CountDownLatch 是一个倒计时计数器；
2. new CountDownLatch(n) 表示要等待 n 个任务完成；
3. countDown() 表示一个任务完成，计数减一；
4. await() 表示当前线程等待计数变成 0；
5. countDown() 最好放在 finally 里；
6. CountDownLatch 只负责等待，不负责限制并发数量；
7. Semaphore 和 CountDownLatch 可以配合使用。
用一句话总结：
CountDownLatch 解决的是“主线程怎么知道这一批任务都结束了”的问题。
下一节看 CyclicBarrier。
它和 CountDownLatch 都有“等待”的味道，但使用场景不太一样。CountDownLatch 更像主线程等子线程，CyclicBarrier 更像多个线程互相等。
