JUC 工具类与并发容器

S6-005建议学高级约 90 分钟JDK 21Java 高级与架构 · 并发编程

1. 本节定位

《Lock、CAS 与 AQS》讲了底层（CAS/AQS）；本节是日常开发真正在用的 JUC 工具：

text

① ConcurrentHashMap  —— 并发容器（读改写原子）
② Semaphore          —— 限流/资源池（共享模式 AQS）
③ CountDownLatch     —— 等待一组任务完成
④ 虚拟线程           —— 海量阻塞 IO（下节）

★ 本节实测（JucVirtualThreadLesson）：

text

CHM 100 线程 merge 10000 次无丢失 ✓✓
Semaphore(3) 峰值恰为 3 ✓✓

2. 学习目标

会用 CHM 的 merge/compute 做原子读改写 ★★★；
理解 Semaphore 是共享模式 AQS，用于资源池/限流 ★★★；
知道 CountDownLatch 与 CyclicBarrier 的区别 ★★；
会用 CompletableFuture 做异步编排：链式、allOf 并行聚合、异常兜底与超时 ★★★。

3. 前置知识检查

知道 AQS 独占/共享模式吗？（《Lock、CAS 与 AQS》）✓

4. 为什么需要这个知识

★ HashMap 并发下不仅丢数据，还可能死循环（JDK7 扩容） ✗✗； Collections.synchronizedMap 全表锁，吞吐极差 ✗✓； CHM 用 CAS + 分段/桶级锁，读几乎无锁 ✓✓✓

5. 核心概念

5.1 ConcurrentHashMap ★★★

java

MAP.merge(key, 1, Integer::sum);    // ★ 原子的「读-改-写」✓✓
MAP.computeIfAbsent(key, k -> new ArrayList<>()).add(x);   // ★ 原子的「不存在则建」✓✓

★ merge/computeIfAbsent 是按 key 原子的 ✓✓ —— 替代了恶心的「if (!map.containsKey) map.put」check-then-act ✗✗ （呼应《操作系统：调度、同步与死锁》与项目 A 的 tryBorrowOne ✓）

★ 分段与锁粒度 ✓：JDK8 起按桶（bin）加锁，读几乎无锁 ✓✓；size() 是近似值 ⚠️

5.2 Semaphore：共享模式 AQS ★★★

java

var sem = new Semaphore(3);
sem.acquire();      // state-1，0 则阻塞
try { ... } finally { sem.release(); }   // state+1

实测：10 个调用方、3 个许可 → 同时在临界区内的峰值恰为 3 ✓✓

★ 用途：连接池、限流、资源池 ——阈值 = 被保护资源的容量 ✓✓

5.3 CountDownLatch vs CyclicBarrier ★★

	CountDownLatch	CyclicBarrier
语义	等 N 个事件完成	N 个线程互相等到齐
可复用	不可 ✓	可（自动重置）✓
典型	主线程等子任务	分阶段计算

5.4 JUC 选择清单 ★★★

需求	工具
并发映射	ConcurrentHashMap ✓✓
限流/资源池	Semaphore ✓
等待一组任务	CountDownLatch ✓
异步编排（并行 IO/远程调用）	CompletableFuture + allOf ✓✓✓
高竞争计数	LongAdder ✓（低竞争 AtomicLong ✓）
延迟/周期任务	ScheduledThreadPoolExecutor ✓
阻塞队列	ArrayBlockingQueue（有界）/ LinkedBlockingQueue（⚠️ 无界）

5.5 CompletableFuture：Java 的异步编排 ★★★

是什么：Future 的增强版 —— Future 只能「提交任务 + 阻塞 get」✗； CompletableFuture 把结果变成可编排的流水线：链式转换、多任务聚合、异常兜底、超时控制，全程不阻塞 ✗✓。

为什么（一个结构性对比）✗✓：订单详情页要查 3 个远程服务（用户 120ms、优惠券 80ms、积分 150ms）——

写法	总耗时
三个 Future 依次 get ✗✗	120+80+150 = 350ms（相加）
`allOf` 并行等待 ✓✓✓	≈150ms（取最慢者）

★ 核心 API（六组就够用）：

类别	API	一句话
创建	`supplyAsync` / `runAsync`	有返回 / 无返回；IO 任务显式传线程池 ✗✓
转换	`thenApply`	同步转换（像 map）
串行依赖	`thenCompose`	下一步要用上一步的结果（像 flatMap）
聚合	`allOf` / `anyOf`	等全部 / 任一完成
异常	`exceptionally` / `handle`	兜底默认值 / 正常异常都处理
超时	`orTimeout` / `completeOnTimeout`（JDK 9+）	超时抛异常 / 超时给默认值

