承接的问题
上一章讨论的是「一把锁怎么选」。锁解决的是同一时刻谁能进临界区，不解决「同时有一千个请求，用几个线程去跑」。第 1 章说过平台线程是 1:1 映射到内核线程的实体，创建有成本、数量有上限；第 1.2 节把任务和执行器拆开了，但只用了 newSingleThreadExecutor 这个最简单的壳。
这一节把壳拆掉，直接面对 ThreadPoolExecutor 的构造函数，回答三个问题：七个参数各在哪一步生效；execute 面对新任务按什么顺序判断；以及 JDK 为什么选择「先入队，队列满了才扩容」，而不是「有空闲线程就复用、没有就新建」。这三个问题答不上来，参数就只能靠抄；答上来之后，newFixedThreadPool 为什么在生产环境危险、max 为什么常常形同虚设，都是自然推论。
七个参数各管一段
ThreadPoolExecutor 的构造签名有七个参数，它们的顺序不是随意的，而是按「先看谁、后看谁」排列：
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler)
| 参数 | 管什么 | 什么时候生效 |
|---|---|---|
corePoolSize | 常驻线程数 | 每次提交都先看它：线程数没到 core，直接新建 |
maximumPoolSize | 线程总数上限 | 只有队列满了才看它 |
keepAliveTime | 空闲线程存活时间 | 线程数超过 core 时，多余线程空闲多久退出 |
unit | 上者的时间单位 | 同上 |
workQueue | 排队容器 | 决定「队列满」这个事件何时发生，也决定容量能不能变 |
threadFactory | 线程怎么造 | 每次新建线程时；命名、daemon、异常处理器都在这里 |
handler | 队列满且线程到顶怎么办 | 最后一道闸门 |
几个容易被忽略的约束：
corePoolSize不能大于maximumPoolSize，否则构造器直接抛IllegalArgumentException。本机实测的报错文案是maximumPoolSize must be greater than or equal to corePoolSize。keepAliveTime默认只作用于超过 core 的那部分线程。核心线程空闲了也留着，这是「常驻」的含义。allowCoreThreadTimeOut(true)可以让核心线程也超时，但要求keepAliveTime > 0，否则抛IllegalArgumentException: Core threads must have nonzero keep alive times。- 队列的容量在构造时定死，
ArrayBlockingQueue之后不能改。要动态调整容量，只能换队列实现或重建线程池。
ThreadFactory 的写法在 1.2 节已经给过完整实现，这里只强调一点：线程名是最便宜的可观测性。一个未命名的池在 jcmd Thread.print 里全是 pool-3-thread-7，出事时无法判断它属于借阅、通知还是对账。
execute 的三段判断
execute 的逻辑可以压缩成三段判断。下面不是 JDK 源码的复制，而是决策层面的等价描述（ctl、Worker 这些实现细节留给卷二第 11 章）：
public void execute(Runnable command) {
if (command == null) throw new NullPointerException();
// 第一段：线程数还没到 core，直接开一个新线程跑它
if (workerCountOf(ctl.get()) < corePoolSize) {
if (addWorker(command, true)) return;
// 走到这里说明并发下状态变了，落到第二段
}
// 第二段：入队。能入队就认为「接下了」
if (isRunning(ctl.get()) && workQueue.offer(command)) {
int recheck = ctl.get();
// 入队后复查：池可能刚刚被关闭，或者最后一个工作线程刚死
if (!isRunning(recheck) && remove(command)) reject(command);
else if (workerCountOf(recheck) == 0) addWorker(null, false);
return;
}
// 第三段：入队失败（队列满），尝试扩容到 max
if (!addWorker(command, false)) reject(command);
}
三段之间有严格的优先级：core → queue → max → 拒绝。注意 maximumPoolSize 排在队列后面，这是理解整个线程池行为的钥匙。
第二段里的「入队后复查」值得单独说。offer 成功不代表任务一定会被执行：
- 池可能在
offer和复查之间被shutdown，任务卡在队列里没人取，所以要remove回来并交给拒绝策略； - 池里可能一个工作线程都没有（比如
corePoolSize = 0），任务进了队列也没有消费者，所以要补一个addWorker(null, false)。
这两条分支对应两种真实事故：关闭期间丢任务、core=0 的池提交任务后永远不执行。后者在本机实测过：core=0、无界队列、提交 5 个阻塞任务，结果是 poolSize=1 queue=4——池用 1 个线程跑完了所有排队任务，max=4 一次都没用上。
为什么先入队再加线程
如果让业务方设计，直觉方案通常是「没有空闲线程就新建，直到 max」。JDK 选了反过来的顺序，原因有三条：
第一，线程创建比入队贵一个数量级。 平台线程要有自己的栈（默认 1 MB 量级）、要进内核调度队列、要初始化 TLS。入队只是一次带锁的数组/链表写入。突发流量下，先让队列吸收，比立刻扩线程便宜得多。
第二，扩容容易，缩容难。 线程一旦建出来，要等它空闲满 keepAliveTime 才回收。如果每次小高峰都扩到 max，池的线程数会长期停在高位，内存和上下文切换成本都下不来。
第三，排队是一种可控的背压。 队列长度是显式的、可观测的；线程数暴涨是隐式的。队列满了会触发拒绝策略，让上游知道「系统到极限了」；而线程数暴涨只会让所有任务一起变慢，最后集体超时。
代价也很直接：队列越长，任务排队时间越长。所以这套顺序适合「短任务、允许排队」的场景；如果是长阻塞任务（等下游 RPC、等大文件 IO），任务会在队列里堆很久，此时应该调大 core、缩小队列，或者干脆换虚拟线程（第 14 章）。
还有一个常见误解：以为 max 是「最多同时跑多少个任务」。不是，max 是「最多存在多少个线程」，队列里的任务一个都没在跑。线程池通常长不到 max——除非队列先满。
实测：逐步提交看池与队列
下面这个类用 core=2, max=4, queue=ArrayBlockingQueue(2), keepAlive=2s，逐个提交 8 个会阻塞在同一个 CountDownLatch 上的任务，每次提交后打印池大小、队列长度和已开始执行的任务数。
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
public final class ExecutorFlowProbe {
public static void main(String[] args) throws Exception {
AtomicInteger seq = new AtomicInteger();
ThreadFactory factory = r -> new Thread(r, "loan-" + seq.incrementAndGet());
ThreadPoolExecutor pool = new ThreadPoolExecutor(
2, 4, 2, TimeUnit.SECONDS,
new ArrayBlockingQueue<>(2), factory);
CountDownLatch release = new CountDownLatch(1);
AtomicInteger started = new AtomicInteger();
AtomicInteger submitted = new AtomicInteger();
for (int i = 1; i <= 8; i++) {
int id = i;
try {
pool.execute(() -> {
started.incrementAndGet();
try {
release.await();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
submitted.incrementAndGet();
} catch (RejectedExecutionException e) {
System.out.println("#" + id + " rejected");
continue;
}
// 等到「已开始 + 已入队」覆盖本次提交，再取快照
awaitSettled(pool, started, submitted);
System.out.printf("#%d poolSize=%d queue=%d active=%d started=%d largest=%d%n",
id, pool.getPoolSize(), pool.getQueue().size(),
pool.getActiveCount(), started.get(), pool.getLargestPoolSize());
}
release.countDown();
boolean backToCore = waitUntil(() -> pool.getPoolSize() == 2, 10_000);
System.out.println("afterKeepAlive poolSize=" + pool.getPoolSize()
+ " reachedCore=" + backToCore);
pool.shutdown();
pool.awaitTermination(5, TimeUnit.SECONDS);
ThreadPoolExecutor ttl = new ThreadPoolExecutor(
2, 4, 1, TimeUnit.SECONDS,
new ArrayBlockingQueue<>(8), factory);
ttl.allowCoreThreadTimeOut(true);
CountDownLatch quick = new CountDownLatch(2);
for (int i = 0; i < 2; i++) {
ttl.execute(quick::countDown);
}
quick.await();
Thread.sleep(200);
System.out.println("coreTimeout before poolSize=" + ttl.getPoolSize());
boolean drained = waitUntil(() -> ttl.getPoolSize() == 0, 10_000);
System.out.println("coreTimeout after poolSize=" + ttl.getPoolSize()
+ " reachedZero=" + drained
+ " allowCoreThreadTimeOut=" + ttl.allowsCoreThreadTimeOut());
ttl.shutdown();
}
private static void awaitSettled(ThreadPoolExecutor pool,
AtomicInteger started,
AtomicInteger submitted) throws InterruptedException {
waitUntil(() -> started.get() + pool.getQueue().size() >= submitted.get(), 5_000);
}
private static boolean waitUntil(java.util.function.BooleanSupplier cond, long timeoutMs)
throws InterruptedException {
long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(timeoutMs);
while (System.nanoTime() < deadline) {
if (cond.getAsBoolean()) {
return true;
}
Thread.sleep(5);
}
return cond.getAsBoolean();
}
}
编译运行：
export JAVA_HOME=~/.jdks/jdk-25.0.5+8/Contents/Home
$JAVA_HOME/bin/javac --release 25 -d classes ExecutorFlowProbe.java
$JAVA_HOME/bin/java -cp classes ExecutorFlowProbe
本机输出（连续两次完全相同）：
#1 poolSize=1 queue=0 active=1 started=1 largest=1
#2 poolSize=2 queue=0 active=2 started=2 largest=2
#3 poolSize=2 queue=1 active=2 started=2 largest=2
#4 poolSize=2 queue=2 active=2 started=2 largest=2
#5 poolSize=3 queue=2 active=3 started=3 largest=3
#6 poolSize=4 queue=2 active=4 started=4 largest=4
#7 rejected
#8 rejected
afterKeepAlive poolSize=2 reachedCore=true
coreTimeout before poolSize=2
coreTimeout after poolSize=0 reachedZero=true allowCoreThreadTimeOut=true
逐行读：
#1、#2：线程数没到 core，直接新建核心线程，队列为空。#3、#4：core 已满，任务入队，queue从 1 涨到 2，poolSize保持在 2。这就是「先入队」的直接证据——队列还有空间时，max=4完全不参与。#5、#6：队列满了（容量 2），offer失败，才开始扩容，poolSize变成 3、4，largest同步涨到 4。#7、#8：队列满且线程到顶，交给拒绝策略，AbortPolicy抛RejectedExecutionException。afterKeepAlive poolSize=2：释放 latch 后任务跑完，两个非核心线程空闲满 2 秒退出，池回落到 core。keepAlive 只回收 core 之外的部分。coreTimeout before/after：第二个池显式allowCoreThreadTimeOut(true)，keepAlive 设 1 秒，两个核心线程在任务结束后也退出了，poolSize归零。
这些数字是结构性快照，不是耗时统计。本机 load averages 在 140 以上，但 poolSize、queue.size() 的取值不依赖 CPU 速度，只依赖线程池的决策顺序，所以可以作为证据；任何耗时类数字在这种负载下都不可引用。
在 libraryd-conc 里落地
把默认的固定线程池换成显式构造，七个参数一次写清，并且用 prestartAllCoreThreads() 避开「第一批请求同时触发建线程」的抖动：
package com.plume.conc.core;
import java.time.Instant;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
public final class PooledLoanService implements LoanService, AutoCloseable {
private final BookRepository books;
private final LoanRepository loans;
private final ThreadPoolExecutor pool;
public PooledLoanService(BookRepository books, LoanRepository loans) {
this.books = books;
this.loans = loans;
this.pool = new ThreadPoolExecutor(
4, 8, 60, TimeUnit.SECONDS,
new ArrayBlockingQueue<>(64),
new ThreadFactory() {
private final AtomicInteger seq = new AtomicInteger();
@Override
public Thread newThread(Runnable r) {
Thread t = new Thread(r, "loan-pool-" + seq.incrementAndGet());
t.setUncaughtExceptionHandler((thread, error) ->
System.err.println("uncaught in " + thread.getName() + ": " + error));
return t;
}
},
new ThreadPoolExecutor.AbortPolicy());
this.pool.prestartAllCoreThreads();
}
@Override
public Reservation reserve(String isbn, String memberId) {
return submitAndWait(() -> reserveLocked(isbn, memberId), "reserve isbn=" + isbn);
}
private synchronized Reservation reserveLocked(String isbn, String memberId) {
Book book = books.findByIsbn(isbn)
.orElseThrow(() -> new LoanException(ErrorCode.BOOK_NOT_FOUND, isbn));
if (book.stock() <= 0) {
throw new LoanException(ErrorCode.OUT_OF_STOCK, isbn);
}
books.save(new Book(book.isbn(), book.title(), book.stock() - 1));
return new Reservation("R-" + isbn, isbn, memberId, Instant.now().plusSeconds(86_400));
}
@Override
public boolean returnBook(String loanId) {
return submitAndWait(() -> returnLocked(loanId), "return loanId=" + loanId);
}
private synchronized boolean returnLocked(String loanId) {
Loan loan = loans.findById(loanId)
.orElseThrow(() -> new LoanException(ErrorCode.BOOK_NOT_FOUND, loanId));
Book book = books.findByIsbn(loan.isbn())
.orElseThrow(() -> new LoanException(ErrorCode.BOOK_NOT_FOUND, loan.isbn()));
books.save(new Book(book.isbn(), book.title(), book.stock() + 1));
return true;
}
private <T> T submitAndWait(Callable<T> task, String what) {
try {
return pool.submit(task).get();
} catch (RejectedExecutionException e) {
throw new LoanException(ErrorCode.CONFLICT, "pool saturated: " + what);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new LoanException(ErrorCode.INTERRUPTED, what + " interrupted");
} catch (ExecutionException e) {
if (e.getCause() instanceof LoanException loan) {
throw loan;
}
throw new IllegalStateException(what + " failed", e.getCause());
}
}
@Override
public void close() throws InterruptedException {
pool.shutdown();
if (!pool.awaitTermination(5, TimeUnit.SECONDS)) {
pool.shutdownNow();
}
}
}
参数选择的依据：
core=4：常规并发下的常驻处理能力，避免每个请求都触发建线程。max=8：留一倍余量吸收突发，再高就会把下游拖垮——线程池不是越大越好，瓶颈通常在数据库连接。queue=64：约等于「每秒请求数 × 可接受排队秒数」再留余量，具体算法在 10.2 节。keepAlive=60s：突发过后一分钟内多出来的线程退出。AbortPolicy：宁可快速失败，也不让请求在队列里无声地等到超时。
两个必须说清的取舍：其一，reserve 在请求线程里 submit().get()，一个请求占两个线程，这是「舱壁隔离」——把借阅计算限制在固定线程数内保护下游，代价是线程数翻倍；上游本来就异步时应该把 Future 往上抛（第 13 章）。其二，库存的读改写仍然需要互斥，synchronized 锁住的是服务实例，线程池只负责调度，不会让非线程安全的逻辑变安全。
submitAndWait 把「提交 → 等结果 → 还原业务异常 → 拒绝时快速失败」收成一个方法，reserve 和 returnBook 只关心领域逻辑。这也是线程池落地时最该抽的一层：任务体内不出现 Future、ExecutionException，异常翻译只写一次。
常见错误与边界
| 错误 | 后果 | 改法 |
|---|---|---|
用 Executors.newFixedThreadPool | 无界队列 + core == max，max 永不生效，队列可无限增长 | 显式 ThreadPoolExecutor + 有界队列 |
core=0 还配无界队列 | 池只会有 1 个线程，max 形同虚设 | 至少给 core 一个正数，或用 SynchronousQueue |
maximumPoolSize < corePoolSize | 构造时抛 IllegalArgumentException | 校验参数，或用配置中心兜底 |
以为 max 是并发上限 | 容量规划按 max 算，实际并发远低于预期 | 记住顺序是 core → queue → max |
把 newSingleThreadExecutor() 强转成 ThreadPoolExecutor | ClassCastException，它返回的是包装过的代理 | 需要改参数就自己 new |
| 在任务里改线程名/优先级 | 线程复用时污染下一个任务 | 属性统一在 ThreadFactory 里设 |
关闭时只调 shutdown() 不等 | 进程退出，队列里的任务丢失 | 两段式关闭，10.3 节展开 |
用 pool.getQueue() 做业务判断 | 队列是内部结构，外部读写会破坏语义 | 只读快照用于监控 |
什么时候不要用固定大小的平台线程池：任务大量阻塞在下游 IO 上（等 RPC、等文件）、并发量在万级、每个任务的生命周期很长。这几种情况下，线程数由「下游能承受多少并发」决定，而不是 CPU 核数，继续加大池只会把压力转嫁给下游。第 14 章的虚拟线程和 14.3 节的信号量限流是更合适的工具。
小结
- 七个参数分成四段生效：core 决定是否直接建线程，queue 决定是否排队，max 决定队列满后能否扩容，handler 决定最后怎么办。
execute的顺序是 core → queue → max → 拒绝；max排在队列之后，所以队列没满时它一次都不会被读到。- 「先入队再扩容」的理由是线程创建昂贵、扩容后难回收、排队是可观测的背压；代价是长任务会在队列里等待。
- 本机实测
core=2/max=4/queue=2：前 2 个任务建核心线程，第 3、4 个入队，第 5、6 个才扩容，第 7 个起被拒绝。 - 核心线程默认不超时；
allowCoreThreadTimeOut(true)后实测poolSize从 2 归零，且要求keepAliveTime > 0。
下一节处理队列满之后的那一步：四种拒绝策略各自适合什么场景，CallerRunsPolicy 为什么能形成反压，以及有界队列的容量到底怎么算。
阅读导航：上一节 9.3 读写锁与 StampedLock ｜ 下一节 10.2 队列与拒绝策略
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。