并发进阶

这一课你将能做到

1. 手动创建并优雅关闭一个线程池，背出 ThreadPoolExecutor 七大参数与任务提交流程
2. 用 CompletableFuture 组合多个异步调用（thenCombine / allOf / exceptionally），并解释它与 Future 的区别
3. 说清 volatile、synchronized、AtomicInteger 各自解决什么问题，给共享可变状态选对保护手段

为什么要学

Web 服务天生并发：Tomcat 一个线程处理一个请求，你的接口里再调下游、再算价格，全在共享同一个 JVM。线程池参数拍脑袋、在 Bean 里用普通 int 当计数器、异步链上吞异常——这些在生产环境的表现就是：高峰期线程耗尽、统计数据莫名丢失、接口偶发 500。

面试角度：并发是 Java 面试的最大高地。线程池七参数与拒绝策略、volatile 与 synchronized 的区别、CompletableFuture vs Future、ConcurrentHashMap 原理，几乎每场后端面试都会命中两道以上。本课答案按能背的标准写。

先建立图像

每请求 new Thread          线程池
┌──────────────┐      ┌─────────────────────────────┐
│ 请求1 → T1   │      │ 请求1 ─┐                     │
│ 请求2 → T2   │      │ 请求2 ─┤→ [核心线程×2]        │
│ ...          │      │ ...    ┤→ [阻塞队列等待]      │
│ 请求1万      │      │ 请求N ─┤→ [临时线程(≤max)]    │
│ → 1万线程 💥 │      │ 超限  ─┘→ [拒绝策略]         │
└──────────────┘      └─────────────────────────────┘
  创建销毁 + 上下文        线程复用、任务排队、超限保护
  切换把机器压垮

并发安全的三个层次：

手段	解决什么	代价
不共享（局部变量/不可变对象）	根本没有竞争	无
AtomicInteger / ConcurrentHashMap	单个读改写、容器级安全	CAS 自旋
synchronized / Lock	一段临界区互斥	阻塞、可能锁竞争

跟着做

步骤 1：线程池 + Future 批量汇总

新建 ConcurrencyDemo.java，完整可运行：

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class ConcurrencyDemo {

    private static final AtomicInteger okCount = new AtomicInteger();

    /** 模拟调用下游（如查订单金额），耗时 200ms */
    static int fetchAmount(String orderId) {
        try {
            Thread.sleep(200);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return 0;
        }
        okCount.incrementAndGet();
        return Math.abs(orderId.hashCode() % 100) + 100;
    }

    public static void main(String[] args) throws Exception {
        // 生产环境由 Spring 托管池；这里手动 new 是为了看清每个参数
        ExecutorService pool = new ThreadPoolExecutor(
                2, 4, 60, TimeUnit.SECONDS,          // 核心2、上限4、空闲60s回收
                new ArrayBlockingQueue<>(100),        // 有界队列
                r -> new Thread(r, "demo-worker"),    // 命名线程，排查日志靠它
                new ThreadPoolExecutor.CallerRunsPolicy()); // 拒绝策略

        List<Future<Integer>> futures = new ArrayList<>();
        for (int i = 1; i <= 4; i++) {
            String orderId = "order-" + i;
            futures.add(pool.submit(() -> fetchAmount(orderId)));
        }
        int total = 0;
        for (Future<Integer> f : futures) {
            total += f.get(); // Future.get 阻塞等待结果
        }
        System.out.println("订单总额=" + total + "，成功次数=" + okCount.get());

        pool.shutdown();
        pool.awaitTermination(10, TimeUnit.SECONDS);
    }
}

运行：java ConcurrencyDemo.java。4 个 200ms 的任务跑在 2 个核心线程上，总耗时约 400ms 而非 800ms——这就是线程池并行化的直观收益。

步骤 2：CompletableFuture 组合异步

