# 第 31 章 并发工具、线程池与 AQS > 线程池决定"任务怎么被调度"，AQS 决定"抢不到锁的线程去哪排队"；前者是容量与背压，后者是并发包的共同骨架。 ## 学习目标 读完本章，你应当能够： 1. 说出 `ThreadPoolExecutor` 七个构造参数的作用，推演任务提交的四步流程与"为什么先入队而不是先扩容"，并判断三种工作队列下线程池的扩容行为与 `Executors` 四个工厂方法的故障模式； 2. 用 CPU 密集、IO 密集、Little 定律三种口径算出线程数，说出压测找拐点的观测指标，并说明 `ReentrantLock` 公平与非公平的实现差异； 3. 画出 AQS 的 `state` + 双向同步队列 + 单向条件队列结构图，说明共享模式为什么必须有 `PROPAGATE`，并辨析 `CountDownLatch`/`CyclicBarrier`/`Semaphore`/`Phaser` 的计数方向与阻塞对象； 4. 说清 `CompletableFuture` 默认线程池的隐患与 `LongAdder` 相对 `AtomicLong` 的退化条件。 ### 前置知识检查 | 你需要知道 | 在哪一节补齐 | |---|---| | `CAS`、`Unsafe` 原子操作与 Mark Word 锁升级 | 见第 30 章 30.4、30.5 节 | | `happens-before`、内存屏障与 `volatile` 语义 | 见第 30 章 30.2、30.3 节；AQS 的 `state` 为何是 `volatile` 依赖它 | | `ThreadLocal` 的 `ThreadLocalMap` 结构与泄漏根因 | 见第 30 章 30.7 节；本章只讲它与线程复用的组合故障 | | 线程与进程的本质区别、`LockSupport` | 见第 4 章 4.3 节；本章只讲它在 AQS 里承担的内核阻塞 | | 阻塞队列、链表、位运算与幂等语义 | 数据结构课程内容；本章直接进实现 | ## 31.1 ThreadPoolExecutor 的七个参数与四步提交流程 ### 31.1.1 七个参数各自管什么 ```java public ThreadPoolExecutor(int corePoolSize, // 常驻线程数下限，空闲也不回收 int maximumPoolSize, // 线程数上限，只在队列满时才被逼近 long keepAliveTime, // 超出 corePoolSize 的空闲线程存活多久后退出 TimeUnit unit, BlockingQueue workQueue, // 满 core 后任务的第一个去处 ThreadFactory threadFactory, // 造线程的地方，命名与异常处理器在此挂 RejectedExecutionHandler handler) // 线程数与队列都满时的最后一道闸 ``` 七个参数里只有 `maximumPoolSize` 与 `keepAliveTime` 作用在非核心线程上：**核心线程默认不超时**、永久存活；开启 `allowCoreThreadTimeOut(true)` 后核心线程也受 `keepAliveTime` 约束，等于把线程数下限交还给池。`prestartAllCoreThreads()` 一次性创建全部核心线程（`prestartCoreThread()` 只建一个），它把"第一次请求额外付一次线程创建成本（约 10 μs 量级，且栈内存要按页触底）"挪到启动阶段，也顺带在启动时暴露 `ulimit` 不足这类问题。 > [!WARNING] > `Executors.defaultThreadFactory()` 造出的线程**全是非守护线程（non-daemon）**，名字形如 `pool-N-thread-M`。非守护线程会阻止 JVM 退出，未命名会让 JStack 里几十个线程无法归属到业务。生产上必须自定义 `ThreadFactory`：命名带业务前缀、设为守护线程、并挂 `uncaughtExceptionHandler` 兜住未捕获异常。 ### 31.1.2 四步流程与"为什么队列优先" ``` execute(task) ▼ ① workerCount < corePoolSize ? ├── 是 ──▶ addWorker(task, true) 新建【核心】线程直接执行该任务，不入队 └── 否 ▼ ② workQueue.offer(task) 成功 ? ├── 是 ──▶ 入队；再校验池状态与线程数，必要时补一个线程 └── 否（队列满）▼ ③ workerCount < maximumPoolSize ? ├── 是 ──▶ addWorker(task, false) 新建【非核心】线程执行该任务 └── 否 ▼ ④ reject(task) → handler.rejectedExecution(task, this) ``` ② 成功后的"再校验"来自源码的 `recheck`：入队后若发现池已不是 `RUNNING`，就把刚入队的任务移出并走拒绝；若线程数为 0 则补一个线程，**否则核心线程刚超时退出时入队的任务会永远没人取**。源码写的是 `else if (isRunning(c) && workQueue.offer(command))`，**只有入队失败才走到第 ③ 步**。于是问题变成：为什么队列满之前不扩到最大线程数？**其一，线程昂贵而队列廉价**：一个平台线程的栈默认 1 MB（`-Xss`，32 位 JDK 上 512 KB），10000 个线程光栈就是 10 GB 虚拟地址空间，切换一次约 1~5 μs，1000 个线程在 8 核上轮转时大量 CPU 时间花在换栈上；**其二，扩容解决"持续阻塞"，队列解决"瞬时突发"**，短时尖峰交给队列慢慢消化比瞬间拉起几千线程再全部回收更划算；**其三，线程池无法识别任务类型，只能取安全默认值**——`execute` 只拿到一个 `Runnable`，不知道它会不会阻塞 I/O，若默认"先扩容"，混合型线程池会被阻塞任务把线程数推到 `maximumPoolSize`，CPU 被大量睡眠线程的换入换出拖垮。**先入队 = 先假设任务是 CPU 密集的**，这个假设最安全。 所以队列与 `maximumPoolSize` 的分工是：**队列抗突发，最大线程数抗持续阻塞**。要改变这个顺序，只能不排队——用 `SynchronousQueue` 让每次提交都必须立刻有线程接手。 > [!WARNING] > 第 ③ 步**每次提交只新建一个线程**：队列满后是"每来一个任务加一个线程"，要连续多次提交才能爬到 `maximumPoolSize`；而且此时新任务优先给新线程，队列里排在前面的老任务反而后执行。 ## 31.2 三种工作队列对行为的决定性影响 线程池的行为不由"线程数"决定，而由**队列的容量语义**决定。 | 队列 | 容量 | 会满? | 后果 | |---|---|---|---| | `ArrayBlockingQueue` | 构造时固定（如 500） | 会满 | 能触发扩容到 `maximumPoolSize`，再满才拒绝 | | `LinkedBlockingQueue` | **默认 `Integer.MAX_VALUE`** | 实际不会满 | `maximumPoolSize` 形同虚设，`maximumPoolSize − corePoolSize` 个线程永不创建，任务无限堆积直到堆 OOM | | `SynchronousQueue` | **容量 0**，不存储元素 | 永远"满" | 每次 `offer` 失败立即走第 ③ 步扩容，线程随任务增长到 `maximumPoolSize` 才拒绝 | | `DelayedWorkQueue`（`ScheduledThreadPoolExecutor` 专用） | 无界二叉堆 | 不会满 | `maximumPoolSize` 被固定为 `Integer.MAX_VALUE`，扩容等于无限 | `LinkedBlockingQueue` 的"等价于无界"要说透：它的 `maximumPoolSize` **不是被忽略，而是永远达不到**——`offer` 在无界队列上恒返回 `true`，第 ③ 步的代码路径永不执行。所以 `newFixedThreadPool(4)` 的真实语义是"最多 4 个线程 + 一个无上限的队列"，而不是"最多 4 个任务在跑"。 ## 31.3 四种拒绝策略与为什么必须自定义 | 策略 | 行为 | 提交方感知 | 适用场景 | 代价 | |---|---|---|---|---| | `AbortPolicy`（默认） | 抛 `RejectedExecutionException` | 立即感知 | 失败必须让上游知道、可重试的任务 | 未捕获异常会打断调用链 | | `CallerRunsPolicy` | 调用 `execute` 的线程**自己执行**该任务 | 隐式感知（线程被占） | 能容忍提交方变慢、但不允许丢任务 | 占住调用方线程（Tomcat 工作线程被占 → 接口整体变慢），形成背压 | | `DiscardPolicy` | 静默丢弃 | 完全无感知 | 几乎无正当场景 | 数据静默丢失，最危险 | | `DiscardOldestPolicy` | 丢弃队列中**最老**（队首）的任务后重试当前任务 | 无感知（丢的是旧任务） | 只关心最新值的场景 | 旧任务被无声吞掉；若队列是优先队列，丢的反而是优先级最高的 | `CallerRunsPolicy` 执行前会检查 `!e.isShutdown()`：这不是"关闭时还要执行任务"，而是**关闭阶段不再让调用方线程承担工作**，否则调用方会被拖住。生产环境为什么要自定义？因为四种都无法同时满足三件事：**可观测**（谁被拒、拒绝率多少）、**可感知**（提交方要知道，否则上层以为成功）、**可降级**（拒绝时走兜底而不是抛异常打穿链路）。 ```java // 拒绝路径由提交线程同步执行，所以这里只打点、不阻塞、不写重日志 public class BusinessRejectHandler implements RejectedExecutionHandler { @Override public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { REJECT_COUNTER.increment(); // 拒绝率本身就是最该告警的指标 if (r instanceof RetryableTask) { // 核心任务抛出让上游重试 throw new RejectedExecutionException("order pool saturated"); } Metrics.markDegraded(); // 非核心任务降级，绝不静默丢失 } } ``` ## 31.4 Executors 四个工厂方法的隐患 | 工厂方法 | 内部实际参数 | 隐患 | 后果 | |---|---|---|---| | `newFixedThreadPool(n)` | 核心 = 最大 = `n`，队列 `LinkedBlockingQueue`（无界） | 队列无界 | 峰值任务无限堆积，堆被 `Runnable` 与其闭包吃光 → `OutOfMemoryError: Java heap space`；线程数恒为 `n` | | `newSingleThreadExecutor()` | 核心 = 最大 = `1`，队列同上 | 队列无界 + 单线程 | 同源的堆积 OOM；一个慢任务阻塞全部后续任务 | | `newCachedThreadPool()` | 核心 `0`，最大 `Integer.MAX_VALUE`，队列 `SynchronousQueue` | 线程数上限是 `2³¹ − 1` | 提交快于执行时无限制建线程 → `OutOfMemoryError: unable to create native thread` | | `newScheduledThreadPool(n)` | 核心 `n`，最大 `Integer.MAX_VALUE`，队列 `DelayedWorkQueue` | 队列无界 + 线程数无界 | 任务堆积 OOM；线程数不随负载扩展 | > [!CAUTION] > 答"`Executors` 是官方工具类所以生产可用"会被直接判定为没读过生产事故。四者的根因只有两条：**要么队列无界（堆积 OOM），要么线程数无界（线程爆炸）**。背下这句，任何追问都能回到它上面。 手工构造的建议（8 核、订单接口这类 IO 密集任务）： ```java // 有界队列 + 有界线程数 + 命名线程 + 自定义拒绝，四个"有界"缺一不可 ThreadPoolExecutor pool = new ThreadPoolExecutor( 8, // 核心数按实测拐点设，不按核数拍 32, // 上限留突发余量，到达上限即拒绝，绝不无限扩张 60L, TimeUnit.SECONDS, // 非核心线程空闲 60 s 回收，缩容靠它 new ArrayBlockingQueue<>(500), // 必须显式给容量，让"满"成为可观测事件 new NamedThreadFactory("order-io-"), // 命名 + 守护线程 + 异常兜底 new BusinessRejectHandler()); // 拒绝时记录业务标识与拒绝量，不做重操作 ``` ## 31.5 线程数怎么确定：公式、推导与压测拐点 ### 31.5.1 CPU 密集：核数 + 1 记核数为 `N`。CPU 密集任务的时间几乎全在计算，`N` 个线程已能让 `N` 个核满载，第 `N + 1` 个的收益来自"当某个线程因缺页、`safepoint` 或 JIT 编译暂停时，还有线程能占住空闲的核"。同理 `ForkJoinPool.commonPool()` 的并行度取 `N − 1`（见 31.11 节），是为了给调用线程留一个核。线程数超过 `N` 后，多出的线程只增加上下文切换（约 1~5 μs/次）与缓存失效，吞吐不再上升。 ### 31.5.2 IO 密集：核数 × (1 + 等待时间/计算时间) ``` 设单任务：计算时间 W_c，等待时间 W_w（网络、磁盘、下游 RPC） 阻塞系数（blocking coefficient）：p = W_w / (W_c + W_w) 要让"等待期间 CPU 不空转"，并发线程数需放大约 1/(1 − p) 倍： N_threads = N × 1/(1 − p) = N × (W_c + W_w)/W_c = N × (1 + W_w/W_c) 算例：8 核、单任务 50 ms（计算 5 ms、等待 45 ms）⟹ N_threads = 8 × (1 + 45/5) = 80 校验：80 个线程里平均只有 80 × (5/50) = 8 个在计算，恰好铺满 8 核 ``` ### 31.5.3 混合型：用 Little 定律反推 ``` L = λ × W L = 系统中的平均请求数（即并发数） λ = 到达率（QPS） W = 单个请求的平均逗留时间（RT） 线程池是瓶颈时，所需线程数 ≈ 并发数 = 目标 QPS × 平均 RT 算例：目标 2000 QPS、平均 RT 50 ms = 0.05 s ⟹ L = 2000 × 0.05 = 100 ``` 线程数少于 100 时排队会把 RT 拉长，而 RT 变长又让 `L` 变大，形成"越慢越多、越多越慢"的正反馈。**这也是 `Semaphore` 能限流的原因**：`new Semaphore(50)` 限住并发 50，单请求 100 ms 时最大吞吐 `50/0.1 = 500 QPS`，与 `L = λ × W` 完全一致。 ### 31.5.4 压测找拐点 公式只给起点，真实线程数必须压测：预热（JIT、连接池、page cache、堆稳定）后从 `N_threads = N` 起步，每轮只调一个参数，观察 QPS、P99/P999、`getActiveCount()`、`getQueue().size()`、`getCompletedTaskCount()`、CPU 使用率，**继续加线程直到 QPS 不再上升而 P99 陡增**，该点即拐点。拐点的三个特征同时出现：`getQueue().size()` 不再持续为 0、`getActiveCount()` 贴住线程数上限、P99 上翘而 QPS 持平；若加线程 QPS 完全不涨且 CPU 也没满，说明瓶颈在下游（数据库连接池、下游 RPC 配额），加线程只会加剧排队。 > [!IMPORTANT] > "线程数越多吞吐越高"错在**超过临界点后额外线程的成本高于收益**：每次切换约 1~5 μs，切换后 CPU 缓存与 TLB 局部性被破坏，CPU 把时间花在换栈与重新填充缓存上，有效计算时间反而减少；线程数继续上涨还会耗尽栈内存（每线程 1 MB 量级），触发 swap 或直接 `unable to create native thread`。 ## 31.6 可观测、钩子与动态调整 ### 31.6.1 三个核心指标与两个钩子 | 方法 | 语义 | 怎么用 | |---|---|---| | `getActiveCount()` | **近似值**，正在执行任务的线程数 | 与 `getPoolSize()` 一起看：长期贴住上限 = 线程不足或下游慢 | | `getQueue().size()` | 队列中待执行任务数 | 常态应接近 0；持续增长说明消费不过来，比 CPU 更早报警 | | `getCompletedTaskCount()` | **近似值**，累计完成数 | 取相邻区间差值，得到实际吞吐与单任务平均耗时 | | `beforeExecute` / `afterExecute` | 任务执行前后回调，同一工作线程内调用 | 提交时打时间戳、`beforeExecute` 里算差值，才能把"排队耗时"从"执行耗时"里剥离 | | `terminated()` | 池完全终止后回调 | 关闭时收尾、释放外部资源、上报 | `getActiveCount()` 与 `getCompletedTaskCount()` 遍历 `workers` 时不加锁（避免监控路径影响吞吐），**只能当趋势指标，不能当准确账本**，用它做资源计量或有状态判断都是错的。 ```java // 钩子在工作线程内执行，因此能做 ThreadLocal 与耗时统计而无需额外同步 @Override protected void beforeExecute(Thread t, Runnable r) { WAIT_HISTOGRAM.record(System.nanoTime() - ((TaskWrapper) r).enqueueNanos); // 排队耗时 MDC.put("taskId", ((TaskWrapper) r).id); } @Override protected void afterExecute(Runnable r, Throwable t) { MDC.clear(); // 线程会被复用，不清理就会把上一个任务的上下文留给下一个任务 } ``` ### 31.6.2 动态调整与 execute/submit/shutdown `setMaximumPoolSize` 调小时，超出新上限的线程会在空闲后被回收；`setCorePoolSize` 调大时源码会**主动创建新线程**补齐到新核心数，调小时**只影响后续的空闲回收判定，不会立刻回收已存在的核心线程**——这就是"缩容不回收核心线程"：核心线程默认不超时，只在下次取任务时看到"线程数 > 核心数"才按 `keepAliveTime` 退出。要打破它得开 `allowCoreThreadTimeOut(true)`，代价是低峰期后需重新建线程；调参也**不会改队列容量**（`ArrayBlockingQueue` 的 `capacity` 是 `final`）。 `submit` 内部把任务包成 `FutureTask` 再调 `execute`，因此带来两点：**有返回值**、**异常被吞**。`FutureTask.run()` 捕获取消异常与一切 `Throwable` 存进 `outcome`，所以**任务抛出的异常不会传回工作线程、工作线程不会因异常退出**，错误被延迟到 `future.get()` 时以 `ExecutionException` 抛出。 ```java pool.submit(() -> { throw new IllegalStateException("boom"); }); // 反例：异常既不到日志也不触发钩子 Future f = pool.submit(() -> doSomething()); try { f.get(3, TimeUnit.SECONDS); } catch (ExecutionException e) { log.error("async task failed", e.getCause()); // 必须取 getCause()，外层只是包装 } catch (TimeoutException e) { f.cancel(true); // cancel 只是发中断，任务不响应中断就仍在跑，见 31.7 节 } ``` > [!WARNING] > `submit` 吞异常是"任务明明失败、监控上却全是成功"的最常见来源。更隐蔽的一点：**`afterExecute(r, t)` 的 `t` 对 `submit` 提交的任务恒为 `null`**，因为异常在 `FutureTask` 内部就被截获了；只有 `execute` 提交的任务才会把异常传进 `t`。 | 维度 | `shutdown()` | `shutdownNow()` | |---|---|---| | 池状态与队列 | `RUNNING → SHUTDOWN`，继续执行完队列中全部任务 | `RUNNING → STOP`，返回未执行任务列表并清空队列 | | 正在执行的任务 | 不打断 | 对每个工作线程 `interrupt()` | | 何时真正终止 | 队列空且线程全退出 | 线程响应中断后退出；**不响应中断的任务仍会跑下去** | 两者都只是"请求关闭"，等待结束要用 `awaitTermination`。`shutdownNow` 的返回值是 `List`，队列里被丢弃的任务只能靠它拿回来，不取就等于静默丢数据。 ## 31.7 线程池的四种典型故障模式 | 故障 | 触发条件 | 现象 | 处置 | |---|---|---|---| | 线程被异常替换 | 用 `execute` 提交且任务抛出未捕获异常 | `runWorker` 因异常退出，finally 里 `processWorkerExit` 移除该线程并按需补建；线程 id 不断变大 | 任务体内兜住异常；用 `submit` + `Future.get`；`ThreadFactory` 挂 `uncaughtExceptionHandler` | | `ThreadLocal` 数据串扰 | 线程被复用，上一个任务的 `ThreadLocal` 没清 | 用户 A 的上下文出现在用户 B 的日志/鉴权里；`ThreadLocalMap` 的 `Entry` 强引用 value，长生命周期线程上永不回收 | `try/finally` 里 `remove()`；框架层成对清理（见第 30 章 30.7 节） | | 死锁：任务等自己提交的子任务 | 单线程池里任务 `future.get()` 自己 `submit` 的子任务；或两层池互等 | 任务永久挂起，队列积压，进程不报错也不前进 | 禁止在任务内同步等待自己提交的任务；子任务用独立线程池；改异步组合 | | `Future` 超时后任务仍在跑 | `get(timeout)` 抛 `TimeoutException` 后任务未被真正停止 | 调用方已放弃，任务还在写库/发消息，产生重复效果 | `cancel(true)` 只是发中断；任务必须检查 `Thread.interrupted()`；**取消不可靠**，副作用必须幂等 | 四种故障的共同背景是：**线程池把"创建线程"变成了"复用线程"，而复用的东西（线程、`ThreadLocal`、栈上的残留状态）不会自动清理**。追问"你怎么知道线程被替换过"：看 `threadFactory` 生成的线程名序号是否持续增长、日志里 `UncaughtExceptionHandler` 的记录。 ## 31.8 AQS 的完整实现 `AbstractQueuedSynchronizer`（AQS）是 `ReentrantLock`、`Semaphore`、`CountDownLatch`、`ReentrantReadWriteLock`、`ThreadPoolExecutor.Worker` 的共同骨架。它自己不做业务判断，只提供两件事：**一个可 CAS 的状态量**，与**一个抢不到就排队的等待队列**。 ### 31.8.1 state 的设计 ``` AbstractQueuedSynchronizer volatile int state; ← 语义由子类解释，AQS 只提供 CAS 与读写 Thread exclusiveOwnerThread; ← 独占模式记录持有者，支持重入判定 Node head; Node tail; ← 同步队列（双向），懒初始化 独占语义：state 0 = 空闲；1 = 已占用；n = 同一线程重入 n 次 ReentrantLock.tryAcquire → compareAndSetState(0, 1)； tryRelease 把 state 减到 0 才算真正释放（重入时返回 false，故不唤醒后继） 共享语义：state = 可用许可数，获取递减、释放递增 Semaphore(3) 初始 state=3；CountDownLatch(5) 初始 state=5，countDown 减 1 ``` 为什么用 `int`：32 位 CAS 在所有平台上都有硬件指令（x86 的 `lock cmpxchg`、ARM 的 `ldrex/strex`），而 64 位 CAS 在部分 32 位平台要退化或要求额外对齐；重入计数与许可数的量级远达不到 `int` 上限（`2³¹ − 1` ≈ 21 亿）。`state` 是 `volatile` + `compareAndSetState`，于是"判断 + 修改"是原子的。条件队列的等待线程数用另一个普通 `int` 字段 `waiters` 记录，**不放进 `state`**——条件队列与同步状态是两个正交概念。 ### 31.8.2 CLH 队列的变体 AQS 队列是 Craig-Landin-Tarjan 自旋锁（CLH lock）的变体：原版是**单向链表 + 前驱节点上自旋**，AQS 改成**双向链表 + 显式 `park`**。 ``` head tail ┌───────────┐ prev ┌───────────┐ prev ┌───────────┐ │ Node(哑) │◀────────│ Node(T2) │◀────────│ Node(T3) │ │ ws = 0 │ ───────▶│ ws=SIGNAL │ ───────▶│ ws=SIGNAL │ next └───────────┘ next └───────────┘ next └───────────┘ 已释放/当前持有者 排队中，将要被唤醒 队尾，刚入队 Node.waitStatus（int 常量，新节点默认 0）： CANCELLED = 1 节点已放弃（超时/中断/异常），必须从队列中剔除 SIGNAL = -1 本节点释放时【有义务】唤醒 next；后继 park 前的"约定" CONDITION = -2 该节点当前在【条件队列】里，不在同步队列上 PROPAGATE = -3 共享模式下，唤醒仍需继续往后传播 ``` 为什么必须改成双向：原版靠前驱自旋看状态，而 AQS 用 `LockSupport.park` 真正挂起线程，唤醒时必须从队尾往前找第一个有效节点（`unparkSuccessor` 从 `tail` 反向遍历、跳过 `CANCELLED`）。正向 `next` 指针在并发入队/取消时可能为 `null` 或指向已取消节点，**只有 `prev` 链是可靠的**。 ### 31.8.3 独占模式：acquire 的三步 ```java // 模板方法：排队、阻塞、中断、超时的逻辑全在基类，子类只实现"能不能拿到" public final void acquire(int arg) { if (!tryAcquire(arg) // ① 快路径 && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) // ② 入队 ③ 队列内循环 selfInterrupt(); // 不在这里抛：把锁语义与中断语义分开，只在最后补中断标记 } final boolean acquireQueued(final Node node, int arg) { boolean interrupted = false; for (;;) { final Node p = node.predecessor(); // 只有前驱是 head 才有资格抢锁：head 代表已释放，可省掉无谓的 park/unpark if (p == head && tryAcquire(arg)) { setHead(node); // 自己成为 head；断开 p.next 避免正向遍历看到脏指针 p.next = null; return interrupted; } // 必须"先置 SIGNAL 再 park"：顺序反了，释放方的 unpark 会早于 park 而丢弃许可 if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) { interrupted = true; // 被中断不退出循环，只记账 —— 这就是"不可中断获取" } } } public final boolean release(int arg) { if (!tryRelease(arg)) { // 子类：state 减到 0 才返回 true（重入时返回 false，不唤醒后继） return false; } if (head != null && head.waitStatus != 0) { unparkSuccessor(head); // head 的 ws 为 0 说明没有后继在等，无需唤醒 } return true; } private void unparkSuccessor(Node node) { if (node.waitStatus < 0) { node.compareAndSetWaitStatus(node.waitStatus, 0); // 清 SIGNAL：唤醒义务已履行 } Node s = node.next; if (s == null || s.waitStatus > 0) { // 后继不存在或已取消，从尾部反向找有效节点 for (Node p = tail; p != node && p != null; p = p.prev) { if (p.waitStatus <= 0) { s = p; } } } if (s != null) { LockSupport.unpark(s.thread); // unpark 早于对方 park 也有效，许可不会丢 } } ``` `shouldParkAfterFailedAcquire` 的状态机是：前驱 `ws == SIGNAL` 才允许 park；前驱 `ws > 0`（`CANCELLED`）就向前跳过连续的取消节点；否则用 `compareAndSetWaitStatus(pred, ws, SIGNAL)` 把前驱置为 `SIGNAL`，**置失败就再循环一次抢锁**——这既避免 CAS 冲突下的忙等，也保证不会在没有后继信号的情况下 park。 ### 31.8.4 共享模式的 PROPAGATE：为什么必须有它 共享模式与独占模式最大的差别是：**一个线程释放后，多个后继线程应当被依次唤醒**（`Semaphore.release(3)` 要让 3 个等待者通过；`CountDownLatch` 归零要让所有等待者通过）。实现方式是 `doReleaseShared` 里**无条件循环传播**： ```java private void doReleaseShared() { for (;;) { Node h = head; if (h != null && h != tail) { // 队列中至少还有一个后继节点 if (h.waitStatus == Node.SIGNAL) { if (h.compareAndSetWaitStatus(Node.SIGNAL, 0)) { unparkSuccessor(h); // 交给被唤醒者继续传播（它会再触发本方法） } continue; // CAS 失败说明别人改过状态，重读再判断 } else if (h.waitStatus == 0 && !h.compareAndSetWaitStatus(0, Node.PROPAGATE)) { continue; // 关键：把 0 改成 PROPAGATE，而不是"什么都不做" } } if (h == head) { // head 没变说明传播已到队尾，退出 break; } } } ``` `PROPAGATE` 解决的是一个**丢失唤醒（lost wakeup）**竞态：初始队列为 `head(ws=0) → n1(SIGNAL) → n2(SIGNAL)` 且共享资源此刻已可获取。释放方 R 读到 `head.ws == 0`，若不置 `PROPAGATE` 就直接判"是否 `h == head`"；与此同时被唤醒者 T 重新 `tryAcquireShared` 成功并 `setHead`，自己也进入本方法，读到的是"自己这个 head，`ws == 0`"；随后 R 发现 `head` 已被 T 改变，判定"传播已到队尾"而 `break` 退出。此时若 T 在 `ws == 0` 时同样跳过、不 `unpark`，n2 就永远停在 `SIGNAL` 上无人唤醒——这就是丢失唤醒。根因是**"还要不要继续唤醒后继"这个信息没有在 `waitStatus` 里留下痕迹**：`ws == 0` 既可能是"没有后继要唤醒"，也可能是"我刚被交接了唤醒义务"。`PROPAGATE`（`-3`，小于 0）把后一种情况显式编码进状态，于是后续任何读到它的线程都会走 `unparkSuccessor` 继续传播，代价只是一个常量与一次 CAS。这个问题表现为 `CountDownLatch.await()` 偶发不返回、`Semaphore.acquire()` 偶发挂死，只在特定线程交织下出现、极难复现。 ### 31.8.5 Condition 的条件队列与 await/signal 迁移 `ConditionObject` 是 AQS 的内部类，维护一条**独立的单向链表**：`firstWaiter` → `nextWaiter` → `lastWaiter`，节点复用 `Node` 但 `waitStatus = CONDITION`。 ``` 同步队列（双向，抢锁用） 条件队列（单向，等条件用） head ──▶ [T2] ⇄ [T3] ──▶ tail firstWaiter ──▶ [T7] ─▶ [T8] ──▶ lastWaiter ws=CONDITION，用 nextWaiter 串联 ▲ │ │ await: 释放锁 + 入条件队列 + park │ signal: 把首节点摘下，用 CAS 接回同步队列尾部（不转移锁） └────────────────────────────────────────┘ ``` `await()` 的顺序必须是"**先释放锁、再入队、后 park**"：① `addConditionWaiter()` 把自己以 `CONDITION` 状态挂到条件队列；② `fullyRelease(node)` 释放**全部**锁（含重入计数）——不释放，`signal` 方根本进不来；③ `while (!isOnSyncQueue(node)) { park(this); checkInterruptWhileWaiting(node); }`，因为被唤醒后可能仍在条件队列上（伪唤醒），必须循环确认；④ `acquireQueued(node, savedState)` 重新竞争锁，拿回后把 `state` 恢复到 `savedState`，所以重入次数不会因 `await` 丢失。`signal()` 走 `doSignal` → `transferForSignal(node)`：用 CAS 把节点 `waitStatus` 从 `CONDITION` 改为 0（**改失败说明该节点已被取消，跳过**），再 `enq(node)` 接进同步队列尾部；若接回后前驱已被取消或 `SIGNAL` 置位失败，则**立即 `unpark` 该线程**。两个高频追问的答案就在这里：**为什么 `await` 必须写在 `while` 循环里**——从 `signal` 到真正抢到锁之间条件可能再次不成立（还有伪唤醒），单次 `if` 是标准 bug；**为什么 `signal` 后调用方还要 `unlock`**——被唤醒者要回到 `acquireQueued` 重新抢锁，不释放它只会从条件队列迁到同步队列后继续阻塞。 ### 31.8.6 模板方法模式在 AQS 中的体现 | 待实现方法 | 语义 | 典型实现要点 | |---|---|---| | `tryAcquire(int)` | 独占获取 | 非公平版直接 `compareAndSetState(0, 1)`；公平版先查 `hasQueuedPredecessors` | | `tryRelease(int)` | 独占释放 | 减 `state`，减到 0 才清 `exclusiveOwnerThread` 并返回 `true`（`true` 才触发唤醒） | | `tryAcquireShared(int)` | 共享获取 | 返回值语义特殊：**负数=失败，0=成功但后续不传播，正数=成功且可继续唤醒后继** | | `tryReleaseShared(int)` | 共享释放 | 用 CAS 循环改 `state`；只有"释放后允许更多获取"时才返回 `true` | | `isHeldExclusively()` | 是否被当前线程独占 | `exclusiveOwnerThread == Thread.currentThread()`，`Condition` 用它做状态校验 | | `hasQueuedPredecessors()` | 队列中是否有前辈 | AQS 提供；它是公平性判定的唯一入口 | > [!IMPORTANT] > `tryAcquireShared` 的返回值必须能区分三种情况，因为 `doAcquireShared` 要用它决定"要不要继续唤醒后继"。`CountDownLatch.Sync.tryAcquireShared` 返回 `(state == 0) ? 1 : -1`，正是"归零了才让所有等待者一起通过"的落点。 ## 31.9 ReentrantLock：公平、非公平与 synchronized 对比 非公平锁的插队只可能发生在 `state == 0` 的瞬间： ```java // 非公平：不检查队列，上来就抢一次 —— 插队就发生在这一行 final boolean nonfairTryAcquire(int acquires) { final Thread current = Thread.currentThread(); if (getState() == 0 && compareAndSetState(0, acquires)) { // 完全无视队列中已等待的线程 setExclusiveOwnerThread(current); return true; } if (current == getExclusiveOwnerThread()) { setState(getState() + acquires); // 重入 count++：只有持有者能走到这里，无需 CAS return true; } return false; } // 公平：唯一的差别是"锁空闲时还要求队列里没有前辈" protected final boolean tryAcquire(int acquires) { if (getState() != 0 || hasQueuedPredecessors() || !compareAndSetState(0, acquires)) { return false; // 队列里有前辈就退出，不去抢那一行 CAS } setExclusiveOwnerThread(Thread.currentThread()); return true; } ``` 锁被占用时新来的线程照样进队列，所谓"不公平"不是"永远抢不到"，而是**在锁交接的窗口里新线程可能抢在被唤醒的队首线程之前**。而队首此刻通常还停在 `park` 上、尚未执行，所以它这次失败并不算浪费一次执行——这正是非公平版本吞吐更高的原因。 | 维度 | 公平锁 | 非公平锁（默认） | |---|---|---| | 判定入口 | `hasQueuedPredecessors()` 必须先查队列 | 直接 CAS，不查队列 | | 唤醒成本 | 队首 `unpark` 后必须成功，无额外竞争 | 队首可能被抢占，需再 park 一轮（每次约 1~5 μs） | | 吞吐与延迟 | 吞吐较低；接近 FIFO，P99 更可控，无饥饿 | 吞吐较高；个别线程可能长时间得不到锁（理论上可饥饿） | 其余要点：`tryLock()` 无参版**不遵循公平策略**——看到 `state == 0` 就直接 CAS，所以公平锁上调它也可能插队成功（这是定义行为，不是 bug）；带超时的 `tryLock(long, TimeUnit)` 遵循公平性，并在超时/中断时把节点标记为 `CANCELLED` 并清理。`newCondition()` 可创建多个条件变量，是它相对 `synchronized` 的结构性优势。 | 对比项 | `synchronized` | `ReentrantLock` | |---|---|---| | 实现层次 | JVM 内建（`monitorenter`/`monitorexit`），靠 Mark Word 与 Monitor | Java 层 AQS + `LockSupport`（见第 30 章 30.5 节） | | 可中断 / 可超时 / 公平性 | 都不支持 | `lockInterruptibly()`、`tryLock(timeout, unit)`、可构造公平锁 | | 条件变量 | 一个（`wait`/`notify`，绑定锁对象） | 多个 `Condition` | | 释放方式 | 不可手动释放，随同步块结束 | 必须在 `finally` 里 `unlock()`，可跨方法持有 | ## 31.10 读写锁、StampedLock 与同步工具 `ReentrantReadWriteLock` 把读写状态压缩进**一个** `state`：高 16 位是读锁持有次数，低 16 位是写锁重入次数。由此推出三条硬结论：**读锁 → 写锁（升级）会死锁**，因为写锁要求所有读锁释放，而升级者自己持有的读锁不会自动释放（"升级"在语义上无法安全实现，两个读者同时升级就互相等待），正确做法是"释放读锁 → 重新获取写锁 → 重新校验数据"；**写锁 → 读锁（降级）允许**，同一线程持写锁时可再拿读锁，先释放写锁即完成平滑降级；**写饥饿**——默认非公平模式下只要读锁能连续获取，写锁就可能一直排队，可用 `new ReentrantReadWriteLock(true)` 打开公平模式。 `StampedLock` 用**乐观读（optimistic read）**换吞吐：`tryOptimisticRead()` 只返回一个版本戳、不写 `state`、不排队（成本接近读一个 `volatile`），随后直接读数据，再用 `validate(stamp)` 校验——期间有写线程获取过锁就会使版本戳失效，此时退化为 `readLock()` 重读一次。它吞吐更高的根本原因是**读路径完全不写共享状态**：读者之间没有 `state` 的 CAS 竞争，也不产生缓存行在核间的来回弹跳。代价同样明确：**不可重入**（重复获取会自己阻塞自己，没有 `exclusiveOwnerThread` 与重入计数）、**未实现 `Lock` 接口**、**不支持 `Condition`**、**不响应中断**。所以它适合"只读热点 + 极短写"，且 `readLock`/`unlock` 必须自己配对写对。 | 工具 | 可否重用 | 计数方向 | 谁等谁 | 归零时 | 实现基类 | |---|---|---|---|---|---| | `CountDownLatch` | **不可重用**（一次性，`state` 只减不增） | 减计数 | 主线程等 N 个子任务 | 所有 `await` 者被唤醒 | AQS 共享模式，初始 `state = N` | | `CyclicBarrier` | **可重用**（换代后计数复原） | 加计数 | **线程之间互相等** | 最后一个到达者执行 `barrierAction`，再唤醒全体并重置 | `ReentrantLock` + `Condition`，**不是** AQS 共享模式 | | `Semaphore` | 可重复获取/释放 | 许可数（获取减、释放加） | 拿不到许可者阻塞 | 许可用尽则后续 `acquire` 阻塞 | AQS 共享模式，初始 `state = permits` | | `Phaser` | 可重用，**参与者可动态注册/注销** | 加计数（`arrive`） | 按阶段（phase）互相等 | 进入下一阶段（phase 递增），可终止 | 自研 `state` | 最容易答错的是"谁等谁"：`CountDownLatch` 是**主线程等子任务**，子任务彼此无关；`CyclicBarrier` 是**参与线程互相等**，最后一个到达者兼职执行 `barrierAction`（它由到达的线程执行，做耗时操作会拖住这批线程）。`Phaser` 相对 `CyclicBarrier` 的两个增量是"参与者数量可在运行中变更"与"天然支持多阶段流水线"。 ```java // CountDownLatch：主线程等初始化步骤；计数只减不增，要复位必须新建对象 CountDownLatch latch = new CountDownLatch(3); pool.execute(() -> { try { init(svc); } finally { latch.countDown(); } }); // finally 里减：抛异常也要放行 latch.await(5, TimeUnit.SECONDS); // 必须带超时：子任务卡死时无超时的 await 会永久拖住启动流程 // CyclicBarrier：线程互等，最后一个到达者执行汇总；可重用，适合分代迭代 new CyclicBarrier(4, () -> mergePartialResults()).await(); // 任一人超时会让其他人收到 BrokenBarrierException // Semaphore：限流，是"线程数公式"的运行期版本 Semaphore permits = new Semaphore(50); // 并发 50、单请求 100 ms ⇒ 吞吐上限 50/0.1 = 500 QPS permits.acquire(); // 拿不到就排队，天然形成对下游的背压 try { callDownstream(); } finally { permits.release(); } // 必须 finally 释放，否则永久少一个许可 ``` ## 31.11 CompletableFuture 与 ForkJoinPool | 方法 | 输入函数形态 | 语义 | 常见误用 | |---|---|---|---| | `thenApply` | `T → U` | 同步转换前一步结果 | 函数里做阻塞调用，会占住完成前一步的那个线程 | | `thenCompose` | `T → CompletionStage` | 摊平嵌套 Future，串起真正的异步链 | 用 `thenApply` 返回 Future，得到 `CF>`，需要两轮 `join` | | `thenCombine` | `(T, U) → V` | 两个**互相独立**的 Future 都完成后合并 | 误用 `thenCompose`，把本该并行的两个调用写成串行 | | `allOf` | `CF...` | 全部完成，返回 `CF` | 以为它会把结果聚合成列表，其实要自己从各 Future 取 | | `anyOf` | `CF...` | 任一完成即完成 | 用于"多副本取最快"却没给其余请求取消/超时 | | `exceptionally` | `Throwable → T` | 异常时给默认值 | 忽略上游异常被吞（它只处理上游） | | `handle` | `(T, Throwable) → U` | 正常/异常两条路都走，且**改变结果类型** | 与 `whenComplete` 混用 | | `whenComplete` | `(T, Throwable) → void` | 只做副作用（日志、埋点），**透传原结果或原异常** | 用它做降级：返回值透传，异常照样往下抛 | `thenApply` 与 `thenCompose` 的差别只有一句：**函数返回的是值还是 `CompletionStage`**。前者在当前线程立即算完，后者把新阶段接到池里——把 IO 调用写在 `thenApply` 里，等于让完成上一步的线程去阻塞，异步链会被压回同步。 所有不带 `Executor` 参数的 `*Async` 方法都落到 `ForkJoinPool.commonPool()`，它的两个默认值决定了它不能承载业务 IO：**并行度默认是 `availableProcessors() − 1`**（8 核机器上只有 7 个工作线程，减 1 是为了给调用线程留一个核）；**它全 JVM 共享且不可扩容**（`parallelism` 在首次使用时确定）。因此提交**阻塞任务**（数据库查询、HTTP 调用）时，7 个线程会全部躺平在 IO 等待上，连锁反应是所有依赖 `commonPool` 的代码（包括并行流 `parallelStream()`）一起卡死，而**线程栈上看不到任何"忙"的迹象**，只有大量 `WAITING`/`park`——这是最典型的排查陷阱。 > [!CAUTION] > 三种正解：(1) 业务线程池显式传入——`supplyAsync(supplier, bizPool)`；(2) 用 JDK 9+ 的 `orTimeout(...)` 与 `completeOnTimeout(...)` 给每个阶段加超时；(3) 用虚拟线程（virtual thread）做每任务一虚拟线程的执行器。注意：自定义 `ForkJoinPool` 只把并行度改大了，阻塞的本质没变；`-Djava.util.concurrent.ForkJoinPool.common.parallelism=N` 也只是调参。 `ForkJoinPool` 本身的机制是**工作窃取（work stealing）**：每个工作线程有一个 `WorkQueue`（双端队列），**自己 fork 的任务 `push` 到 `base` 端、`pop` 也从 `base` 端取（LIFO）**，因为刚 fork 的子任务数据大概率还在缓存里；**空闲线程从别人队列的 `top` 端窃取（FIFO）**，从另一端拿能把两端操作者的竞争降到最低（这也是 `base`/`top` 用 `volatile` + CAS 而不是锁的原因）。这两条规则合起来既保住局部性，又让大任务切成小任务后能被全池消化——这正是 `parallelStream` 与 `RecursiveTask` 的吞吐来源，也是**它偏爱 CPU 密集短任务**的原因：窃取只能救"分工不均"，救不了"所有人都阻塞在 IO 上"。 > [!TIP] > 一句话区分：`ThreadPoolExecutor` 用**共享队列**做全局调度，队列是竞争点，适合任务粒度大、有阻塞、需要背压与拒绝语义的场景；`ForkJoinPool` 用**每线程队列 + 窃取**做分布式调度，几乎没有队列竞争，适合可递归拆分的 CPU 密集任务。 ## 31.12 并发容器选择与 CAS 的退化 `AtomicLong.incrementAndGet()` 是一个无限循环的 CAS：所有线程争抢**同一个** `value`，失败即重试，最坏情况下开销与竞争线程数同阶。`LongAdder` 把一份计数分段：`add(1)` 先 CAS `base`，失败则按 `(getProbe() & (cells.length − 1))` 定位到一个 `Cell` 再 CAS 该 `Cell`；被争抢的分片用 CAS 扩容 `Cell[]`（容量上限为最接近 `2 × 核数` 的 2 的幂），每个 `Cell` 用 `@Contended` 填充一个缓存行以避免相邻 `Cell` 伪共享；`sum()` 则是 `base + Σ cells[i]` 的普通读，不阻塞写。 它降低 CAS 竞争的原理是**分散热点**：竞争线程数固定时分片越多，两个线程碰撞同一分片的概率越低，CAS 失败率随之下降（空间换时间，代价是每个 `Cell` 占一个缓存行）。代价与边界必须一起记住： - **`sum()` 不是精确的瞬时值。** 它逐个读 `base` 与各 `Cell`，读取过程中其他线程可能正在改某个 `Cell`，`Cell` 之间也没有统一快照，所以 `sum()` 只是"某个瞬间的近似汇总"。它适合**监控指标、统计计数**（QPS、调用次数、命中率），**不适合**余额、库存、序号分配这类需要精确值的场景，也不适合"先 `sum()` 再 `compareAndSet`"这种复合判断；需要精确累加用 `AtomicLong`，需要复杂聚合用 `LongAccumulator`。 - **低竞争时 `AtomicLong` 可能更快**：没有分片数组的间接寻址与内存占用，路径更短。 `AtomicReference` 的 ABA 问题：`compareAndSet` 只比较**引用是否相同**，看不出"值被人改走又改回来"，典型形态是无锁栈 `pop` 里"头节点被换走又换回"，于是 CAS 成功但中间态已被破坏。解法是加版本号：`AtomicStampedReference`（`int` 版本戳）或 `AtomicMarkableReference`（布尔标记）。要补一句面试官爱听的：**ABA 在"值只是被覆盖"的场景无害，在"节点身份参与结构"的场景才致命**——判断要不要防 ABA，看的是数据结构有没有依赖"引用同一性"。 > [!NOTE] > `LongAdder` 是 `Striped64` 的子类，`DoubleAdder`/`LongAccumulator`/`DoubleAccumulator` 共享同一套分片机制。`ThreadLocalRandom.getProbe()` 生成的探针值决定线程落在哪个分片：让同一线程稳定命中同一分片（局部性）与让不同线程分散（降低冲突）是矛盾的，实现上选择了后者。 ## 本章小结 | 关键结论 | 一句话表述 | 常见错误 | |---|---|---| | 四步流程与队列容量 | 队列优先于扩容（`else if` 决定）；`LinkedBlockingQueue` 默认 `Integer.MAX_VALUE`，`maximumPoolSize` 永远达不到 | 答"核心满 → 扩到最大 → 再入队"；答"无界队列只是会变慢" | | 四工厂隐患 | 两条根因：队列无界（堆积 OOM）或线程数无界（线程爆炸） | 答"`Executors` 是官方推荐所以生产可用" | | 线程数公式 | CPU 密集 `N + 1`；IO 密集 `N × (1 + W_w/W_c)`；混合用 `L = λ × W` 反推 | 只背公式，说不出 `1/(1 − p)` 的推导与压测拐点 | | AQS state 与 CLH 变体 | `int` + `volatile` + CAS；双向链表 + `waitStatus` 四态，唤醒时**从尾部反向**找有效节点 | 答"state 只能是 0 和 1"、"队列是单向链表" | | `PROPAGATE` | 共享模式用 `-3` 解决 `doReleaseShared` 的丢失唤醒竞态 | 答"它只是共享模式的标记，没什么用" | | 同步工具 | `CountDownLatch` 减计数一次性；`CyclicBarrier` 加计数可重用且线程互等 | 说反"谁等谁"，或答 `CyclicBarrier` 也是 AQS 共享模式 | ## 自测题 **Q1**（推导）8 核机器上建了一个 `new ThreadPoolExecutor(8, 64, 60, SECONDS, new LinkedBlockingQueue<>(), factory, new AbortPolicy())`，随后提交 100 万个任务，每个任务耗时 1 ms。请说明线程数最终是多少、`AbortPolicy` 会不会触发、内存会发生什么。
答案与解析 **线程数恒为 8，`AbortPolicy` 永远不会触发，最终是堆内存 OOM。** 推导链：`LinkedBlockingQueue` 默认容量是 `Integer.MAX_VALUE`，所以 `execute` 第 ② 步的 `workQueue.offer(command)` 恒返回 `true`，第 ③ 步的扩容分支是死代码——`maximumPoolSize = 64` 形同虚设，`64 − 8 = 56` 个非核心线程永远不会创建。队列不断增长，100 万个 `Runnable` 连其闭包一起常驻堆，最终 `OutOfMemoryError: Java heap space`。拒绝策略只在第 ④ 步被调用，第 ④ 步走不到，所以 `AbortPolicy` 也不会抛——**最阴的一点：任务没被拒绝，只是永远排不到**。常见错误是答"会扩到 64 个线程"（把队列当成有界）或答"队列满了触发 `AbortPolicy`"（无界队列不会满）。 **面试官会追问**：(1) 换成 `ArrayBlockingQueue<>(500)` 后如何？——队列满才扩容，最高到 64 个线程；64 个也满了才开始拒绝。(2) 任务改成耗时 1 s 的阻塞调用呢？——吞吐掉到 8/s，堆积速度快 1000 倍，OOM 来得更早；此时应按 31.5.2 节把核心数提上去，或降低队列容量让拒绝早发生。
**Q2**（辨析）有人说"用 `Executors.newCachedThreadPool()` 只要设置好 `keepAliveTime` 就很安全，因为空闲线程 60 秒后会自动回收"。请判断对错。
答案与解析 **错，错在同一个前提上。** `keepAliveTime` 管的是**空闲回收**，与**峰值线程数**是两个无关维度。`newCachedThreadPool` 的核心数是 0、队列是 `SynchronousQueue`（不存储）、最大线程数是 `Integer.MAX_VALUE`：每来一个任务，若没有空闲线程立刻接手就走第 ③ 步新建线程，**没有任何上限**。提交速率持续高于执行速率时，线程数在 60 秒内可以涨到几千甚至上万；每个线程占 1 MB 量级栈空间，每次切换约 1~5 μs，CPU 大量时间花在换栈上，最终报 `OutOfMemoryError: unable to create native thread`（线程栈是本地内存，受 `ulimit` 与物理内存约束）。**正确做法**：`new ThreadPoolExecutor(0, 200, 60, SECONDS, new SynchronousQueue<>(), factory, handler)`——保留"不排队、直接移交"的语义，但**给最大线程数一个明确上限**，越界即拒绝并形成背压。 **面试官会追问**：(1) 为什么用 `SynchronousQueue` 而不是长度 1 的队列？——它的容量为 0，`offer` 在有空闲线程接手时成功、否则失败，正对应"有闲线程就复用、没有就新建"；用容量 1 的队列会让任务先被缓存，复用率下降。(2) 什么场景适合它的形态？——任务短小、突发性强、提交速率可控且能接受线程数波动；生产上仍应显式写明上限与拒绝策略。
**Q3**（推导）某订单接口目标 3000 QPS，压测测得单请求平均 RT 为 40 ms，其中 CPU 计算 8 ms，其余为下游 RPC 与数据库等待。机器 16 核。请分别用 Little 定律与 IO 密集公式估算线程数，说明不一致时以哪个为准，并给出压测找拐点的观测指标。
答案与解析 **Little 定律口径**：`L = λ × W = 3000 × 0.04 s = 120`，线程数至少 120 才不排队。**IO 密集口径**：`W_c = 8 ms`、`W_w = 32 ms`，`N_threads = 16 × (1 + 32/8) = 80`。**以哪个为准**：前者算的是"支撑目标吞吐所需的并发"，后者算的是"让 CPU 不空转所需的线程"，取**较大值**——按 80 个线程的吞吐上限只有 `80/0.04 = 2000 QPS`，达不到目标，故取 120（预留余量到 140~160）。若 120 个线程仍达不到 3000 QPS 且 CPU 未满，瓶颈不在线程池而在下游（数据库连接池上限、下游 RPC 配额），加线程只会让 RT 更差。 **拐点观测指标与特征**：`getActiveCount()` 是否贴住线程数上限、`getQueue().size()` 是否从 0 开始持续增长、`getCompletedTaskCount()` 的相邻区间差值给出实际吞吐，外加 P99/P999 与 CPU 使用率；**拐点 = QPS 不再上升而 P99 陡增、队列持续非空**。若加线程后 QPS 与 CPU 都不动，立刻停止加线程去查下游。**面试官会追问**：`L = λ × W` 的前提是系统处于**稳态**（到达率等于离开率、队列不无限增长）；队列持续堆积时算出的 `L` 只是下界，因为 RT 会因排队继续变大，形成正反馈。
**Q4**（推导/画图）请说明 AQS 共享模式中 `Node.PROPAGATE` 的必要性：如果 `doReleaseShared` 里处理 `ws == 0` 的分支改成"什么都不做直接跳过"，会出现什么问题？
答案与解析 **会出现丢失唤醒（lost wakeup）**：某个等待线程永远不会被 `unpark`，表现为 `CountDownLatch.await()` 偶发不返回、`Semaphore.acquire()` 偶发挂死，只在特定线程交织下出现、极难复现。时序是：t1 释放方 R 读到 `ws == 0`，若不置 `PROPAGATE` 就直接判"是否 `h == head`"；t2 被唤醒者 T 重新 `tryAcquireShared` 成功并 `setHead`，随后自己也调用本方法；t3 T 读到"自己这个 head，`ws == 0`"；t4 R 发现 `head` 已被 T 改变，判定"传播已到队尾"而退出；t5 若 T 在 `ws == 0` 时同样跳过、不 `unpark`，n2 就在 `SIGNAL` 上永久挂起——丢失唤醒。根因是**"要不要继续唤醒后继"这个信息没有在 `waitStatus` 里留下痕迹**：`ws == 0` 既可能是"没有后继要唤醒"，也可能是"我刚被交接了唤醒义务"。`PROPAGATE`（小于 0）把后一种情况显式编码进状态，于是后续任何读到它的线程都会走 `unparkSuccessor` 继续传播，代价只是一个常量与一次 CAS。常见错误是答"`PROPAGATE` 用来区分独占与共享模式"（区分靠节点构造时的 `shared` 字段）。**面试官会追问**：(1) 独占模式需要它吗？——不需要，独占一次只放行一个线程，`release` → `unparkSuccessor` 是一条直路。(2) 为什么 `acquireQueued` 被中断不直接抛异常？——为了把"不可中断获取"与"可中断获取"分成两条 API，`acquire` 只在最后 `selfInterrupt()` 补一个中断标记，既不吞掉中断也不破坏锁语义。
**Q5**（设计）一个服务用 `CompletableFuture.supplyAsync` 并发调用 5 个下游接口（1 个数据库查询 + 3 个 RPC + 1 个缓存），QPS 上来后出现"整机所有异步任务突然一起卡死，线程栈全是 `park`"。请指出根因、给出三种修复方案，并说明为什么 `Semaphore` 限流与线程数调整是同一个数学关系。
答案与解析 **根因**：不带 `Executor` 参数的 `supplyAsync` 落在 `ForkJoinPool.commonPool()`，并行度默认是 `availableProcessors() − 1`（8 核机器上只有 7 个线程），且**全 JVM 共享、不可扩容**。5 个下游调用都是阻塞 IO，7 个线程很快全部躺在 IO 等待上；此后任何依赖 `commonPool` 的代码（包括 `parallelStream()`）都排队等不到线程——线程既不忙也不报错，只是没有可用工作线程，所以线程栈上只有大量 `park`。 **三种修复**：(1) **显式指定业务线程池**，按 31.5 节公式算出线程数后建独立池，并让不同下游用不同池隔离（避免一个慢下游耗尽全部线程）；(2) **给每个阶段加超时与降级**——`orTimeout(...)` 或 `completeOnTimeout(default, ...)` 配合 `exceptionally` 给默认值；(3) **改用虚拟线程执行器**或在阻塞调用处换异步客户端。注意自定义 `ForkJoinPool` 或调 `common.parallelism` 只是把 7 改成 32，**没有改变"阻塞会耗尽池"这个本质**，属于调参不是修复。**`Semaphore` 与线程数的关系**：`L = λ × W` 中的 `L` 就是"同时在处理中的请求数"；线程池用**线程数**给 `L` 设上限，`Semaphore(n)` 用**许可数**给 `L` 设上限，约束的是同一个量——单请求 100 ms 时 `Semaphore(50)` 的吞吐上限是 `50/0.1 = 500 QPS`，与"50 个线程"完全等价。区别只在形式：线程池的许可与线程一一绑定（阻塞时线程被占），`Semaphore` 的许可可跨线程、跨阶段传递。 **面试官会追问**：(1) 怎么确认是 `commonPool` 的问题？——线程名前缀是 `ForkJoinPool.commonPool-worker-N`，`N` 不超过 `核数 − 1` 且状态全是 `WAITING`。(2) `thenApply` 和 `supplyAsync` 哪个更危险？——`thenApply` 更隐蔽：它不切换线程，会**占用上一个阶段完成时的那个线程**（可能正是 `commonPool` 的工作线程）。
## 面试高频问题与陷阱 | 面试问法 | 考查角度 | 陷阱 / 加分点 | 对应本章小节 | |---|---|---|---| | 线程池的四种拒绝策略分别是什么？你们生产环境用的是哪个？ | 拒绝语义与背压 | 错答"用默认策略"却说不出默认是 `AbortPolicy`、也不知道它会抛 `RejectedExecutionException`。加分点：`CallerRunsPolicy` 由提交线程执行形成背压（代价是占用 Tomcat 工作线程），`DiscardPolicy` 静默丢弃最危险 | 31.3 | | 线程池线程数应该设多少？CPU 密集和 IO 密集分别怎么算？ | 公式推导与压测方法 | 错答"设成核数两倍比较合适"这类口诀。加分点：`N + 1` 与 `N × (1 + W_w/W_c)` 的推导、用 `L = λ × W` 从目标 QPS 反推、压测看 `getActiveCount`/`getQueue().size()`/`getCompletedTaskCount` | 31.5 | | 请描述一下 AQS 的原理，`state` 是什么？队列是单向还是双向、为什么？ | AQS 状态与队列结构 | 错答"`state` 是 0 空闲 1 占用"就没了，或答"队列是单向链表"（那是原版 CLH）。加分点：`state` 是 `volatile int` + CAS，语义由子类定义；改成双向链表是为了唤醒时能从尾部反向找到第一个有效节点 | 31.8 | | AQS 的 `Node.waitStatus` 有哪几个值？`PROPAGATE` 是干什么用的？ | 共享模式的并发正确性 | 错答"就是标记节点状态的，`PROPAGATE` 没什么用"。加分点：`CANCELLED`/`SIGNAL`/`CONDITION`/`PROPAGATE` 四态加默认 0；`PROPAGATE` 解决 `doReleaseShared` 里 `ws == 0` 分支导致的**丢失唤醒** | 31.8.4 | | `CountDownLatch` 和 `CyclicBarrier` 有什么区别？`Semaphore` 又能解决什么问题？ | 同步工具的语义辨析 | 错答把"谁等谁"说反，或答"两者差不多，`CyclicBarrier` 只是能重置"。加分点：`CountDownLatch` 减计数、一次性、主线程等子任务；`CyclicBarrier` 加计数、可重用、最后一个到达者执行 `barrierAction`；`Semaphore` 管许可数即限流 | 31.10 | | `CompletableFuture` 的 `thenApply` 和 `thenCompose` 有什么区别？不传线程池会怎样？ | 编排语义与默认线程池 | 错答"都是接下一步，没区别"，或压根不知道默认用 `commonPool`。加分点：区别在返回值是值还是 `CompletionStage`；`commonPool` 并行度默认 `核数 − 1`、全 JVM 共享不可扩容，阻塞任务会把公共池连 `parallelStream` 一起拖死 | 31.11 | | `AtomicLong` 和 `LongAdder` 怎么选？`LongAdder.sum()` 为什么不能当精确值？ | CAS 退化与分段累加 | 错答"`LongAdder` 全面优于 `AtomicLong`"。加分点：`LongAdder` 用 `base + Cell[]` 分散 CAS 热点，代价是 `sum()` 为逐 `Cell` 的近似汇总（非原子快照），适合监控指标不适合余额库存 | 31.12 | ## 延伸阅读 - Oracle 官方文档：`java.util.concurrent.ThreadPoolExecutor`（七个参数语义、`prestartAllCoreThreads`、`allowCoreThreadTimeOut`、`RejectedExecutionHandler`）、`Executors`（四个工厂方法的等价参数与告警说明）、`AbstractQueuedSynchronizer`（模板方法清单与 `Node.waitStatus` 常量）、`Condition`、`ForkJoinPool`（`parallelism` 默认值与工作窃取说明）、`LongAdder`/`Striped64`（`sum()` 的近似性说明）。 - OpenJDK 源码：`ThreadPoolExecutor.execute`（三段 `else if` 与 `recheck`）、`runWorker`/`processWorkerExit`（异常替换线程的路径）、`AbstractQueuedSynchronizer.doReleaseShared`（`PROPAGATE` 的处理）、`ConditionObject.await`/`transferForSignal`、`ReentrantReadWriteLock`（`state` 高低 16 位的位域划分）。 - Doug Lea, *The java.util.concurrent Synchronizer Framework*（AQS 的设计说明，含 CLH 变体与 `PROPAGATE` 的引入动机）。 - 见第 30 章 30.2、30.3 节（`happens-before` 与内存屏障）、第 30 章 30.5 节（Mark Word 与锁升级）、第 30 章 30.7 节（`ThreadLocalMap` 结构与泄漏根因）、第 32 章 32.2 节（`TLAB` 与对象分配）。 --- > 作者: [破戒僧](https://onlyblacktea.cn) > URL: https://onlyblacktea.cn/book/backend/ch31/