一行CompletableFuture代码引发的P0级事故

苏三说技术

发布于 2025-03-10 14:42:38

5600

代码可运行

文章被收录于专栏：苏三说技术苏三说技术

运行总次数：0

代码可运行

昨晚凌晨 2 点，我司电商平台的订单服务突发崩溃。用户支付请求堆积超20万条，数据库连接池耗尽，直接损失预估百万级。

根本原因：一行未指定线程池的 CompletableFuture 代码，在高并发下触发默认线程池资源耗尽，导致任务队列无限堆积，最终内存溢出（OOM）。

你以为这只是偶然？数据揭示真相：

80% 的异步编程事故源于线程池配置不当；
90% 的开发者对 CompletableFuture 异常处理一知半解；
70% 的线上问题因任务依赖链断裂导致。

今天，我们通过这起真实事故，拆解 CompletableFuture 的正确使用姿势，教你实战避坑！

1. 事故还原

以下代码完全复现线上问题，请勿在生产环境运行：

public class OrderSystemCrash {

    // 模拟高并发场景
    public static void main(String[] args) {
        for (int i = 0; i < Integer.MAX_VALUE; i++) {
            processPayment();
        }
        // 阻塞主线程观察结果
        try {
            Thread.sleep(Long.MAX_VALUE);
        } catch (InterruptedException e) {
        }
    }

    // 模拟订单服务接口：支付完成后发送通知
    public static void processPayment() {
        // 致命点：使用默认线程池 ForkJoinPool.commonPool()
        CompletableFuture.runAsync(() -> {
            // 1. 查询订单（模拟耗时操作）
            queryOrder();
            // 2. 支付（模拟阻塞IO）
            pay();
            // 3. 发送通知（模拟网络请求）
            sendNotification();
        });
    }

    // 模拟数据库查询（耗时100ms）
    private static void queryOrder() {
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
        }
    }

    // 模拟支付接口（耗时500ms）
    private static void pay() {
        try {
            Thread.sleep(500);
        } catch (InterruptedException e) {
        }
    }

    // 模拟通知服务（耗时200ms）
    private static void sendNotification() {
        try {
            Thread.sleep(200);
        } catch (InterruptedException e) {
        }
    }
}

运行结果：

2. 问题分析

接下来，我们深入探究 CompletableFuture 的源码。

当我们运用 CompletableFuture 执行异步任务时，比如调用 CompletableFuture.runAsync(Runnable runnable) 或者 CompletableFuture.supplyAsync(Supplier<U> supplier) 这类未明确指定线程池的方法，CompletableFuture 会自动采用默认线程池来处理这些异步任务。

而这个默认线程池，正是ForkJoinPool.commonPool()。

下面，我们一同查看 CompletableFuture 中与之相关的源码片段。

public static CompletableFuture<Void> runAsync(Runnable runnable) {
    return asyncRunStage(asyncPool, runnable);
}

private static final Executor asyncPool = useCommonPool ?
    ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();
 
private static final boolean useCommonPool =
    (ForkJoinPool.getCommonPoolParallelism() > 1);

从代码可知：

runAsync 调用 asyncRunStage 并传入 asyncPool；
asyncPool 依据 useCommonPool 取值选定：
- useCommonPool 为 true 用 ForkJoinPool.commonPool()；
- 为 false 则用 new ThreadPerTaskExecutor()。
useCommonPool 取决于 ForkJoinPool.getCommonPoolParallelism()是否大于 1。
- 该方法返回ForkJoinPool.commonPool()的并行度（即线程数量，默认是系统 CPU 核心数减 1）。
- 若并行度大于 1，就以ForkJoinPool.commonPool()为默认线程池。

不过，话说回来，ForkJoinPool.commonPool() 作为默认线程池，到底存在哪些问题呢？

3. ForkJoinPool.commonPool() 的致命陷阱

1、全局共享：资源竞争的 “修罗场”

ForkJoinPool.commonPool() 是 JVM 全局共享的线程池，所有未指定线程池的 CompletableFuture 任务和并行流（parallelStream()）都会共享它。

这就像早高峰的地铁，所有人都挤在同一节车厢，资源争夺不可避免。

2、无界队列：内存溢出的 “导火索”

ForkJoinPool.commonPool() 使用无界队列，理论上能存储大量任务，但实际受内存限制。

大量任务到来时，队列会不断消耗内存，一旦超过系统承受能力，会触发 OutOfMemoryError，服务直接宕机。

4. 修复方案

public class OrderSystemFix {
    // 1. 自定义线程池（核心参数：核心线程数=50，队列容量=1000，拒绝策略=降级）
    private static final ExecutorService orderPool = new ThreadPoolExecutor(
            50, 50, 0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<>(1000), // 有界队列
            new ThreadPoolExecutor.AbortPolicy() { // 自定义拒绝策略
                @Override
                public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                    // 记录日志 + 降级处理
                    System.err.println("任务被拒绝，触发降级");
                    // 异步重试或写入死信队列
                }
            }
    );

    // 2. 修复后的订单服务
    public static void processPayment() {
        CompletableFuture.runAsync(() -> {
            try {
                queryOrder();
                pay();
                sendNotification();
            } catch (Exception e) {
                // 3. 异常捕获 + 降级
                System.err.println("支付流程异常：" + e.getMessage());
            }
        }, orderPool); // 关键：显式指定线程池
    }

    // 其他代码同上...
}

修复方案：