把 main 里 Future 循环替换为：

        CompletableFuture<Integer> price =
                CompletableFuture.supplyAsync(() -> fetchAmount("price"), pool);
        CompletableFuture<Integer> discount =
                CompletableFuture.supplyAsync(() -> 20, pool);

        price.thenCombine(discount, (p, d) -> p - d)      // 两个结果汇合
             .thenAccept(pay -> System.out.println("应付=" + pay))
             .exceptionally(ex -> {                        // 异常不吞
                 System.out.println("查询失败: " + ex.getMessage());
                 return null;
             });

再体验并行等待：

        CompletableFuture<Void> all = CompletableFuture.allOf(price, discount);
        all.join(); // 阻塞直到两个都完成

步骤 3：亲眼看看非原子操作的丢失

    static int unsafeCount = 0;

    static void raceDemo() throws Exception {
        for (int round = 0; round < 3; round++) {
            unsafeCount = 0;
            okCount.set(0);
            Thread t1 = new Thread(() -> bump(10_000));
            Thread t2 = new Thread(() -> bump(10_000));
            t1.start(); t2.start();
            t1.join(); t2.join();
            System.out.println("普通int=" + unsafeCount + " AtomicInteger=" + okCount.get());
        }
    }

    static void bump(int times) {
        for (int i = 0; i < times; i++) {
            unsafeCount++;        // 读-改-写三步，非原子
            okCount.incrementAndGet(); // CAS 原子自增
        }
    }

运行三次 raceDemo()：普通 int 几乎每次都小于 20000，AtomicInteger 永远精确——count++ 不是一条指令，这是并发 bug 最常见的根源。

讲清楚要点

1. 线程池七大参数：corePoolSize（常驻）、maximumPoolSize（上限）、keepAliveTime + unit（临时线程空闲回收时间）、workQueue（阻塞队列）、threadFactory（命名/守护属性）、handler（拒绝策略）。执行顺序：核心线程 → 队列 → 临时线程 → 拒绝，注意是「先排队再扩容」。
2. 四种拒绝策略：AbortPolicy（默认，抛 RejectedExecutionException）、CallerRunsPolicy（提交线程自己跑，天然限流背压）、DiscardPolicy（静默丢弃，危险）、DiscardOldestPolicy（丢最老任务）。有界队列 + CallerRuns 是多数业务池的稳妥组合。
3. volatile 只保证可见性与禁止指令重排，不保证原子性：count++ 是读-改-写三步，volatile 挡不住丢失更新；要么 AtomicInteger（CAS），要么 synchronized 临界区。
4. CompletableFuture 与 Future 的区别：Future 只能阻塞 get；CompletableFuture 支持回调（thenApply/thenAccept）、组合（thenCombine/thenCompose/allOf）、异常传播（exceptionally/handle），并可显式指定 Executor。默认跑在 ForkJoinPool.commonPool 上，IO 密集任务必须显式传自己的池。
5. 容器选并发版本：多线程读写用 ConcurrentHashMap（JDK 8 后 CAS + synchronized 锁单个桶头），不要 Collections.synchronizedMap 全表锁；任务传递用 BlockingQueue（put/take 自动阻塞，生产者-消费者标配）。
6. 池大小估算：CPU 密集 ≈ 核数 + 1；IO 密集 ≈ 核数 × (1 + 等待时间/计算时间)。Web 请求线程（Tomcat）与业务线程池分开看，排障先查「池是否耗尽 + 超时配置」。

你可能会懵的地方

