并发基础与 CompletableFuture
学完你要能自己回答
- 固定线程池 + 无界队列有什么风险?线程池参数应该怎么定?
- thenApply、thenCompose、thenCombine 分别解决什么组合问题?
- synchronized、ReentrantLock、volatile、Atomic 各自保证什么?
执行器:线程从哪里来01
不要直接 new Thread 管理业务并发。执行器把「任务提交」与「线程生命周期」解耦。JDK 21 之后,第一选择通常是每个任务一个虚拟线程,而不是固定平台线程池。
// 平台线程池:适合 CPU 密集、任务数量可控
try (ExecutorService pool = Executors.newFixedThreadPool(4)) {
Future<Integer> f = pool.submit(() -> 1 + 1);
System.out.println(f.get()); // 2
}
// 虚拟线程:适合大量阻塞任务
try (ExecutorService vt = Executors.newVirtualThreadPerTaskExecutor()) {
Future<String> f = vt.submit(() -> {
Thread.sleep(1000);
return Thread.currentThread().toString();
});
System.out.println(f.get());
}
Executors.newFixedThreadPool 使用无界队列。当任务提交速度超过消费速度时,队列会无限增长直到 OOM。生产环境更推荐显式构造 ThreadPoolExecutor,使用有界队列和明确的拒绝策略。
ThreadPoolExecutor pool = new ThreadPoolExecutor(
4, // corePoolSize
16, // maximumPoolSize
60, TimeUnit.SECONDS, // keepAlive
new ArrayBlockingQueue<>(200),
r -> new Thread(r, "biz-" + r.hashCode()),
new ThreadPoolExecutor.CallerRunsPolicy()); // 队列满时由提交线程执行
pool.execute(() -> handle(request));
pool.shutdown();
shutdown()不再接受新任务,等待已提交任务完成;shutdownNow()尝试中断并返回未执行任务。- JDK 19+ 的
ExecutorService实现了AutoCloseable,try-with-resources 的close()会等待任务结束。 - CPU 密集任务线程数约等于核数;I/O 密集任务在平台线程模型下需要更多线程,或改用虚拟线程。
- 拒绝策略:
AbortPolicy(默认,抛异常)、CallerRunsPolicy(背压)、DiscardPolicy(静默丢弃,慎用)。
CompletableFuture:组合异步结果02
CompletableFuture 的价值不在「异步执行」,而在组合:串行、并行、合并、异常恢复、超时。把回调金字塔改写成声明式流水线。
| 方法 | 语义 | 典型场景 |
|---|---|---|
supplyAsync / runAsync | 异步产生值 / 无返回值 | 入口 |
thenApply | 对结果做同步转换 | map |
thenCompose | 把返回 CF 的函数摊平 | 串行依赖(flatMap) |
thenCombine | 合并两个独立 CF | 并行聚合 |
allOf / anyOf | 等待全部 / 任意一个 | 批量任务 |
exceptionally | 异常时给默认值 | 降级 |
handle | 同时处理正常值与异常 | 统一出口 |
whenComplete | 完成后观察,不改变结果 | 日志、清理 |
orTimeout | 超时后异常完成 | 防悬挂 |
record User(String name) {}
record Order(String item) {}
record Profile(User user, List<Order> orders) {}
ExecutorService io = Executors.newVirtualThreadPerTaskExecutor();
CompletableFuture<User> userF = CompletableFuture
.supplyAsync(() -> loadUser(42), io);
CompletableFuture<List<Order>> ordersF = userF.thenCompose(user ->
CompletableFuture.supplyAsync(() -> loadOrders(user), io));
CompletableFuture<Profile> profileF = userF.thenCombine(
ordersF, Profile::new);
Profile profile = profileF
.orTimeout(2, TimeUnit.SECONDS)
.exceptionally(ex -> new Profile(new User("guest"), List.of()))
.join();
System.out.println(profile);
注意 thenApply 与 thenApplyAsync 的区别:前者可能由完成上游的线程直接执行,后者提交到执行器。没有指定执行器时使用 ForkJoinPool.commonPool(),不要在公共池里跑阻塞任务。
CompletableFuture<String> a = CompletableFuture.supplyAsync(() -> "A");
CompletableFuture<String> b = CompletableFuture.supplyAsync(() -> "B");
String combined = a.thenCombine(b, (x, y) -> x + y).join();
System.out.println(combined); // AB
CompletableFuture<Void> all = CompletableFuture.allOf(a, b);
all.join();
System.out.println(a.join() + b.join()); // AB
allOf 返回 CompletableFuture<Void>,需要再对每个 future 调 join() 取结果。任何一个 future 异常,allOf 会以该异常完成;需要部分成功时用 handle 包装每个子任务。同步原语:保证什么,不保证什么03
| 工具 | 保证 | 不保证 | 适用 |
|---|---|---|---|
volatile | 可见性、禁止重排 | 原子性 | 状态标志、单写多读 |
synchronized | 互斥 + 可见性 | 公平性 | 简单临界区 |
ReentrantLock | 互斥 + 可中断/超时/公平 | — | 需要 tryLock 或条件变量 |
AtomicInteger | 单变量 CAS 原子性 | 复合操作原子性 | 计数器 |
LongAdder | 高并发计数 | 精确即时快照 | 统计计数 |
ConcurrentHashMap | 单操作线程安全 | 复合操作原子性 | 共享映射 |
Semaphore | 许可数量控制 | — | 限流、资源池 |
CountDownLatch | 一次性的等待/放行 | 重用 | 启动协调 |
CyclicBarrier | 可重用的屏障 | — | 分阶段并行 |
// volatile 只保证可见性,i++ 仍不是原子操作
private volatile boolean running = true;
public void stop() { running = false; }
public void loop() {
while (running) {
// do work
}
}
ReentrantLock lock = new ReentrantLock();
if (lock.tryLock(1, TimeUnit.SECONDS)) {
try {
// 临界区
} finally {
lock.unlock();
}
} else {
// 获取锁失败,降级或返回
}
synchronized 在 JDK 21 的虚拟线程上还有特殊问题:在 synchronized 块内阻塞会钉住(pin)载体线程。虚拟线程代码里需要阻塞时优先用 ReentrantLock,或升级到 JDK 24+(JEP 491 已消除该限制)。
private final AtomicInteger counter = new AtomicInteger();
public int next() {
return counter.incrementAndGet();
}
// 高竞争计数可以考虑 LongAdder
private final LongAdder hits = new LongAdder();
public void hit() { hits.increment(); }
public long total() { return hits.sum(); }
并发集合与复合操作04
ConcurrentHashMap 保证单个方法线程安全,但「先检查再写入」仍然不是原子的。必须使用它提供的原子复合方法。
ConcurrentHashMap<String, Integer> counts = new ConcurrentHashMap<>();
// 错误:check-then-act
if (!counts.containsKey("a")) {
counts.put("a", 1);
}
// 正确
counts.putIfAbsent("a", 0);
counts.computeIfAbsent("b", k -> 0);
counts.merge("a", 1, Integer::sum);
System.out.println(counts); // {a=1, b=0}
computeIfAbsent的映射函数应短小,且不要在其中修改同一个 Map,否则可能死锁或抛异常。size()是近似值;并发场景不要依赖精确 size 做业务判断。Collections.synchronizedMap只包装单方法,遍历仍需手动同步。CopyOnWriteArrayList适合读多写极少;写操作会复制整个数组。BlockingQueue是生产者-消费者首选:ArrayBlockingQueue(有界)、LinkedBlockingQueue(可有界)、SynchronousQueue(直接交接)。
BlockingQueue<Task> queue = new ArrayBlockingQueue<>(100);
// 生产者
queue.put(task); // 队列满时阻塞
// 消费者
Task task = queue.take(); // 队列空时阻塞
// 非阻塞变体
if (queue.offer(task, 100, TimeUnit.MILLISECONDS)) {
// 入队成功
}
ThreadLocal 与线程上下文05
ThreadLocal 为每个线程保存独立副本,适合传递请求上下文。但在线程池中,线程会被复用,不清理就会泄漏,甚至把上一个请求的数据带给下一个请求。
private static final ThreadLocal<UserContext> CONTEXT = new ThreadLocal<>();
public void handle(Request req) {
CONTEXT.set(new UserContext(req.userId()));
try {
doHandle(req);
} finally {
CONTEXT.remove(); // 必须清理
}
}
虚拟线程数量可能非常多,每个线程一份 ThreadLocal 会放大内存占用。JDK 21 的 Scoped Values(预览)提供了不可变、结构化作用域的替代方案;JDK 25 转正。
// JDK 21 预览:需要 --enable-preview
static final ScopedValue<String> USER = ScopedValue.newInstance();
ScopedValue.where(USER, "alice").run(() -> {
System.out.println(USER.get()); // alice
});
// 离开作用域后 USER.isBound() == false
排障:死锁、线程转储与超时06
并发问题最终都要落到「线程现在在干什么」。先拿线程转储,再看锁等待关系,不要靠猜。
# 找到 Java 进程
jps -l
# 线程转储(平台线程;虚拟线程需 JDK 21 的 JSON 格式)
jstack <pid> > threads.txt
jcmd <pid> Thread.dump_to_file -format=json dump.json
# 死锁检测
jcmd <pid> Thread.print -l | grep -A 20 "Found one Java-level deadlock"
// 代码内检测死锁(适合健康检查)
ThreadMXBean mx = ManagementFactory.getThreadMXBean();
long[] deadlocked = mx.findDeadlockedThreads();
if (deadlocked != null) {
for (ThreadInfo info : mx.getThreadInfo(deadlocked)) {
System.err.println(info);
}
}
- 死锁四条件:互斥、持有并等待、不可抢占、循环等待。破坏任意一个即可。
- 统一锁顺序是最实用的预防手段;
tryLock超时可以打破无限等待。 CompletableFuture链中的阻塞调用会占住线程,异步代码里避免join()/get()嵌套。- 给外部调用设置超时:
orTimeout、HttpClient timeout、数据库 query timeout。
版本边界BOUNDARY
这张表比 API 本身更值得记。写代码时先确认你的运行时版本。
| 特性 / API | 状态 | 说明 |
|---|---|---|
| CompletableFuture | 8 正式 | JDK 9 增加 orTimeout、completeOnTimeout 等。 |
| ExecutorService AutoCloseable | 19 正式 | close() 等待任务结束。 |
| 虚拟线程 | 21 正式 | 见「虚拟线程」速查页。 |
| 结构化并发 | 21 预览 | JDK 21 需要 --enable-preview,后续版本仍多次预览。 |
| Scoped Values | 21 预览 / 25 正式 | JDK 21 不能直接用于生产。 |
费曼自检SELF CHECK
先合上页面,用自己的话回答,再展开对照。能讲出因果和反例,才算真正掌握。