⭐⭐⭐ Spring Boot 项目实战	⭐⭐⭐ Spring Cloud 项目实战
《Dubbo 实现原理与源码解析 —— 精品合集》	《Netty 实现原理与源码解析 —— 精品合集》
《Spring 实现原理与源码解析 —— 精品合集》	《MyBatis 实现原理与源码解析 —— 精品合集》
《Spring MVC 实现原理与源码解析 —— 精品合集》	《数据库实体设计合集》
《Spring Boot 实现原理与源码解析 —— 精品合集》	《Java 面试题 + Java 学习指南》

摘要: 原创出处 juejin.cn/post/7589893746081169449 「一只叫煤球的猫」欢迎转载，保留摘要，谢谢！

🙂🙂🙂关注微信公众号：【芋道源码】有福利：

RocketMQ / MyCAT / Sharding-JDBC 所有源码分析文章列表

RocketMQ / MyCAT / Sharding-JDBC 中文注释源码 GitHub 地址

您对于源码的疑问每条留言都将得到认真回复。甚至不知道如何读源码也可以请教噢。

新的源码解析文章实时收到通知。每周更新一篇左右。

认真的源码交流微信群。

前言

上周的某一天，运维同学找到我说生产环境的订单处理接口响应时间突然飙到了 3 秒。

我先让后端同学和运维一起查监控，CPU 使用率倒是不高，线程池的活跃线程数却涨到了 500 多。

再看了眼最近上线的代码，里面一段原本串行的代码，改成了 CompletableFuture 并行处理，本意肯定是想快一点，结果反而变慢了。

这也不是个例。

很多时候我们拿到一段串行代码，第一反应就是看能不能改成并行，觉得多线程跑起来肯定快。

但个别情况下，改完之后不仅不会变快，可能还会比原来还慢。

问题出在哪里？线程池不够大吗？还是机器配置太低吗？其实不然。

真正的原因，可能还是我们使用 CompletableFuture 的姿势不正确。

任务拆的太碎：上下文切换吃掉了所有收益

假设你要处理 1000 个订单号，每个订单号需要调用一个方法做校验，单次调用耗时 0.1 毫秒。串行处理总共需要 100 毫秒。你觉得用 CompletableFuture 并行处理，开 10 个线程应该能快 10 倍吧？

让我们写个测试：

// 串行处理
long start = System.nanoTime();
for (String orderId : orderIds) {
    validateOrder(orderId);
}
long serialTime = System.nanoTime() - start;

// 并行处理
ExecutorService executor = Executors.newFixedThreadPool(10);
start = System.nanoTime();
List<CompletableFuture<Void>> futures = orderIds.stream()
    .map(orderId -> CompletableFuture.runAsync(
        () -> validateOrder(orderId), executor))
    .collect(Collectors.toList());
CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
long parallelTime = System.nanoTime() - start;

结果串行用了 98 毫秒，并行用了 145 毫秒。

为什么会这样？

每提交一个 CompletableFuture，系统都要做一次任务调度，涉及线程唤醒、上下文切换、CPU 缓存失效。这些开销加起来可能就要几微秒。当单个任务本身只需要 0.1 毫秒时，调度开销占比就很高了。1000 个任务就是 1000 次调度，累积起来的开销超过了并行带来的收益。

更稳妥的做法是把粒度收粗，让每个任务足够大，至少覆盖调度成本。

我们改成每 100 个订单号作为一批，提交一个 CompletableFuture：

List<CompletableFuture<Void>> futures = new ArrayList<>();
for (int i = 0; i < orderIds.size(); i += 100) {
    List<String> batch = orderIds.subList(i, 
        Math.min(i + 100, orderIds.size()));
    futures.add(CompletableFuture.runAsync(() -> {
        for (String orderId : batch) {
            validateOrder(orderId);
        }
    }, executor));
}
CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();

这次只用了 12 毫秒。调度次数从 1000 次降到了 10 次，上下文切换的开销大幅减少，并行的优势才真正体现出来。

这其实就是最常见的策略——分批处理或分片处理。

当然还有一点经验主义，单个任务的执行时间至少要在 1 毫秒以上，并行才有意义。如果任务本身很轻量，就应该打包成批次再提交。

共享资源：所有线程挤在同一扇门前

并行处理时，如果多个线程需要访问同一个资源，这个资源就会成为瓶颈。最常见的就是数据库连接池。

比方说用 CompletableFuture 并行查询用户信息，每个 future 都要从连接池拿一个数据库连接。连接池配置了 20 个连接，但同时提交了 100 个 CompletableFuture。

结果就是 80 个线程在等待连接释放，白白浪费时间。

// 连接池只有 20 个连接
HikariConfig config = new HikariConfig();
config.setMaximumPoolSize(20);
DataSource dataSource = new HikariDataSource(config);

// 同时发起 100 个查询
ExecutorService executor = Executors.newFixedThreadPool(100);
List<CompletableFuture<User>> futures = userIds.stream()
    .map(userId -> CompletableFuture.supplyAsync(() -> {
        try (Connection conn = dataSource.getConnection()) {
            return queryUser(conn, userId);
        }
    }, executor))
    .collect(Collectors.toList());