• Executors.newFixedThreadPool 不香吗？为什么手动 new？ 阿里规约禁用 Executors 工厂：FixedThreadPool 队列无界（OOM 风险），CachedThreadPool 线程数无界。手动 new 强制你显式给出队列容量与拒绝策略，风险前置。
• synchronized 和 ReentrantLock 怎么选？ 简单互斥用 synchronized（JVM 级优化成熟、自动释放）；需要尝试获取、超时、公平锁、多条件变量才用 ReentrantLock。锁不出错比锁得花哨重要。
• 线程池里跑阻塞 IO 会怎样？ 线程在 sleep/IO 期间不释放，池被占满后任务排队甚至触发拒绝——「假异步」仍是资源瓶颈。IO 密集要调大池或改用虚拟线程（JDK 21 正式版，本课不展开）。
• CompletableFuture 的异常在哪丢的？ 每个阶段独立传播，没有终端消费（join/get/exceptionally）时异常静默消失；调试技巧：链尾统一加 .whenComplete((r, ex) -> ...) 打日志。
• Spring @Async 和这有什么关系？ @Async 把方法提交到 TaskExecutor（本质就是 ExecutorService），不配置自定义池则用 SimpleAsyncTaskExecutor（每次新建线程，生产必须换掉）。模型与本课完全同源。

错法对照

错法	问题	正解
用 volatile 修饰计数器再 ++	可见性有了，原子性没有	AtomicInteger 或 synchronized
newFixedThreadPool + 无界队列	高峰积压直接 OOM	有界队列 + 显式拒绝策略
supplyAsync 不传 Executor	抢占公共 ForkJoinPool	显式传入自己的线程池
catch (InterruptedException) 只打印	中断信号被吞	恢复中断标志 Thread.currentThread().interrupt()
异步链没有终端异常处理	异常静默丢失	exceptionally/handle/whenComplete 收口

卡点急救

1. 程序跑完不退出：池线程是非守护线程在等任务。已无新任务则 shutdown() + awaitTermination；确认没有遗忘提交的任务，必要时 shutdownNow()。
2. 接口偶发 RejectedExecutionException：队列满 + 线程到 max 后触发拒绝。先看队列容量与流量是否匹配，再确认拒绝策略是否符合业务（丢弃不可接受就换 CallerRuns 或扩容）。
3. 统计数总是偏小：普通 long/int 跨线程自增丢更新；换 LongAdder（高并发统计场景比 AtomicInteger 竞争更小）或 AtomicLong。

面试常问

• ThreadPoolExecutor 的参数和任务提交流程？ 可背版：corePoolSize、maximumPoolSize、keepAliveTime、unit、workQueue、threadFactory、rejectedExecutionHandler。提交任务时：核心线程未满则新建核心线程执行；满了先入队；队列满且线程数 < max 则建临时线程；达到 max 再来任务执行拒绝策略。运行完空闲的临时线程按 keepAliveTime 回收。
• volatile 和 synchronized 的区别？ volatile：可见性 + 有序性，无原子性，无阻塞，只能修饰变量；synchronized：原子性 + 可见性 + 有序性，可修饰方法与代码块，存在锁竞争阻塞。计数场景 volatile 无效，必须原子类或锁。
• CompletableFuture 相比 Future 强在哪？ Future 只能阻塞式 get 一个结果；CompletableFuture 支持非阻塞回调链（thenApply/thenAccept）、多任务编排（thenCombine 汇合、thenCompose 串联、allOf/anyOf 并行）、异常传播与兜底（exceptionally/handle）、显式指定线程池——把「提交-等待」升级成可组合的异步管道。
• ConcurrentHashMap 在 JDK 8 里怎么保证线程安全？ 放弃分段锁，改为 CAS 初始化/插入空桶 + synchronized 锁定单个桶头节点，锁粒度从段细化到桶；读操作无锁，size 用 CounterCell 分散计数。

自测三问

1. 核心线程 2、队列容量 100、max 4：第 3 个任务来时去哪？第 107 个呢？
2. 为什么 count++ 加了 volatile 还是会丢更新？底层发生了什么？
3. thenApply 和 thenCompose 的区别是什么？

小结一句话

线程池控资源（七参数 + 拒绝策略），CompletableFuture 组合异步（回调 + 编排 + 异常收口），共享可变状态要么不共享、要么原子、要么加锁。

延伸

• java.util.concurrent 官方文档
• 下一课：依赖注入概念