Java 执行器用固定的线程池运行任务，并交回一个 Future。学会 submit、invokeAll、取消、CompletableFuture 链、超时、信号量和阻塞队列，每个示例的输出都稳定不变。
每个任务都手动启动一个线程，在小程序里行得通。在服务器里就不行了：每个任务都要付出新建线程的代价，而且一阵突发的工作量可能创建出成千上万个线程，没有任何东西拦得住。java.util.concurrent 包提供了另一种办法，执行器：一组固定数量的线程从队列里取任务，并交回一个结果让你等待。
本文讲 ExecutorService、Future、任务中的异常和取消、线程池大小、CompletableFuture、CountDownLatch、Semaphore，以及两种并发集合。下面每个程序都在 Java 25 上运行过，输出直接从运行结果复制而来。想自己运行，就把代码保存为 Main.java，然后执行 java Main.java。和讲线程与共享状态的那一部分一样，这些并发程序用闭锁强制安排时机，所以每次运行输出都一样。
为什么要用执行器：一个线程池加一个队列
每个任务一个线程，有两笔开销。每个平台线程都是一个操作系统线程，有自己的栈，在 64 位 Linux 上默认预留 1 MB，创建一个也要花不少功夫。而且数量没有上限：一万个请求同时到达，就意味着一万个线程。
执行器把这两个问题都解决了。Executors.newFixedThreadPool(4) 创建四个线程并一直留着。交给它的任务在队列里等着，直到四个线程中有一个空出来。下面的程序提交十个任务，它们都阻塞在一道闸门上，然后看看线程池内部：
void main() throws InterruptedException {
var running = new AtomicInteger();
var mostAtOnce = new AtomicInteger();
var fourStarted = new CountDownLatch(4);
var gate = new CountDownLatch(1);
var done = new AtomicInteger();
try (ExecutorService pool = Executors.newFixedThreadPool(4)) {
for (int i = 0; i < 10; i++) {
pool.execute(() -> {
mostAtOnce.accumulateAndGet(running.incrementAndGet(), Math::max);
fourStarted.countDown();
awaitQuietly(gate);
running.decrementAndGet();
done.incrementAndGet();
});
}
fourStarted.await();
var queue = ((ThreadPoolExecutor) pool).getQueue();
IO.println("tasks running: " + running.get());
IO.println("tasks waiting in the queue: " + queue.size());
gate.countDown();
} // close() waits for every task to finish
IO.println("tasks finished: " + done.get());
IO.println("most tasks running at once: " + mostAtOnce.get());
}
void awaitQuietly(CountDownLatch latch) {
try {
latch.await();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}输出：
tasks running: 4
tasks waiting in the queue: 6
tasks finished: 10
most tasks running at once: 4四个任务启动后卡在闸门前。另外六个待在队列里，因为没有空闲的线程来运行它们。main 打开闸门后，四个线程依次处理完队列，任何时刻运行的任务都没有超过四个。
execute 把一个 Runnable 交给线程池，不返回任何东西。强制转换成 ThreadPoolExecutor 只是为了偷看一眼队列，真实代码里用不着。
这个 try 块是 try-with-resources。ExecutorService 从 Java 19 起实现了 AutoCloseable（我们验证过：javac --release 18 会拒绝这个程序），close() 让线程池不再接受新任务，并等待队列里的任务完成。没有它，main 返回后，线程池的线程会让 JVM 一直活着。
一个固定为四个线程的线程池。四个任务在运行，六个在队列里等待，调用方为提交的每个任务都拿着一个 Future。线程只创建一次，所有任务反复使用它们。
用十岁孩子能懂的话说
线程池就像餐厅后厨，厨师人数固定，还有一根挂单的横杆。餐厅不会每来一张订单就雇一个新厨师。你递上一张点菜单，它被挂到横杆上。哪个厨师空了，就从横杆上取下一张单子，做那道菜。
递上单子，你会拿到一个取餐呼叫器。那就是 Future。你可以坐下做别的事，饭好了呼叫器就会亮。如果你宁愿站在柜台前等，那就是 get()。
打烊时，经理可以不再接单，让厨师把横杆上的单子做完。也可以把横杆上的单子全部撤下，叫厨师马上停手。
准确的说法
newFixedThreadPool(n) 返回一个 ThreadPoolExecutor，它有 n 个工作线程和一个存放任务的 LinkedBlockingQueue。每个工作线程循环执行：从队列取一个任务，运行它，再重复。submit 把你的任务包装进一个 FutureTask，它既是工作线程运行的东西，也是你拿着的 Future。任务返回或抛出异常时，FutureTask 存下结果或异常，所有阻塞在 get() 里的线程都会被唤醒。
这个比喻的局限：newFixedThreadPool 里的横杆没有长度限制，所以大量涌来的任务会在内存里越堆越多，而不是被拒之门外。呼叫器也不只是一个信号，它还装着结果，包括失败。而且被叫停的厨师可以不理睬：在 Java 里，停止一个正在运行的任务只是一个请求，任务得自己配合，讲取消的那一节会展示这一点。
submit、invokeAll 和 invokeAny
submit 接受一个返回值的 Callable，或者一个 Runnable，交给你一个 Future。invokeAll 提交一组任务并等待全部完成。invokeAny 等待第一个成功的任务：
void main() throws Exception {
try (var pool = Executors.newFixedThreadPool(3)) {
Future<Integer> length = pool.submit(() -> "executor".length());
IO.println("submit: get() returned " + length.get());
var thirdDone = new CountDownLatch(1);
List<Callable<String>> jobs = List.of(
() -> {
thirdDone.await(); // finish last on purpose
return "first";
},
() -> "second",
() -> {
thirdDone.countDown();
return "third";
});
var results = new ArrayList<String>();
for (Future<String> f : pool.invokeAll(jobs)) {
results.add(f.get());
}
IO.println("invokeAll: " + results);
List<Callable<String>> mirrors = List.of(
() -> { throw new IOException("mirror A is down"); },
() -> "downloaded from mirror B",
() -> { throw new IOException("mirror C is down"); });
IO.println("invokeAny: " + pool.invokeAny(mirrors));
}
}输出：
submit: get() returned 8
invokeAll: [first, second, third]
invokeAny: downloaded from mirror Bget() 一直阻塞到任务完成，然后返回它的值。
第一个任务要等第三个完成，所以任务完成的顺序是乱的。invokeAll 仍然按你提交任务的顺序返回 future，所以不管线程怎样运行，在循环里读取它们得到的都是固定的顺序。
invokeAny 返回了镜像 B 的结果，因为另外两个任务抛了异常。一旦有一个任务成功，它就取消其余任务。如果所有任务都失败，它抛出 ExecutionException，里面带着其中一个失败。
停止线程池：shutdown 和 shutdownNow
线程池有两种停止方式，区别在于队列里的任务怎么处理。这里每个线程池只有一个线程，一个正在运行的任务阻塞在闸门上，后面还排着三个任务：
void main() throws InterruptedException {
IO.println("shutdown(): " + finishedTasks(false));
IO.println("shutdownNow(): " + finishedTasks(true));
}
String finishedTasks(boolean now) throws InterruptedException {
var pool = Executors.newFixedThreadPool(1);
var started = new CountDownLatch(1);
var gate = new CountDownLatch(1);
var finished = new AtomicInteger();
pool.execute(() -> {
started.countDown();
try {
gate.await();
finished.incrementAndGet();
} catch (InterruptedException e) {
Thread.currentThread().interrupt(); // interrupted: give up
}
});
for (int i = 0; i < 3; i++) {
pool.execute(finished::incrementAndGet);
}
started.await(); // one task running, three in the queue
int neverStarted = 0;
if (now) {
neverStarted = pool.shutdownNow().size();
} else {
pool.shutdown();
}
String rejected = "";
try {
pool.execute(finished::incrementAndGet);
} catch (RejectedExecutionException e) {
rejected = "new task rejected, ";
}
gate.countDown();
boolean terminated = pool.awaitTermination(5, TimeUnit.SECONDS);
return rejected + "finished " + finished.get() + ", never started " + neverStarted
+ ", terminated " + terminated;
}输出：
shutdown(): new task rejected, finished 4, never started 0, terminated true
shutdownNow(): new task rejected, finished 0, never started 3, terminated true两个方法都会让线程池用 RejectedExecutionException 拒绝新任务。
shutdown() 让正在运行的任务和整个队列都完成，所以四个任务全部完成。shutdownNow() 移除了排队的三个任务，把它们作为列表返回，并中断正在运行的那个。那个任务捕获了中断，放弃了，所以一个都没完成。
awaitTermination 在超时时间内等待线程池的线程结束，并返回它们是否结束了。close() 大致相当于先调用 shutdown()，再循环调用 awaitTermination。
任务里的异常
任务里抛出的异常，不会立刻传到提交它的代码那里。它去了哪里，取决于你怎样交出这个任务。下面的程序给线程池一个线程工厂，工厂会安装未捕获异常处理器，这样我们就能看到有什么到达了处理器：
void main() throws InterruptedException {
var handlerSaw = new LinkedBlockingQueue<String>();
var threadsMade = new AtomicInteger();
ThreadFactory withHandler = task -> {
threadsMade.incrementAndGet();
var thread = new Thread(task);
thread.setUncaughtExceptionHandler((t, e) -> handlerSaw.add(e.getMessage()));
return thread;
};
try (var pool = Executors.newFixedThreadPool(1, withHandler)) {
Future<Integer> parsed = pool.submit(() -> Integer.parseInt("forty-two"));
try {
parsed.get();
} catch (ExecutionException e) {
IO.println("submit: get() threw " + e.getClass().getSimpleName());
IO.println(" cause: " + e.getCause());
}
pool.execute(() -> Integer.parseInt("seven"));
IO.println("execute: the handler got " + handlerSaw.take());
IO.println(" threads the pool has made: " + threadsMade.get());
Future<?> forgotten = pool.submit(() -> Integer.parseInt("nine"));
while (!forgotten.isDone()) {
Thread.sleep(1);
}
IO.println("submit, no get(): state is " + forgotten.state());
IO.println(" and the handler got " + handlerSaw.poll());
}
}输出：
submit: get() threw ExecutionException
cause: java.lang.NumberFormatException: For input string: "forty-two"
execute: the handler got For input string: "seven"
threads the pool has made: 2
submit, no get(): state is FAILED
and the handler got null用 submit 时，Future 捕获异常并存起来。get() 抛出 ExecutionException，getCause() 就是原来的 NumberFormatException。
用 execute 时，没有 Future 来接住它。异常逃出任务，杀死了工作线程，于是线程的未捕获异常处理器收到它，线程池再造一个线程顶替。没有自定义处理器时，默认处理器把栈跟踪打印到标准错误，main 对此一无所知。
第三种情况会丢掉错误。一个用 submit 提交的任务失败了，却没人调用 get()。处理器什么也没收到，哪里都没有打印任何东西。Java 19 加入的 Future.state() 会显示 FAILED，但得有人去问。提交任务时，要留住它的 Future，并对它调用 get()。
超时与取消
get 有一个带超时的版本，cancel(true) 会中断已经在运行的任务。这里任务等待一个永远不会打开的闭锁，所以 50 毫秒的超时一定会到期：
void main() throws Exception {
var neverOpens = new CountDownLatch(1);
var started = new CountDownLatch(1);
var sawInterrupt = new CountDownLatch(1);
try (var pool = Executors.newFixedThreadPool(1)) {
Future<String> slow = pool.submit(() -> {
started.countDown();
try {
neverOpens.await();
return "finished";
} catch (InterruptedException e) {
sawInterrupt.countDown();
throw e;
}
});
started.await();
try {
slow.get(50, TimeUnit.MILLISECONDS);
} catch (TimeoutException e) {
IO.println("get(50 ms) threw TimeoutException");
}
IO.println("the task is still running: " + !slow.isDone());
boolean cancelled = slow.cancel(true);
sawInterrupt.await();
IO.println("cancel(true) returned " + cancelled);
IO.println("the task saw the interrupt and stopped waiting");
IO.println("state: " + slow.state());
try {
slow.get();
} catch (CancellationException e) {
IO.println("get() now throws CancellationException");
}
}
}输出：
get(50 ms) threw TimeoutException
the task is still running: true
cancel(true) returned true
the task saw the interrupt and stopped waiting
state: CANCELLED
get() now throws CancellationExceptionTimeoutException 只表示你不再等了。任务还在继续运行。接着 cancel(true) 中断了它的线程，await() 抛出 InterruptedException，任务结束。此后，get() 抛出 CancellationException。
cancel(false) 不会中断。它照样把 future 标记为已取消，但已经在运行的任务会一直运行到结束。无论哪种 cancel，都能让排队中的任务永远不会开始。
中断只是一个请求
中断并不会停止线程。它只在线程上设置一个标志，await、sleep 和 BlockingQueue.take 这类阻塞方法会注意到这个标志，并抛出 InterruptedException。它们抛出异常时，会清除这个标志。所以一个捕获异常后继续干活的任务，就把这个请求抹掉了。
下面两个任务完全一样，只有 catch 里的一行不同：
void main() throws InterruptedException {
var neverOpens = new CountDownLatch(1);
var bothStarted = new CountDownLatch(2);
var politeStopped = new CountDownLatch(1);
var rudeIgnoredIt = new CountDownLatch(1);
var pool = Executors.newFixedThreadPool(2, Thread.ofPlatform().daemon().factory());
pool.execute(() -> {
bothStarted.countDown();
while (!Thread.currentThread().isInterrupted()) {
try {
neverOpens.await();
} catch (InterruptedException e) {
Thread.currentThread().interrupt(); // put the flag back
}
}
politeStopped.countDown();
});
pool.execute(() -> {
bothStarted.countDown();
while (!Thread.currentThread().isInterrupted()) {
try {
neverOpens.await();
} catch (InterruptedException e) {
rudeIgnoredIt.countDown(); // swallowed: the flag stays clear
}
}
});
bothStarted.await();
pool.shutdownNow(); // interrupts both tasks
politeStopped.await();
rudeIgnoredIt.await();
IO.println("the task that restored the flag stopped");
IO.println("the task that swallowed the interrupt went back to waiting");
IO.println("pool terminated within 200 ms: "
+ pool.awaitTermination(200, TimeUnit.MILLISECONDS));
}输出：
the task that restored the flag stopped
the task that swallowed the interrupt went back to waiting
pool terminated within 200 ms: false第一个任务用 Thread.currentThread().interrupt() 把标志放了回去，所以循环条件看到了它，任务结束。第二个任务吞掉了异常，标志保持清除状态，循环直接回去继续等待。shutdownNow 只发出一次中断，没有第二次可发，所以线程池无法结束。这里的线程设成守护线程，只是为了让程序能够退出。
规则是：要么让 InterruptedException 继续向上传播，要么捕获它并调用 Thread.currentThread().interrupt()。没有阻塞调用的长循环，应该自己检查 Thread.currentThread().isInterrupted()。静态方法 Thread.interrupted() 也能读取标志，但它同时会清除标志，所以只在你就地处理中断时才用它。
该用多少个线程？
线程池合适的大小，取决于任务把时间花在什么上，没有现成的精确答案可查。
CPU 密集型任务，比如解析、压缩或数值计算，全程都要占着一个核心。线程比核心多，线程就得轮流运行，而在它们之间切换要花时间。先从 Runtime.getRuntime().availableProcessors() 附近开始。
IO 密集型任务，比如调用数据库或另一个服务，大部分时间都在等待。等待中的线程不占 CPU，所以线程比核心多反而有帮助。
来自《Java Concurrency in Practice》的经典经验法则是：
线程数 = 核心数 × (1 + 等待时间 / 计算时间)
一个任务等待 90 毫秒、计算 10 毫秒，在 8 核机器上得出 8 × (1 + 9) = 80 个线程。把它当作初步估计，而不是答案。数据库可能只允许 20 个连接，那 80 个线程只会在它门口排队。要在真实负载下测量，再做调整。
对 IO 密集型工作，Java 21 的虚拟线程通常让这个问题不复存在。下一部分会讲它们。
CompletableFuture：把步骤串起来
Future 只给你一种使用结果的方式，就是阻塞在 get() 里。CompletableFuture 让你挂上下一步，值一到，下一步就运行：
record User(int id, String name) {}
void main() {
try (var pool = Executors.newFixedThreadPool(4)) {
CompletableFuture<User> user = CompletableFuture.supplyAsync(() -> findUser(7), pool);
CompletableFuture<String> name = user.thenApply(User::name);
IO.println("thenApply: " + name.join());
CompletableFuture<CompletableFuture<Integer>> nested =
user.thenApply(u -> cartTotal(u, pool));
CompletableFuture<Integer> total = user.thenCompose(u -> cartTotal(u, pool));
IO.println("thenApply, nested: " + nested.join().join());
IO.println("thenCompose: " + total.join());
CompletableFuture<Integer> shipping = CompletableFuture.supplyAsync(() -> 5, pool);
CompletableFuture<Integer> toPay = total.thenCombine(shipping, Integer::sum);
IO.println("thenCombine: " + toPay.join());
}
}
User findUser(int id) {
return new User(id, "Ana"); // stands in for a database call
}
CompletableFuture<Integer> cartTotal(User u, Executor pool) {
return CompletableFuture.supplyAsync(() -> 37, pool); // stands in for another service
}输出：
thenApply: Ana
thenApply, nested: 37
thenCompose: 37
thenCombine: 42supplyAsync 在你传入的执行器上运行 supplier。thenApply 在值就绪时转换它，就像流上的 map。
当下一步本身返回一个 future 时，thenApply 和 thenCompose 就不一样了。用 thenApply，你得到的是 future 套 future，CompletableFuture<CompletableFuture<Integer>>，需要调用两次 join()。thenCompose 把它展平成一个 CompletableFuture<Integer>，就像 flatMap。thenCombine 等待两个相互独立的 future，把它们的值合并起来。
join() 像 get() 一样等待，但不抛受检异常，所以放在 lambda 表达式里很合适。
要传入执行器。 不传的话，supplyAsync 使用 ForkJoinPool.commonPool()。这个池和并行流以及 JVM 里的其他一切共用，而且是按 CPU 工作来定大小的，线程数比机器的核心数少一个。在里面做阻塞调用，可能会把不相干的代码饿死。它的线程还是守护线程（我们验证过：在其中一个线程里，Thread.currentThread().isDaemon() 返回了 true），所以你的任务还在运行时，JVM 就可能退出。
用 allOf 等待多个 future
CompletableFuture.allOf 返回一个 future，你传入的所有 future 都完成时，它才完成。它不带任何值，所以之后要逐个读取。这里每个天气预报都要等前一个完成，所以它们按相反的顺序完成：
void main() {
var finishOrder = new ConcurrentLinkedQueue<String>();
try (var pool = Executors.newFixedThreadPool(3)) {
CompletableFuture<?> nothing = CompletableFuture.completedFuture(null);
// each forecast waits for the one before it, so they finish in reverse
var beijing = forecast("Beijing", nothing, finishOrder, pool);
var madrid = forecast("Madrid", beijing, finishOrder, pool);
var lisbon = forecast("Lisbon", madrid, finishOrder, pool);
CompletableFuture.allOf(lisbon, madrid, beijing).join();
IO.println("finished: " + finishOrder);
for (var f : List.of(lisbon, madrid, beijing)) {
IO.println(f.join());
}
}
}
CompletableFuture<String> forecast(
String city, CompletableFuture<?> after, Queue<String> finishOrder, Executor pool) {
return CompletableFuture.supplyAsync(() -> {
after.join();
finishOrder.add(city);
int celsius = switch (city) {
case "Lisbon" -> 24;
case "Madrid" -> 31;
default -> 28;
};
return city + ": " + celsius + "C";
}, pool);
}输出：
finished: [Beijing, Madrid, Lisbon]
Lisbon: 24C
Madrid: 31C
Beijing: 28C天气预报是北京最先完成的。allOf(...).join() 之后，每个 future 都已完成，所以循环里的每次 join() 都立即返回。循环按你选定的顺序读取它们，所以输出顺序是固定的。
某个阶段失败时
CompletableFuture 怎样包装失败，取决于你怎样等待它：
void main() throws InterruptedException {
try (var pool = Executors.newFixedThreadPool(2)) {
CompletableFuture<Integer> port =
CompletableFuture.supplyAsync(() -> Integer.parseInt("eighty"), pool);
try {
port.join();
} catch (CompletionException e) {
IO.println("join() threw CompletionException");
IO.println(" cause: " + e.getCause());
}
try {
port.get();
} catch (ExecutionException e) {
IO.println("get() threw ExecutionException");
IO.println(" cause: " + e.getCause());
}
int withDefault = port.exceptionally(ex -> 8080).join();
IO.println("exceptionally: " + withDefault);
String report = port
.thenApply(p -> "listening on " + p)
.handle((value, ex) -> {
if (ex == null) {
return value;
}
Throwable real = ex instanceof CompletionException ? ex.getCause() : ex;
return "handle got " + ex.getClass().getSimpleName()
+ ", real problem: " + real.getMessage();
})
.join();
IO.println(report);
}
}输出：
join() threw CompletionException
cause: java.lang.NumberFormatException: For input string: "eighty"
get() threw ExecutionException
cause: java.lang.NumberFormatException: For input string: "eighty"
exceptionally: 8080
handle got CompletionException, real problem: For input string: "eighty"join() 把失败包装在非受检的 CompletionException 里。get() 把它包装在受检的 ExecutionException 里，和 Future.get() 一样。两种情况下，getCause() 都是真正的异常。
exceptionally 只在失败时运行，提供一个替代值。handle 无论成败都会运行，收到值或异常，两者之一为 null。
这一点让我们意外：handle 收到的是 CompletionException，不是 NumberFormatException。从前面阶段传下来的失败，到达时是被包装过的。我们还发现，直接在 supplyAsync 返回的 future 上调用 exceptionally，拿到的也是包装。所以查看异常之前先把它拆开，就像程序里做的那样。
CompletableFuture 上的超时
orTimeout 在一段延迟后让 future 失败，completeOnTimeout 则改为用一个后备值完成它。这里的两次查询都在等一个永远不会打开的闭锁：
void main() {
var neverOpens = new CountDownLatch(1);
var slowTaskEnded = new CountDownLatch(2);
try (var pool = Executors.newFixedThreadPool(2)) {
Supplier<String> slowLookup = () -> {
try {
neverOpens.await();
return "live price";
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return "interrupted";
} finally {
slowTaskEnded.countDown();
}
};
CompletableFuture<String> strict = CompletableFuture
.supplyAsync(slowLookup, pool)
.orTimeout(50, TimeUnit.MILLISECONDS);
try {
strict.join();
} catch (CompletionException e) {
IO.println("orTimeout: " + e.getCause().getClass().getSimpleName());
}
String price = CompletableFuture
.supplyAsync(slowLookup, pool)
.completeOnTimeout("cached price", 50, TimeUnit.MILLISECONDS)
.join();
IO.println("completeOnTimeout: " + price);
IO.println("slow lookups that have ended: " + (2 - slowTaskEnded.getCount()));
neverOpens.countDown(); // let them finish, or close() would wait forever
}
IO.println("slow lookups that have ended: " + (2 - slowTaskEnded.getCount()));
}输出：
orTimeout: TimeoutException
completeOnTimeout: cached price
slow lookups that have ended: 0
slow lookups that have ended: 2orTimeout 让 future 以 TimeoutException 失败，join() 又把它包装进 CompletionException。completeOnTimeout 给出了缓存的价格。
两者都没有停止工作本身。两次超时之后，慢查询一个都没结束：它们的线程仍然阻塞着。main 必须打开闭锁，否则 close() 会永远等下去。cancel(true) 也帮不上忙。它的文档说中断标志不起作用，我们也验证过：任务从来没看到中断。main 打开闭锁后，两次查询都结束了，这就是最后一行。CompletableFuture 上的超时停止的是等待，而不是任务。
协调工具
java.util.concurrent 还有一些小工具，让线程互相等待。讲线程的那一部分用过的 CountDownLatch，可以让一个线程等到其他线程完成某件事：
void main() throws InterruptedException {
var services = List.of("search", "cache", "database");
var allReady = new CountDownLatch(services.size());
var ready = new ConcurrentSkipListSet<String>();
try (var pool = Executors.newFixedThreadPool(3)) {
for (String service : services) {
pool.execute(() -> {
ready.add(service); // stands in for slow start-up work
allReady.countDown();
});
}
allReady.await();
IO.println("every service is up: " + ready);
}
}输出：
every service is up: [cache, database, search]闭锁从 3 开始，每个服务倒数一次，减到 0 时 await() 返回。ConcurrentSkipListSet 让元素保持有序，所以打印顺序是固定的。
Semaphore（信号量）持有若干个许可。acquire() 取走一个，没有剩余时就等待，release() 把它还回去。它限制同时做某件事的线程数，比如调用一个只接受两个连接的服务。这里六个任务共用两个许可，一个闭锁确保确实有两个任务同时在里面：
void main() throws InterruptedException {
var permits = new Semaphore(2);
var inside = new AtomicInteger();
var mostInside = new AtomicInteger();
var twoInside = new CountDownLatch(2);
var gate = new CountDownLatch(1);
try (var pool = Executors.newFixedThreadPool(6)) {
for (int i = 0; i < 6; i++) {
pool.execute(() -> {
try {
permits.acquire();
try {
mostInside.accumulateAndGet(inside.incrementAndGet(), Math::max);
twoInside.countDown();
gate.await();
inside.decrementAndGet();
} finally {
permits.release();
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
}
twoInside.await();
while (permits.getQueueLength() < 4) {
Thread.sleep(1);
}
IO.println("holding a permit: " + inside.get());
IO.println("waiting for one: " + permits.getQueueLength());
gate.countDown();
}
IO.println("most inside at once: " + mostInside.get());
}输出：
holding a permit: 2
waiting for one: 4
most inside at once: 2前两个任务拿到许可，在闸门前等着。另外四个在 acquire() 里等待。峰值是 2。twoInside 闭锁强制两个持有者同时在里面，信号量把第三个挡在了外面。和锁一样，要在 finally 里释放，否则失败的任务会永远占着它的许可。
CyclicBarrier 让固定数量的线程等待，直到所有线程都到达同一个点，然后让它们一起继续。
再看并发集合
讲线程的那一部分说过，单独一次 ConcurrentHashMap 调用是安全的，但跨两次调用的先检查后执行并不安全。compute 系列方法把整个更新放进一次调用里：
void main() {
var stock = new ConcurrentHashMap<String, Integer>();
stock.put("tea", 20);
stock.put("cake", 3);
var requests = new ConcurrentHashMap<String, Integer>();
var basket = List.of("tea", "cake", "tea", "soup", "cake", "tea");
try (var pool = Executors.newFixedThreadPool(4)) {
for (int i = 0; i < 4; i++) {
pool.execute(() -> {
for (String item : basket) {
requests.merge(item, 1, Integer::sum);
// take one; the last one removes the entry
stock.computeIfPresent(item, (name, left) -> left == 1 ? null : left - 1);
}
});
}
}
IO.println("requests: " + new TreeMap<>(requests));
IO.println("stock: " + new TreeMap<>(stock));
}输出：
requests: {cake=8, soup=4, tea=12}
stock: {tea=8}四个任务一共发出 24 次请求。merge 把它们都计上了，没有丢失任何更新。computeIfPresent 每次取走一件，返回 null 会移除条目，所以蛋糕卖了三件就卖完，条目也消失了。汤从来不在库存里，所以函数从没为它运行过。
BlockingQueue（阻塞队列）把工作从一个线程交给另一个线程。队列满时 put 等待，队列空时 take 等待：
void main() throws Exception {
var queue = new ArrayBlockingQueue<String>(2);
var received = new ArrayList<String>();
try (var pool = Executors.newFixedThreadPool(2)) {
Future<?> producer = pool.submit(() -> {
for (int i = 1; i <= 5; i++) {
queue.put("order-" + i); // waits while the queue is full
}
queue.put("DONE");
return null;
});
Future<?> consumer = pool.submit(() -> {
String order = queue.take(); // waits while the queue is empty
while (!order.equals("DONE")) {
received.add(order);
order = queue.take();
}
return null;
});
producer.get();
consumer.get();
}
IO.println("consumer received: " + received);
}输出：
consumer received: [order-1, order-2, order-3, order-4, order-5]容量是 2，所以生产者最多只能领先消费者两个订单。"DONE" 这个值告诉消费者停下来。一个生产者加一个消费者时，订单按进去的顺序出来。生产者或消费者有多个时，队列本身仍然安全，但跨线程的顺序就不固定了。
两个任务都是 Callable lambda，因为它们以 return null 结尾。这样 put 和 take 不用写 try，就能把 InterruptedException 抛进 Future。
要点
- 执行器反复使用一组固定的线程，多出来的任务放进队列。用 try-with-resources 打开它，这样
close()会等工作完成。 submit返回一个Future。要对它调用get()，否则失败任务的异常会悄无声息地丢掉。invokeAll按提交顺序返回 future。get()抛出ExecutionException，join()抛出CompletionException。真正的异常是 cause。- 超时停止的是等待，不是任务。中断只是一个请求，所以捕获
InterruptedException时要恢复标志。 - 做 CPU 工作的线程池，大小接近核心数。做 IO 工作的，先用等待与计算时间之比的经验法则估个数，再去测量。
- 给
CompletableFuture传入你自己的执行器。下一步返回 future 时用thenCompose，用allOf等全部完成，再按固定顺序join。 Semaphore限制同时做某件事的线程数。BlockingQueue按顺序在线程之间传递工作。
把任务交给线程池，留住每一个 future，并提前想好任务失败或运行太久时该怎么办。