监控显示，平均每个查询耗时从串行时的 5 毫秒涨到了 50 毫秒。大部分时间都花在了等待连接上。

除了数据库连接池，还有很多这种典型的场景案例：

锁与同步块
数据库连接池、Redis 连接池
HTTP 客户端连接池
限流器、熔断器
单线程日志 append、单队列缓冲

一般来讲，这种问题的应对思路通常是：

把共享区缩小，把锁粒度压缩到必要范围
让并发度受资源容量约束，别超过连接池和下游的可承受并发
对 I/O 做隔离池，避免同一个池里既跑 CPU 又跑阻塞 I/O

最简单的方式就是把线程池改成 20，当然还有另一个朴素但有效的手段，是用信号量把并发度卡在资源容量之内：

Semaphore semaphore = new Semaphore(20); // 不要超过连接池或下游并发上限
List<CompletableFuture<User>> futures = userIds.stream()
    .map(userId -> CompletableFuture.supplyAsync(() -> {
        try {
            semaphore.acquire();
            try (Connection conn = dataSource.getConnection()) {
                return queryUser(conn, userId);
            }
        } finally {
            semaphore.release();
        }
    }, executor))
    .collect(Collectors.toList());

如果你的任务需要获取这些资源，并发度就不能设得太高，否则大量线程会阻塞在资源竞争上。

内存爆炸：一口气提交十万个 Future

有个定时任务需要处理数据库里的所有待处理订单，数量大概十几万。后端同学写了这样的代码：

List<Order> orders = orderRepository.findAllPending(); // 10 万条
ExecutorService executor = Executors.newFixedThreadPool(50);

List<CompletableFuture<Void>> futures = orders.stream()
    .map(order -> CompletableFuture.runAsync(
        () -> processOrder(order), executor))
    .collect(Collectors.toList());

CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();

这段代码在测试环境跑得好好的，因为测试数据只有几百条。上了生产之后（草台班子没有做好代码审查 😭），JVM 直接 OOM 了。

问题在于这行代码：

.collect(Collectors.toList());

它会一口气创建 10 万个 CompletableFuture 对象，每个对象占用几百字节到上千字节不等。还没开始处理，光是这些 Future 对象就占用了几百 MB 内存。加上线程池队列里堆积的任务对象，内存很快就撑不住了。

正确的做法是分批处理：

List<Order> orders = orderRepository.findAllPending();
ExecutorService executor = Executors.newFixedThreadPool(50);

int batchSize = 1000;
for (int i = 0; i < orders.size(); i += batchSize) {
    List<Order> batch = orders.subList(i, 
        Math.min(i + batchSize, orders.size()));
    
    List<CompletableFuture<Void>> futures = batch.stream()
        .map(order -> CompletableFuture.runAsync(
            () -> processOrder(order), executor))
        .collect(Collectors.toList());
    
    CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
}

每次只创建 1000 个 Future，处理完一批再处理下一批。内存占用稳定在可控范围内。

这个问题的本质是缺少背压机制。

生产者一股脑地提交任务，完全不管消费者的处理能力。当任务提交速度远大于处理速度时，未处理的任务就会在内存里堆积，最终导致 OOM。

线程池配置：不是越大越好

遇到性能问题时，很多人的第一反应是加大线程池。原来是 50，改成 100，还是慢就改成 200。结果往往是线程越多越慢。

以前做过一个实验，用不同大小的线程池处理同一批 IO 密集型任务。每个任务需要调用一个外部 HTTP 接口，平均耗时 200 毫秒。一共 1000 个任务。

线程数	总耗时(秒)	CPU 使用率
10	20.5	15%
50	4.2	45%
100	2.8	68%
200	3.1	82%
500	4.8	91%

可以看到，线程数从 10 增加到 100 时，性能持续提升。但超过 100 之后，性能反而下降了。这是因为线程数太多会带来几个问题：

线程本身要占用内存，每个线程的栈空间默认是 1MB。500 个线程就是 500MB。

线程越多，操作系统的调度开销越大。CPU 要在不同线程之间频繁切换，每次切换都要保存和恢复上下文。

对于 IO 密集型任务，经验公式是 线程数等于 CPU 核心数乘以 (1 + IO 时间 / CPU 时间)。

假设 IO 时间是 200 毫秒，CPU 时间是 1 毫秒，那么 8 核机器的理想线程数是 8 * (1 + 200/1) = 1608 。

但这只是理论值，实际上很少需要这么多线程。

更实用的做法是从一个合理的初始值开始，比如 CPU 核心数的 2 到 4 倍，然后通过压测逐步调整，观察 CPU 使用率、响应时间和吞吐量，找到最佳的平衡点。

另一个常见错误是混用线程池。有的代码里创建了多个 ExecutorService，每个地方都用 newFixedThreadPool(50)。

实际运行时可能有 5 个这样的线程池，总共 250 个线程在跑。每个线程池单独看都不大，合在一起就超负荷了。

