如何让CompletableFuture真正并行处理大数据列表分区任务？

我有一个约5万条数据的大列表，需要对每个列表项执行操作，将处理结果映射为Java对象后写入CSV。原串行代码可以正常运行，但处理2000条数据耗时约4小时，因此打算用CompletableFuture实现异步处理来提速。我编写了并行处理代码：通过Executors.newFixedThreadPool创建线程池，将列表按500条分区后，利用Stream创建CompletableFuture并调用join，但发现各分区任务实际是串行执行（不同线程依次处理），未获得性能提升，希望实现各分区任务并行执行并最终汇总结果。

原串行代码 #

list.stream().map(listItem->
           service.methodA(listItem)
                    .map(result ->mapToBean(result, listItem))
    ).flatMap(Optional::stream).collect(Collectors.toList());

我的并行代码 #

ExecutorService service = Executors.newFixedThreadPool(noOfCores-1);
Lists.partition(list, 500).stream()
                .map(item-> CompletableFuture.supplyAsync(()-> executeListPart (item),service))
                        .map(CompletableFuture::join).flatMap(List::stream).collect(Collectors.toList());

当前代码串行执行的核心原因是普通Stream的map是顺序执行的：每调用一次map(CompletableFuture::join)，都会阻塞当前线程，必须等这个CompletableFuture执行完成后，才会处理Stream中的下一个分区任务。相当于逐个创建任务、逐个等待完成，完全没有并行性。

要实现真正的并行，需要先把所有CompletableFuture都创建并提交到线程池，再统一等待所有任务完成后收集结果，具体步骤如下：

1. 先收集所有分区对应的CompletableFuture，不立即调用join阻塞；
2. 使用CompletableFuture.allOf()等待所有异步任务完成；
3. 遍历所有CompletableFuture，调用join()获取结果并汇总。

修改后的代码示例 #

ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
try {
    // 1. 批量创建异步任务，暂不阻塞
    List<CompletableFuture<List<YourBean>>> futures = Lists.partition(list, 500)
            .stream()
            .map(partition -> CompletableFuture.supplyAsync(() -> executeListPart(partition), executor))
            .collect(Collectors.toList());

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

    // 3. 汇总所有任务结果
    List<YourBean> finalResult = futures.stream()
            .map(CompletableFuture::join)
            .flatMap(List::stream)
            .collect(Collectors.toList());

    // 写入CSV的逻辑...
} finally {
    // 关闭线程池，避免资源泄漏
    executor.shutdown();
}

• 线程池大小调整：如果service.methodA()是IO密集型操作（如调用外部接口、数据库查询），线程池大小可设置为CPU核心数*2甚至更高（比如Runtime.getRuntime().availableProcessors() * 4），避免线程因IO阻塞闲置，充分利用CPU资源。
• 分区大小优化：500条的分区大小可根据实际情况调整——若单个分区任务执行时间过长，可适当缩小分区；若任务过小，线程切换开销会增加，可适当增大分区。
• 线程安全检查：确保service.methodA()在多线程环境下是线程安全的，避免并发调用导致的数据异常。
• 异常处理：可在CompletableFuture中添加exceptionally()或handle()方法，处理异步任务中的异常，避免单个任务失败导致整个流程中断。

内容的提问来源于stack exchange，提问作者Anurag Srivastava