承接的问题
10.1 的最后一个动作是 reject(command)。这一行背后是线程池最容易被忽略、也最容易在事故里被放大的决策：任务接不下了，怎么办？
三种选择摆在那里：抛异常让调用方知道系统满了；在调用方线程里跑，用调用方的时间换系统的稳定；或者丢掉，假装没发生过。JDK 把它们做成了四个现成的 RejectedExecutionHandler：抛异常、调用方执行、丢弃、丢最老的。它们各自适合什么场景、默认是哪个、为什么 CallerRunsPolicy 能形成反压、无界队列又为什么让整个机制失效——这是本节要回答的问题。
四种拒绝策略
RejectedExecutionHandler 只有一个方法 void rejectedExecution(Runnable r, ThreadPoolExecutor executor)。JDK 提供四个实现，都是 ThreadPoolExecutor 的静态内部类：
| 策略 | 行为 | 调用方感知 | 任务去向 |
|---|---|---|---|
AbortPolicy（默认） | 抛 RejectedExecutionException | 有异常 | 不执行 |
CallerRunsPolicy | 在提交任务的线程里直接 run() | 被拖慢 | 调用方线程执行 |
DiscardPolicy | 静默丢弃 | 无 | 消失 |
DiscardOldestPolicy | 丢掉队列头，再重试提交 | 无 | 老任务消失，新任务入队 |
两个细节藏在实现里：CallerRunsPolicy 和 DiscardOldestPolicy 都会先检查 executor.isShutdown()，已关闭时直接丢弃，不抛异常也不执行。这意味着关闭期间提交的任务，用这两个策略会静默消失。
实测：同一场景，四种结果
固定 core=1, max=1, queue=ArrayBlockingQueue(1)，任务睡 400 ms 并记录执行它的线程名。依次提交 A、B、C：A 占住唯一的工作线程，B 进队列，C 触发拒绝。
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
public final class RejectionProbe {
static final ConcurrentLinkedQueue<String> RAN = new ConcurrentLinkedQueue<>();
public static void main(String[] args) throws Exception {
runPolicy("AbortPolicy", new ThreadPoolExecutor.AbortPolicy());
runPolicy("CallerRunsPolicy", new ThreadPoolExecutor.CallerRunsPolicy());
runPolicy("DiscardPolicy", new ThreadPoolExecutor.DiscardPolicy());
runPolicy("DiscardOldestPolicy", new ThreadPoolExecutor.DiscardOldestPolicy());
CountingHandler counting = new CountingHandler();
runPolicy("Counting+CallerRuns", counting);
System.out.println("metrics rejected=" + counting.rejected.get()
+ " ranInCaller=" + counting.ranInCaller.get()
+ " lastThread=" + counting.lastThread);
unboundedQueue();
}
static void runPolicy(String label, RejectedExecutionHandler handler) throws Exception {
RAN.clear();
ThreadPoolExecutor pool = new ThreadPoolExecutor(
1, 1, 0, TimeUnit.SECONDS,
new ArrayBlockingQueue<>(1), r -> new Thread(r, "loan-worker"), handler);
String outcome = "none";
try {
submit(pool, "A");
submit(pool, "B");
submit(pool, "C");
} catch (RejectedExecutionException e) {
outcome = e.getClass().getSimpleName();
}
pool.shutdown();
boolean done = pool.awaitTermination(10, TimeUnit.SECONDS);
System.out.printf("%-19s outcome=%-24s ran=%s terminated=%s%n",
label, outcome, RAN, done);
}
static void submit(ThreadPoolExecutor pool, String id) {
pool.execute(() -> {
RAN.add(id + "@" + Thread.currentThread().getName());
try {
Thread.sleep(400);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
}
static final class CountingHandler implements RejectedExecutionHandler {
final AtomicInteger rejected = new AtomicInteger();
final AtomicInteger ranInCaller = new AtomicInteger();
volatile String lastThread = "-";
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
rejected.incrementAndGet();
lastThread = Thread.currentThread().getName();
if (!executor.isShutdown()) {
ranInCaller.incrementAndGet();
r.run();
}
}
}
static void unboundedQueue() throws Exception {
ThreadPoolExecutor pool = new ThreadPoolExecutor(
2, 8, 30, TimeUnit.SECONDS,
new LinkedBlockingQueue<>(), r -> new Thread(r, "loan-unbounded"));
for (int i = 0; i < 20; i++) {
submit(pool, "T" + i);
}
Thread.sleep(300);
System.out.println("unbounded core=2 max=8 submit=20 poolSize=" + pool.getPoolSize()
+ " queue=" + pool.getQueue().size()
+ " largest=" + pool.getLargestPoolSize());
pool.shutdownNow();
pool.awaitTermination(5, TimeUnit.SECONDS);
}
}
编译运行：
export JAVA_HOME=~/.jdks/jdk-25.0.5+8/Contents/Home
$JAVA_HOME/bin/javac --release 25 -d classes RejectionProbe.java
$JAVA_HOME/bin/java -cp classes RejectionProbe
本机输出：
AbortPolicy outcome=RejectedExecutionException ran=[A@loan-worker, B@loan-worker] terminated=true
CallerRunsPolicy outcome=none ran=[C@main, A@loan-worker, B@loan-worker] terminated=true
DiscardPolicy outcome=none ran=[A@loan-worker, B@loan-worker] terminated=true
DiscardOldestPolicy outcome=none ran=[A@loan-worker, C@loan-worker] terminated=true
Counting+CallerRuns outcome=none ran=[C@main, A@loan-worker, B@loan-worker] terminated=true
metrics rejected=1 ranInCaller=1 lastThread=main
unbounded core=2 max=8 submit=20 poolSize=2 queue=18 largest=2
逐条读：
- AbortPolicy：
outcome是RejectedExecutionException，A、B 正常执行，C 没有出现。异常抛在execute调用处，不处理就一路抛到栈顶；如果提交发生在裸线程的run()里，这个线程会随之终止。 - CallerRunsPolicy：C 出现在
ran里，线程名是main——它在提交任务的线程上跑完了，主线程在这 400 ms 里被完全占住。C@main排在 A 之前只是因为 A 的工作线程还没来得及记录，顺序取决于调度，不可依赖。 - DiscardPolicy：没有异常，
ran里只有 A、B。如果ran不是测试代码而是业务侧效果，这个任务就是彻底消失了，日志里一个字都没有。 - DiscardOldestPolicy：
ran是 A、C——B 被丢掉了。丢的是队列头，也就是等得最久的那个任务。 - Counting+CallerRuns：自定义策略记到了 1 次拒绝，并且在
main上执行。 - unbounded：无界队列下，
core=2, max=8的池提交 20 个任务，poolSize停在 2，队列涨到 18，largest是 2。max=8一次都没用上。
这些是结构性观测，不依赖 CPU 速度，因此在本机 load averages 140 以上的环境里仍然可以引用。任务里的 400 ms 只影响总时长，不影响判定。
DiscardOldestPolicy 还值得多想一步：FIFO 里最早入队通常也是最接近超时的任务，丢掉它等于把「快要过期」换成「刚刚提交」，延迟指标可能变好，业务上是随机丢失。
CallerRunsPolicy 的反压原理
反压（backpressure）的本质是：消费者处理不过来时，把压力传回生产者，让生产变慢。
CallerRunsPolicy 用最直接的方式实现了这一点——它不排队、不新建线程，而是在提交者的线程里同步执行任务。于是提交者「提交一个任务」的耗时从微秒级变成了「任务执行时间」，提交速率自然被压到处理速率，队列不再增长。
这和消息队列里的「生产者限流」是同一个闭环：上游请求 → 提交任务 → 队列满 → 调用线程亲自执行 → 上游请求变慢 → 提交速率下降。它成立需要三个前提：
| 前提 | 说明 | 违反后的后果 |
|---|---|---|
| 提交者是请求处理线程 | 拖慢它能直接减少上游流量 | 如果提交者是定时器/巡检线程，只是拖慢了自己 |
| 任务不持有调用方的锁 | 任务在调用线程里跑，若调用方持锁，锁被延长到任务结束 | 临界区膨胀，其他线程排队 |
| 任务本身不调用同一个池 | 否则是同池递归提交，仍可能再次被拒绝并再次内联 | 递归加深，最坏栈溢出 |
三个典型失效场景：在持锁代码里提交，锁的持有时间从「提交」变成「提交 + 任务」，吞吐断崖式下跌，还可能和任务内部的另一把锁形成死锁；调用方是 Web 容器的 IO 线程，内联执行占住 IO 线程，反压变成连接堆积；池已经关闭，策略直接丢弃，任务静默丢失。
因此 CallerRunsPolicy 适合「提交者就是业务请求线程、任务短、无锁」的场景；写操作、长任务、持锁路径都不适合。
无界队列的危害
实测最后一行是本节最该记住的数字：poolSize=2 queue=18 largest=2。
无界队列（LinkedBlockingQueue 无参构造，容量是 Integer.MAX_VALUE）带来三个连锁后果：
maximumPoolSize永远不生效。队列永远offer成功，第三段扩容分支永远走不到，池的线程数被钉死在 core。- 没有反压，只有延迟。任务不会失败，只会越排越久。上游看到的是「接口越来越慢」，而不是「系统满了」，于是继续加压，直到超时雪崩。
- 内存不可控。每个排队任务都持有闭包、上下文、请求体；队列涨到十万级时，OOM 只是时间问题。而且 OOM 发生时栈里看到的是业务代码，很难联想到队列。
Executors 的三个快捷工厂里有两个踩了这个坑：newFixedThreadPool(n) 是 core = max = n 加无界 LinkedBlockingQueue；newSingleThreadExecutor() 同上，n = 1，外面还包了一层代理。newFixedThreadPool 不仅是无界队列，还把 core 和 max 设成相等——就算队列有界，也不会有扩容。newCachedThreadPool 是另一个极端：core=0、max=Integer.MAX_VALUE、SynchronousQueue，队列不存任务，来一个建一个线程，高并发下线程数可以冲到几千。
生产代码的默认答案因此是：自己 new，队列必须有界。
有界队列容量怎么定
容量不能拍脑袋，可以从排队论最基本的公式推。Little 定律说，稳定系统里「平均排队任务数 = 到达速率 × 平均停留时间」；落到线程池上，用峰值而不是均值，就是 capacity ≈ 峰值 QPS × 可接受的排队秒数 - 并发线程数。
举例（示例推导，不是本机测量）：借阅查询峰值 400 QPS，一次查询平均 200 ms。按 Little 定律，需要的并发是 400 × 0.2 = 80；如果池里只有 8 个线程，无论队列多大都撑不住，应该先扩容或优化查询。假设下游能承受 80 并发、max=80，正常时队列接近空，容量只需覆盖抖动和下游短暂变慢：
| 场景 | 计算 | 建议容量 |
|---|---|---|
| 抖动 100 ms | 400 × 0.1 = 40 | 64 |
| 下游变慢到 500 ms | 400 × 0.5 - 80 = 120 | 128 |
| 下游变慢到 2 s | 400 × 2 - 80 = 720 | 不排队，直接拒绝 |
最后一行是关键判断：当排队时间已经超过上游超时时间时，排队毫无意义。上游 1 秒超时，任务在队列里等 2 秒，执行完也没人要结果，还占着线程和内存。这种情况下队列应该小、拒绝应该早。
容量的三条约束一起看：
- 延迟：容量 / 处理速率 ≤ 上游超时；
- 内存：容量 × 单任务持有内存 ≤ 可接受增量；
- 恢复：池被打满后，队列里的任务在多久内能清空。
队列类型也会影响容量语义：
| 队列 | 容量 | 特点 | 适用 |
|---|---|---|---|
ArrayBlockingQueue | 构造时固定 | 数组、单锁两条件、可选公平 | 容量确定的常规池 |
LinkedBlockingQueue | 有参时固定，无参近似无界 | 链表、吞吐略高 | 必须显式传容量 |
SynchronousQueue | 0（手递手） | 没有消费者就必然扩容 | newCachedThreadPool 式弹性池 |
PriorityBlockingQueue | 无界 | 按优先级出队 | 有优先级且另有总量控制 |
DelayQueue | 无界 | 到时间才能取 | 重试/定时，不适合做工作队列 |
PriorityBlockingQueue 和 DelayQueue 都是无界的，用它们做工作队列时，必须有别的机制兜住总量，否则同样会堆到 OOM。
自定义拒绝策略记录指标
AbortPolicy 抛异常是对的，但异常本身不告诉你「满了多久、排了多长、哪些任务被拒」。生产环境需要一个既记录指标、又保留快速失败的策略：
package com.plume.conc.core;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.LongAdder;
public final class MetricsRejectionHandler implements RejectedExecutionHandler {
private final String poolName;
private final LongAdder rejected = new LongAdder();
private final AtomicLong lastRejectNanos = new AtomicLong();
private final AtomicLong lastQueueSize = new AtomicLong(-1);
public MetricsRejectionHandler(String poolName) {
this.poolName = poolName;
}
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
rejected.increment();
lastRejectNanos.set(System.nanoTime());
lastQueueSize.set(executor.getQueue().size());
if (executor.isShutdown()) {
return; // 关闭期间不再抛，避免关闭流程被异常打断
}
throw new java.util.concurrent.RejectedExecutionException(
poolName + " saturated, queue=" + executor.getQueue().size()
+ " pool=" + executor.getPoolSize());
}
public long rejectedCount() {
return rejected.sum();
}
public long lastRejectNanos() {
return lastRejectNanos.get();
}
public long lastQueueSize() {
return lastQueueSize.get();
}
}
四条记录原则：
- 计数器用
LongAdder，拒绝路径可能被高频调用，AtomicLong在高竞争下会自旋。 - 不要在拒绝策略里打日志。队列满时往往是流量峰值，每个拒绝打一行日志会把磁盘和日志框架一起打爆；记录计数和最后一次的时间戳，由监控线程定期采样。
- 不要在拒绝策略里阻塞。有人会写
executor.getQueue().put(r)来「一定要接住」，这在池已满时会把提交线程永久阻塞，且put抛InterruptedException时任务直接丢失。 - 关闭期间区别对待。关闭时继续抛异常会让
shutdown流程里的清理代码不断失败；返回即可，调用方本来就不该在关闭后提交。
实测里的 CountingHandler 是最小版本，输出是 metrics rejected=1 ranInCaller=1 lastThread=main：一次拒绝、一次内联执行、发生在主线程。真实系统里这三个数字应该进监控，而不是进标准输出。
在 libraryd-conc 里落地
借阅是写操作，满了就该快速失败，不该让请求线程替它写库。PooledLoanService 的队列与拒绝策略定成这样：
// 领域方法 reserve/returnBook 与 close() 同 10.1，这里只给差异
public final class PooledLoanService implements LoanService, AutoCloseable {
private final BookRepository books;
private final LoanRepository loans;
private final ThreadPoolExecutor pool;
private final MetricsRejectionHandler rejection;
public PooledLoanService(BookRepository books, LoanRepository loans) {
this.books = books;
this.loans = loans;
this.rejection = new MetricsRejectionHandler("loan-pool");
AtomicInteger seq = new AtomicInteger();
this.pool = new ThreadPoolExecutor(
4, 16, 60, TimeUnit.SECONDS,
new ArrayBlockingQueue<>(32),
r -> {
Thread t = new Thread(r, "loan-pool-" + seq.incrementAndGet());
t.setUncaughtExceptionHandler((thread, error) ->
System.err.println("uncaught in " + thread.getName() + ": " + error));
return t;
},
rejection);
}
private <T> T submitAndWait(Callable<T> task, String what) {
try {
return pool.submit(task).get();
} catch (RejectedExecutionException e) {
throw new LoanException(ErrorCode.CONFLICT,
"loan pool saturated, rejected=" + rejection.rejectedCount()
+ " lastQueue=" + rejection.lastQueueSize());
} 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());
}
}
}
参数背后的判断：
core=4, max=16：查询型负载可以给大 max，写型负载 max 受数据库连接池约束，16 是「连接池 20 减去其他用途」的结果。queue=32：按「32 / 16 线程 ≈ 2 轮」估算，峰值时最多等两轮任务的时间；再多就该让上游收到 409。- 拒绝策略用「记录指标 + 抛异常」：写操作快速失败，调用方拿到
ErrorCode.CONFLICT后可以重试或降级，而不是无声等待。 - 异常消息里带上
rejectedCount和lastQueueSize，日志里能直接看到拒绝次数和当时的队列长度。
如果是只读查询（比如书目检索），可以换成 CallerRunsPolicy：让请求线程亲自查一次库，把压力传回上游，比直接报错更平滑。写路径不要这么做——请求线程内联写库会让事务边界和超时控制全部失控。
常见错误与边界
| 错误 | 后果 | 改法 |
|---|---|---|
| 用无界队列 | max 失效、内存无上限、只有延迟没有反压 | ArrayBlockingQueue 或带容量的 LinkedBlockingQueue |
| 在拒绝策略里打日志 | 峰值时日志风暴，磁盘打满 | 只记计数，由监控采样 |
在拒绝策略里 queue.put | 提交线程被永久阻塞 | 快速失败或 offer(timeout) |
用 DiscardPolicy 处理写请求 | 任务静默消失，数据不一致 | 写路径用 AbortPolicy |
依赖 CallerRunsPolicy 但持有锁 | 锁被延长到任务结束，吞吐暴跌 | 提交移出临界区 |
| 关闭后继续提交 | AbortPolicy 抛异常打断关闭流程 | 拒绝策略里判断 isShutdown |
queue=0（SynchronousQueue）配小 max | 稍有波动就拒绝 | 弹性池要把 max 设大，或换有界队列 |
| 队列容量按平均值算 | 峰值直接打满 | 按峰值 QPS × 可接受排队时间算 |
什么时候不要用拒绝策略兜底：当你还没有监控、没有容量评估时，拒绝策略只是把「慢慢崩」变成「立刻崩」。先把队列长度、拒绝次数、任务耗时的指标做出来（10.3 节），再决定用哪一种。
小结
- 默认策略是
AbortPolicy；CallerRunsPolicy在调用线程内联执行，DiscardPolicy静默丢弃，DiscardOldestPolicy丢队列头再重试。 - 本机实测同一场景：Abort 抛异常且 C 不执行；CallerRuns 的 C 在
main上执行；Discard 的 C 消失；DiscardOldest 丢掉 B、执行 C。 CallerRunsPolicy用「拖慢提交者」形成反压，前提是提交者就是请求线程、任务短、不持锁、不递归提交同一池。- 无界队列让
max彻底失效：实测core=2/max=8提交 20 个任务，poolSize停在 2、队列堆到 18。 - 有界队列容量按「峰值 QPS × 可接受排队秒数 - 并发线程数」估，并用上游超时和内存上限双向约束。
- 自定义拒绝策略只做三件事：计数、记录最后一次队列长度、快速失败；日志和阻塞都不要放进去。
下一节把线程池从「配好」推进到「跑起来之后」：指标怎么读、setCorePoolSize 改下去会发生什么、shutdown 与 shutdownNow 的两段式关闭怎么写。
阅读导航：上一节 10.1 线程池七参数与执行流程 ｜ 下一节 10.3 监控、动态调参与关闭
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。