最好是在应用启动时创建一个全局的线程池，所有 CompletableFuture 共用这一个。

如果确实需要隔离，也应该仔细规划各个线程池的大小，确保总数可控。

异常处理：一个任务挂了，其他任务也白跑

CompletableFuture 有个特点，如果其中一个任务抛出异常，调用 allOf().join() 时会立即抛出异常，但其他任务不会停止，它们会继续在后台执行完。

List<CompletableFuture<Void>> futures = Arrays.asList(
    CompletableFuture.runAsync(() -> task1()),
    CompletableFuture.runAsync(() -> { throw new RuntimeException(); }),
    CompletableFuture.runAsync(() -> task3())
);

try {
    CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
} catch (Exception e) {
    // 进入这里时，task1 和 task3 可能还在执行
}

如果你的任务之间有依赖关系，这种行为会导致资源浪费。比如第一步查询订单失败了，后面的计算价格、更新库存都没有意义，但它们还在继续跑。

更糟糕的情况是，如果没有正确处理异常，有些任务可能会永远卡住。比如任务内部获取了锁但没有在 finally 里释放，抛出异常后锁就永远不会释放了。

建议每个任务都用 exceptionally 或 handle 方法捕获异常：

List<CompletableFuture<Result>> futures = tasks.stream()
    .map(task -> CompletableFuture.supplyAsync(() -> task.execute())
        .exceptionally(ex -> {
            log.error("Task failed", ex);
            return Result.failure(ex);
        }))
    .collect(Collectors.toList());

List<Result> results = CompletableFuture.allOf(
    futures.toArray(new CompletableFuture[0]))
    .thenApply(v -> futures.stream()
        .map(CompletableFuture::join)
        .collect(Collectors.toList()))
    .join();

这样可以确保异常被正确记录，而且不会中断其他任务。最后收集结果时，可以检查每个 Result 是成功还是失败，做相应的处理。

性能测试：看清并发度曲线

为了验证上面提到的几个问题，我们来做一组对比实验。任务场景是查询用户信息并计算积分，单次查询耗时约 10 毫秒。数据库连接池配置 20 个连接。

测试代码：

// 串行处理
long start = System.currentTimeMillis();
for (int userId : userIds) {
    User user = queryUser(userId);
    calculatePoints(user);
}
long serialTime = System.currentTimeMillis() - start;

// 并行处理（不同并发度）
for (int concurrency : newint[]{10, 20, 50, 100, 200}) {
    ExecutorService executor = Executors.newFixedThreadPool(concurrency);
    Semaphore semaphore = new Semaphore(20); // 限制数据库并发
    
    start = System.currentTimeMillis();
    List<CompletableFuture<Void>> futures = userIds.stream()
        .map(userId -> CompletableFuture.runAsync(() -> {
            try {
                semaphore.acquire();
                User user = queryUser(userId);
                calculatePoints(user);
            } finally {
                semaphore.release();
            }
        }, executor))
        .collect(Collectors.toList());
    
    CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
    long parallelTime = System.currentTimeMillis() - start;
    
    System.out.printf("Concurrency: %d, Time: %dms, Speedup: %.2fx%n",
        concurrency, parallelTime, (double)serialTime / parallelTime);
    
    executor.shutdown();
}

测试结果（处理 1000 个用户）：

Serial: 10200ms
Concurrency: 10, Time: 1050ms, Speedup: 9.71x
Concurrency: 20, Time: 530ms, Speedup: 19.25x
Concurrency: 50, Time: 540ms, Speedup: 18.89x
Concurrency: 100, Time: 650ms, Speedup: 15.69x
Concurrency: 200, Time: 890ms, Speedup: 11.46x

可以看到，并发度在 20 时达到最佳性能，这正好等于数据库连接池大小。再往上加，性能反而下降了。

我又去掉了信号量限制，让所有线程直接竞争连接池：

Concurrency: 50 (no semaphore), Time: 1200ms
Concurrency: 100 (no semaphore), Time: 1850ms

性能显著变差，因为大量线程阻塞在等待连接上。

最后测试任务粒度的影响。把 1000 个任务打包成不同批次：

1000 tasks individually: 2100ms
100 batches (10 tasks each): 180ms
10 batches (100 tasks each): 95ms
1 batch (1000 tasks): 10200ms (serial)

批次大小在 50 到 100 之间时效果最好，既减少了调度开销，又保留了并行度。

写在最后

开头的事故，后来怎么修复的？其实并不复杂：任务粒度收粗、I/O 和 CPU 分池、并发度按连接池容量卡住、批量提交加窗口、每个任务加超时与降级。

CompletableFuture 适合用在需要并发表达、需要组合与编排的场景。

吾日三省吾身。

使用CompletableFuture时，最好也先想清楚几个问题：单任务是否足够重、共享资源是否充足、下游是否能承受并发、队列是否有上界、失败和超时是否可控。有了答案，再去落地。

芋道源码 —— 纯源码解析博客

并行不等于更快：CompletableFuture 让你更慢的 5 个姿势

前言