★★ 两个必踩的坑 ✗✗：

IO 任务用默认 ForkJoinPool.commonPool() ✗✗ —— 公用池被 IO 占满，全 JVM 的并行流和其他任务跟着遭殃；显式传自定义池 ✓✓
链上任何一步抛异常会沿链传播 —— 在链尾一个 exceptionally 统一兜底，不用每层 try-catch ✓✓（对拍实验见第 7 节）

6. 递进示例

6.1 原子计数

java

MAP.merge(word, 1, Integer::sum);   // 词频统计，无锁 ✓✓

6.2 限制并发下载

java

var sem = new Semaphore(5);         // 最多 5 个并发下载
for (var url : urls) {
    executor.submit(() -> { sem.acquire(); try { download(url); } finally { sem.release(); } });
}

6.3 反例：用 HashMap 做词频统计

java

// ✗ 多线程下 map.put(word, map.getOrDefault(word,0)+1) 丢计数 ✗✗

6.4 聚合三个远程调用（allOf 的结构性收益）

java

CompletableFuture<String> user   = CompletableFuture.supplyAsync(() -> queryUser(id),   pool);
CompletableFuture<String> coupon = CompletableFuture.supplyAsync(() -> queryCoupon(id), pool);
CompletableFuture<String> points = CompletableFuture.supplyAsync(() -> queryPoints(id), pool);
CompletableFuture.allOf(user, coupon, points).get(3, TimeUnit.SECONDS);   // 等全部
// ★ 总耗时 ≈ 最慢的那个，不是三者之和（实测见第 7 节）✓✓✓

7. 完整可运行代码与实测

代码：labs/java-deep/src/main/java/cn/kb/learn/deep/concurrency/JucVirtualThreadLesson.java、labs/java-deep/src/main/java/cn/kb/learn/deep/concurrency/CompletableFutureLesson.java

实测输出：

text

== 1. CHM: 100 threads x 100 merge ==
  total=10000 (expect 10000) OK
== 2. Semaphore(3): 10 callers ==
  peak inside = 3 (limit 3)

CompletableFuture 实测输出（CompletableFutureLesson，sleep 模拟 IO）：

text

===== 3. 聚合：三个并行查询，allOf =====
  用户:张三 / 券:满100减20 / 积分:1200
  总耗时 165 ms ≈ 最慢的积分查询（150ms），不是 120+80+150=350ms
  ★ 串行 get 三次 = 耗时相加 ✗✗；allOf 并行 = 耗时取最大 ✓✓✓
===== 4. 异常兜底与超时 =====
  异常兜底：降级默认值（原因：远程服务挂了）
  超时给默认值：超时默认值
  ★ exceptionally=兜底；orTimeout=超时抛异常；completeOnTimeout=超时给默认值 ✓

8. Debug 观察点

把 CHM 换成 HashMap：观察计数丢失 ✗✓；
Semaphore(1)：观察峰值变 1 ✓；
computeIfAbsent 里再操作同一 key：观察死锁/ISE ⚠️。

9. 常见错误

#	错误	说明
1	CHM 的 size() 当精确值 ✗	近似值 ✓
2	computeIfAbsent 里再改同 key ✗✗	死锁/ISE ⚠️
3	Semaphore 忘 release ✗✗	许可泄漏 → 永久限死 ✓✓
4	用 synchronized Map 替代 CHM ✗	全表锁吞吐差 ✓
5	无界队列 + 固定消费者 ✗	积压 OOM ⚠️
6	IO 任务用默认 ForkJoinPool ✗✗	公用池被占满，显式传池 ✓✓

10. 动手实验

跑 CHM 实验，把 CHM 换成 HashMap 对比 ✓✓；
Semaphore(1) 与 (10) 各跑一次，观察峰值 ✓；
用 CountDownLatch 改写项目 B 的并发测试等待逻辑 ✓。

验收方式：能说出每个 JUC 工具「在 AQS 里是独占还是共享」。

扩展任务：实现一个基于 Semaphore 的批量下载器（限并发 + 全部完成回调 + 超时取消）✓✓

11. 练习题

