14. Semaphore：控制同一时间最多处理几个 PDF
发布于 • 阅读量 1
前面几节讲的 synchronized 和 ReentrantLock，核心都是“加锁”。
它们主要解决的是这种问题：
多个线程同时修改同一份共享数据，怎么保证结果不乱。
比如 count++、共享状态、共享集合这些。
这一节开始看另一个工具：Semaphore。
它和锁有点像，但思路不太一样。
我现在更愿意把 Semaphore 理解成：
它不是为了保护某一个变量，而是为了控制同一时间最多允许几个线程进入某段代码。
放到 PDF 水印项目里，就是：
input 目录下可能有很多 PDF；
但我不想同时处理太多；
我希望同一时间最多只处理 3 个 PDF。
这个场景就很适合 Semaphore。
为什么需要 Semaphore
如果我现在有 100 个 PDF，最直接的写法可能是：
for (File file : files) {
new Thread(() -> {
processPdf(file);
}).start();
}
这样确实能把所有任务都启动起来。
但问题是，100 个 PDF 就会创建 100 个线程。
如果每个 PDF 都要读文件、生成水印、写新文件，压力不会小。
可能出现的问题包括：
线程太多，CPU 频繁切换；
同时读写太多 PDF，磁盘 IO 压力变大；
PDF 文件比较大时，内存占用上升；
程序看起来是并发了，但整体反而不稳定。
所以我不一定想让所有线程同时干活。
更合理的方式是：
可以创建多个任务；
但同一时间只允许 3 个任务真正进入 PDF 处理逻辑。
这就是 Semaphore 的作用。
Semaphore 可以理解成许可证
Semaphore 的核心概念是“许可证”。
比如：
Semaphore semaphore = new Semaphore(3);
意思是创建一个有 3 个许可证的信号量。
谁想执行受限制的代码，就先拿许可证：
semaphore.acquire();
执行完以后，再归还许可证：
semaphore.release();
可以简单理解成：
许可证有 3 个；
拿到许可证的线程才能进去处理 PDF；
没有拿到许可证的线程，就在外面等；
处理完 PDF 后，把许可证还回去；
等待的线程再继续进去。
这个思路其实挺直观。
就像一个停车场只有 3 个车位。
来了 10 辆车，不是 10 辆车都能进去。前 3 辆进去，后面的车只能等。等里面有车出来，外面的车才能进去。
先写一个简单例子
新建类：
com.succos.thread.SemaphoreStepDemo
代码如下：
package com.succos.thread;
import java.util.concurrent.Semaphore;
public class SemaphoreStepDemo {
private static final Semaphore semaphore = new Semaphore(2);
public static void main(String[] args) {
for (int i = 1; i <= 5; i++) {
int taskId = i;
new Thread(() -> {
boolean acquired = false;
try {
System.out.println("任务 " + taskId + " 等待进入，线程：" + Thread.currentThread().getName());
semaphore.acquire();
acquired = true;
System.out.println("任务 " + taskId + " 开始执行，线程：" + Thread.currentThread().getName());
Thread.sleep(3000);
System.out.println("任务 " + taskId + " 执行完成，线程：" + Thread.currentThread().getName());
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
if (acquired) {
semaphore.release();
System.out.println("任务 " + taskId + " 释放许可证，线程：" + Thread.currentThread().getName());
}
}
}, "task-thread-" + taskId).start();
}
}
}
这里我创建了 5 个线程。
但是信号量只有 2 个许可证：
private static final Semaphore semaphore = new Semaphore(2);
所以同一时间最多只有 2 个线程能进入真正的执行逻辑。
运行结果大概是什么样
运行后，输出可能类似这样：
任务 1 等待进入，线程：task-thread-1
任务 2 等待进入，线程：task-thread-2
任务 3 等待进入，线程：task-thread-3
任务 4 等待进入，线程：task-thread-4
任务 5 等待进入，线程：task-thread-5
任务 1 开始执行，线程：task-thread-1
任务 2 开始执行，线程：task-thread-2
等待 3 秒...
任务 1 执行完成，线程：task-thread-1
任务 1 释放许可证，线程：task-thread-1
任务 3 开始执行，线程：task-thread-3
任务 2 执行完成，线程：task-thread-2
任务 2 释放许可证，线程：task-thread-2
任务 4 开始执行，线程：task-thread-4
顺序不一定完全一样，但现象是一样的：
同一时间最多只有 2 个任务在执行。
其他任务虽然线程已经启动了，但它们会卡在：
semaphore.acquire();
等许可证。
acquire() 是等待许可证
这行代码很关键：
semaphore.acquire();
它的意思是：
我要拿一个许可证。
如果当前还有许可证，线程就能继续往下执行。
如果许可证已经被拿完了，当前线程就会在这里等待。
比如 Semaphore(2)，前两个线程进来时：
初始许可证：2
线程1 acquire，许可证变成 1；
线程2 acquire，许可证变成 0；
线程3 再 acquire，没有许可证，只能等待。
等某个线程执行完，调用：
semaphore.release();
许可证数量才会增加，等待的线程才有机会继续执行。
release() 是归还许可证
release() 的意思是归还许可证。
比如：
semaphore.release();
它不是释放锁，而是把许可证还回去。
如果有线程正在 acquire() 那里等，就会有一个线程被唤醒，继续往下执行。
所以基本流程是：
进入之前 acquire；
执行完成 release。
这两个动作要成对出现。
如果只 acquire() 不 release()，许可证会越来越少，最后其他线程可能一直等。
如果没拿到许可证却错误地 release()，许可证数量可能会被多释放，限制就失效了。
所以这也是为什么我在代码里写了：
boolean acquired = false;
为什么要写 acquired 标记
代码里有这一段：
boolean acquired = false;
try {
semaphore.acquire();
acquired = true;
// 执行业务逻辑
} finally {
if (acquired) {
semaphore.release();
}
}
这个写法不是多余的。
因为 acquire() 本身会抛 InterruptedException。
如果线程在等待许可证时被中断了，acquire() 可能还没成功，代码就进入 finally。
这时如果我直接写：
finally {
semaphore.release();
}
就可能出现一个问题：
明明没有拿到许可证，却释放了一次许可证。
这样许可证数量就会被错误增加。
比如本来只有 2 个许可证，结果因为错误 release，可能变成 3 个、4 个。
这样 Semaphore 的并发控制就失效了。
所以我更稳妥的写法是：
boolean acquired = false;
semaphore.acquire();
acquired = true;
...
if (acquired) {
semaphore.release();
}
只有确认拿到了许可证，才释放。
用 Semaphore 控制 PDF 并发数量
现在把这个思路放回 PDF 项目。
假设我希望：
可以给每个 PDF 创建一个任务；
但同一时间最多只允许 3 个 PDF 进入处理逻辑。
核心写法大概就是：
Semaphore semaphore = new Semaphore(3);
每个任务里：
semaphore.acquire();
try {
// 处理 PDF
} finally {
semaphore.release();
}
这样不管我启动了多少个线程，真正同时处理 PDF 的数量最多就是 3。
PDF 场景的简单代码
新建类：
com.succos.thread.SemaphorePdfDemo
先用 Thread.sleep() 模拟 PDF 处理：
package com.succos.thread;
import java.io.File;
import java.util.concurrent.Semaphore;
public class SemaphorePdfDemo {
private static final Semaphore semaphore = new Semaphore(3);
public static void main(String[] args) {
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;
}
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());
}
}
}, "pdf-thread-" + file.getName());
thread.start();
}
}
}
这段代码的效果是：
每个 PDF 都对应一个线程；
但是同一时间最多只有 3 个线程真正执行处理逻辑；
其他线程会等在 acquire()。
这就比无限制同时处理要稳一些。
这段代码还缺什么？
上面的代码已经能限制并发数量，但它还不是很完整。
主要缺两个东西。
第一个是：主线程没有等待所有 PDF 处理完成。
也就是说，代码里没有 join()，也没有 CountDownLatch。
如果后面我要统计总耗时、成功数量、失败数量，就需要等待所有任务结束。
第二个是：每个 PDF 一个线程，文件多了还是会创建很多线程。
Semaphore 只是限制了“同一时间有几个线程进入处理逻辑”，但它没有减少线程创建数量。
比如 1000 个 PDF：
还是会创建 1000 个线程；
只是同一时间只有 3 个线程真正处理 PDF；
剩下的线程都在 acquire() 那里等。
所以 Semaphore 能控制并发进入数量，但不能替代线程池。
这点要分清楚。
Semaphore 和线程池的区别
Semaphore 控制的是：
同一时间最多几个线程可以进入某段代码。
线程池控制的是：
创建多少线程；
任务怎么排队；
空闲线程怎么复用；
队列满了怎么办。
所以它们不是一回事。
在学习阶段，用 Thread + Semaphore 很适合观察并发控制。
但在真实项目里，我一般不会为每个 PDF 都手动创建线程，再靠 Semaphore 去拦。
更常见的方式是用线程池：
线程池本身就限制了工作线程数量；
任务多了进入队列等待；
线程处理完一个任务，再继续取下一个任务。
也就是说，后面学到 ThreadPoolExecutor 后，很多 PDF 批量处理场景会更适合用线程池。
不过 Semaphore 仍然有用。
比如线程池里某一段代码还要限制访问外部资源数量，就可以继续用它。
Semaphore 更适合限制资源访问
我现在觉得 Semaphore 特别适合控制“资源数量有限”的场景。
比如：
最多同时调用 3 个外部接口；
最多同时处理 2 个大文件；
最多允许 5 个任务访问某个模型服务；
最多允许 1 个线程执行某个 Python 脚本。
这种场景本质上不是要保护某个变量，而是资源有限。
资源只有几个，就用几个许可证。
比如：
private static final Semaphore PYTHON_SEMAPHORE = new Semaphore(1);
这就表示：
同一时间只允许一个线程执行 Python 相关逻辑。
这种写法在调用外部脚本、模型推理、GPU 任务时挺常见。
Semaphore 和 Lock 的区别
到这里，Semaphore 和锁的区别就比较清楚了。
ReentrantLock 一般是：
同一时间只允许一个线程进入；
主要保护共享数据。
而 Semaphore 可以是：
同一时间允许 N 个线程进入；
主要控制资源访问数量。
比如：
new Semaphore(1);
效果有点像锁，同一时间只允许一个线程进入。
但：
new Semaphore(3);
就和普通锁不一样了，它允许 3 个线程同时进入。
所以我不会简单把 Semaphore 当成锁。
它更像是一个“并发数量阀门”。
这一节小结
这一节我主要记住几点：
1. Semaphore 用来控制同一时间最多允许几个线程进入某段代码；
2. acquire() 是获取许可证，拿不到就等待；
3. release() 是归还许可证；
4. acquire 和 release 要成对出现；
5. 最好用 acquired 标记，避免没拿到许可证却错误释放；
6. Semaphore 能限制并发进入数量，但不能减少线程创建数量；
7. PDF 批量处理最终更适合线程池，但 Semaphore 很适合理解并发限制。
用一句话总结：
Semaphore 不是用来保护某个变量的，它更像是在入口处放了几个许可证，谁拿到谁进去。
下一节继续写一个更完整的 Semaphore 实战版本。
重点是：限制 PDF 并发处理数量，同时让主线程知道所有任务什么时候结束。
