如何用CompletableFuture等待任务完成？求助重复调用问题

看起来你在并行处理Article的Categories获取与保存时，不小心触发了重复的保存操作。我来帮你梳理下可能的原因和修复方案：

最可能的错误原因 #

重复调用的核心问题通常是对同一个CompletableFuture绑定了多个执行保存逻辑的回调，或者在任务收集时重复添加了相同的处理任务。举个常见的错误写法例子：

for (Article article : articles) {
    CompletableFuture<Categories> catFuture = CompletableFuture.supplyAsync(() -> api.getCategories(article.getId()));
    // 错误：同一个catFuture绑定了两个thenAccept，都会在Future完成时执行
    futures.add(catFuture.thenAccept(cats -> articleService.setCatgoriesForArticle(article, cats)));
    futures.add(catFuture.thenAccept(cats -> articleService.setCatgoriesForArticle(article, cats)));
}

或者在链式调用中重复执行保存逻辑：

catFuture.thenApply(cats -> {
    articleService.setCatgoriesForArticle(article, cats); // 第一次调用
    return cats;
}).thenAccept(cats -> articleService.setCatgoriesForArticle(article, cats)); // 第二次调用

正确的实现方案 #

我们需要确保每个Article对应一条完整的异步处理链（获取Categories → 保存到DB），并且只将这条链的最终Future加入等待集合。同时要保证所有任务并行执行，最后等待全部完成：

// 建议自定义线程池，避免使用默认ForkJoinPool影响其他任务
ExecutorService executor = Executors.newFixedThreadPool(10);

// 为每个Article创建独立的异步处理任务
List<CompletableFuture<Void>> allFutures = articles.stream()
    .map(article -> {
        // 步骤1：异步调用API获取Categories
        CompletableFuture<Categories> fetchCatFuture = CompletableFuture.supplyAsync(
            () -> apiClient.getCategoriesForArticle(article.getId()),
            executor
        );
        
        // 步骤2：获取到结果后，异步保存到数据库（这里的异步可选，也可以同步执行）
        return fetchCatFuture.thenAcceptAsync(
            categories -> articleService.setCatgoriesForArticle(article, categories),
            executor
        )
        // 可选：添加异常处理，避免单个任务失败导致整个流程阻塞
        .exceptionally(throwable -> {
            log.error("处理Article[{}]的Categories时失败", article.getId(), throwable);
            return null;
        });
    })
    .collect(Collectors.toList());

// 等待所有异步任务完成，确保后续日志后无残留异步任务
CompletableFuture.allOf(allFutures.toArray(new CompletableFuture[0])).join();

log.info("所有Article的Categories已完成保存，无残留异步任务");

// 记得关闭自定义线程池（如果是全局复用的池可以忽略）
executor.shutdown();

额外注意事项 #

1. 线程池选择：不要依赖默认的ForkJoinPool.commonPool()，尤其是在Spring等容器环境中，自定义线程池能更好地控制并发数，避免资源耗尽。
2. 异常处理：一定要添加异常捕获逻辑，否则单个API调用或DB保存失败会导致allOf()一直阻塞，或者抛出未捕获的异常。
3. 避免重复绑定回调：确保每个获取Categories的Future只绑定一次保存逻辑的回调，不要在多个地方重复添加执行保存的操作。

内容的提问来源于stack exchange，提问作者Evgeny M