CHM 的 merge 相当于哪个 SQL 概念？（提示：原子读改写）
Semaphore(0) 配合 release 能实现什么？（提示：倒计时放行）
CountDownLatch 与 join 的区别？
LongAdder 为什么高竞争下更快？
设计：限制「每个用户同时最多 1 个下载任务」。
三个远程查询（120/80/150ms）用 allOf 聚合，总耗时约多少？为什么不是 350ms？

参考答案（要点）

条件 UPDATE / 原子读改写（对比项目 A 的 tryBorrowOne）✓★
初始 0 个许可 + N 次 release = CountDownLatch 等价物 ✓★
join 只能等线程；latch 可以等「任意 N 个事件」（不一定是线程结束）✓★
分段求和（Cell 数组），汇总时才相加 —— 把竞争摊开 ✓✓★
Semaphore(1) per user（按用户维度一个许可，或用键控限流 Map<user, Semaphore>）✓✓
≈150ms（最慢者） —— 三个任务在各自线程上并行执行，allOf 只是「等全部完成」；实测 165ms（多出的 15ms 是线程调度开销）✓✓★

12. 掌握标准

[ ] 会用 CHM 的 merge/computeIfAbsent ★★★；
[ ] 理解 Semaphore 共享模式与用途 ★★★；
[ ] 知道 CountDownLatch 与 CyclicBarrier 区别 ★★；
[ ] 知道 LongAdder 的分段原理 ★★；
[ ] 会用 CompletableFuture：链式（thenApply/thenCompose）、allOf 并行聚合、 exceptionally 兜底、超时，且 IO 任务显式传线程池 ★★★。

13. 扩展资源

14. 下一步学习

《虚拟线程》。

前置知识

S6-004 线程池与 ThreadPoolExecutor

后续知识

暂无

进度只保存在本机浏览器，可随时清除

外部学习资源

面试题库索引

项目与练手索引

工具与软件索引

JUC 工具类与并发容器

1. 本节定位

2. 学习目标

3. 前置知识检查

4. 为什么需要这个知识

5. 核心概念

5.1 ConcurrentHashMap ★★★

5.2 Semaphore：共享模式 AQS ★★★

5.3 CountDownLatch vs CyclicBarrier ★★

5.4 JUC 选择清单 ★★★

5.5 CompletableFuture：Java 的异步编排 ★★★

6. 递进示例

6.1 原子计数

6.2 限制并发下载

6.3 反例：用 HashMap 做词频统计

6.4 聚合三个远程调用（allOf 的结构性收益）

7. 完整可运行代码与实测

8. Debug 观察点

9. 常见错误

10. 动手实验

11. 练习题

12. 掌握标准

13. 扩展资源

14. 下一步学习

前置知识

后续知识

JUC 工具类与并发容器 ​

1. 本节定位 ​

2. 学习目标 ​

3. 前置知识检查 ​

4. 为什么需要这个知识 ​

5. 核心概念 ​

5.1 ConcurrentHashMap ★★★ ​

5.2 Semaphore：共享模式 AQS ★★★ ​

5.3 CountDownLatch vs CyclicBarrier ★★ ​

5.4 JUC 选择清单 ★★★ ​

5.5 CompletableFuture：Java 的异步编排 ★★★ ​

6. 递进示例 ​

6.1 原子计数 ​

6.2 限制并发下载 ​

6.3 反例：用 HashMap 做词频统计 ​

6.4 聚合三个远程调用（allOf 的结构性收益） ​

7. 完整可运行代码与实测 ​

8. Debug 观察点 ​

9. 常见错误 ​

10. 动手实验 ​

11. 练习题 ​

12. 掌握标准 ​

13. 扩展资源 ​

14. 下一步学习 ​

前置知识

后续知识

JUC 工具类与并发容器

1. 本节定位

2. 学习目标

3. 前置知识检查

4. 为什么需要这个知识

5. 核心概念

5.1 ConcurrentHashMap ★★★

5.2 Semaphore：共享模式 AQS ★★★

5.3 CountDownLatch vs CyclicBarrier ★★

5.4 JUC 选择清单 ★★★

5.5 CompletableFuture：Java 的异步编排 ★★★

6. 递进示例

6.1 原子计数

6.2 限制并发下载

6.3 反例：用 HashMap 做词频统计

6.4 聚合三个远程调用（allOf 的结构性收益）

7. 完整可运行代码与实测

8. Debug 观察点

9. 常见错误

10. 动手实验

11. 练习题

12. 掌握标准

13. 扩展资源

14. 下一步学习