Java 8 Spring Boot用CompletableFuture处理大数据集堆内存溢出如何解决

• futures列表无限制累加所有CompletableFuture实例：所有批次的DB查询结果、处理中间结果、返回的WorkerResult都会被Future持有，直到所有任务执行完成才会释放，100万条数据对应的全量对象无法被GC，是堆溢出的核心诱因。
• Executors.newFixedThreadPool默认使用无界阻塞队列：只要任务提交速度快于处理速度，待执行任务就会不断在队列中堆积，每个任务绑定的批量DB数据会持续占用堆内存。
• 分页逻辑存在隐患：offset+limit的分页方式在offset过大时查询效率极低，且你当前的分页偏移量计算workerCount += rp +1会跳过1条数据，存在数据遗漏问题。

1. 自定义有界线程池，替换默认无界实现 #

不要用Executors工具类生成的线程池，手动定义ThreadPoolExecutor，设置有界队列和合理的拒绝策略，强制控制任务提交速度：

// 工作线程池：处理第三方API调用（IO密集型，线程数可根据第三方接口限流调整）
ExecutorService executorService = new ThreadPoolExecutor(
        15,
        15,
        0L, TimeUnit.MILLISECONDS,
        new ArrayBlockingQueue<>(20), // 有界队列，最多积压20个待处理任务
        new ThreadPoolExecutor.CallerRunsPolicy() // 队列满时由提交任务的主线程执行该任务，自动降速
);
// 结果处理线程池：处理DB更新
ExecutorService resultProcessor = new ThreadPoolExecutor(
        10,
        10,
        0L, TimeUnit.MILLISECONDS,
        new ArrayBlockingQueue<>(20),
        new ThreadPoolExecutor.CallerRunsPolicy()
);

2. 实现滑动窗口控制，避免Future无限制累加 #

不要等全量任务提交完成再统一等待，限制同时存在的未完成Future数量，完成一批就清理一批，释放内存：

int maxPendingTask = 30; // 最多同时存在30个未完成的任务，可根据内存调整
List<CompletableFuture<WorkerResult>> futures = new ArrayList<>();
long lastId = 0; // 改用游标分页，替换原来的offset

while (true) {
    // 游标分页查询：每次查上次最大id之后的rp条数据，效率远高于offset分页
    List<Model> dbRecords = repo.getDataByLastId(lastId, rp, date);
    if (dbRecords.isEmpty()) {
        break;
    }
    lastId = dbRecords.get(dbRecords.size() - 1).getId(); // 更新游标为当前批次最大id

    // 提交任务
    CompletableFuture<WorkerResult> future = CompletableFuture.supplyAsync(() -> {
        try {
            return new Worker(dbRecords).call();
        } catch (Exception ex) {
            throw new CompletionException(ex);
        }
    }, executorService).thenApplyAsync(result -> {
        service.resultReceived(result);
        return result;
    }, resultProcessor);
    futures.add(future);

    // 达到待处理任务阈值时，等待至少一个任务完成，清理已完成的Future
    if (futures.size() >= maxPendingTask) {
        CompletableFuture.anyOf(futures.toArray(new CompletableFuture[0])).join();
        // 移除所有已完成的Future，释放对应内存
        futures.removeIf(CompletableFuture::isDone);
    }
}

// 等待剩余所有任务执行完成
CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();

// 关闭线程池
executorService.shutdown();
executorService.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
resultProcessor.shutdown();
resultProcessor.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS);

3. 辅助优化项 #

• 调整批次大小rp：根据单条Model的内存占用，设置单批次拉取500-1000条即可，避免单批次数据过大占用过多内存。
• 轻量化中间对象：WorkerResult只保留后续DB更新需要的字段，不要把全量Model对象塞入结果中。
• 避免不必要的对象持有：Worker处理完批次数据后，及时清除对dbRecords的引用，方便GC回收。

内容的提问来源于stack exchange，提问作者Learning Developer