如何将IntStream生成的CompletableFuture值累加至最终CompletableFuture变量

根因分析 #

• CompletableFuture的链式方法（thenCompose/thenApply等）返回的是全新的实例，原有代码没有将每次调用thenCompose的返回值赋值给结果变量，导致finalAmount始终指向初始的completedFuture(0L)，所以返回值永远是0
• 原有逻辑没有实现累加效果，即使赋值也只会用单次计算结果覆盖原有值，不会累计多次计算的金额总和

修复方案 #

方案1：串行累加（请求按顺序发起） #

需要用AtomicReference封装CompletableFuture引用，解决lambda无法修改外部局部变量的问题，每次计算完成后把新值累加到总和里：

import java.util.concurrent.atomic.AtomicReference;

AtomicReference<CompletableFuture<Long>> finalAmountRef = new AtomicReference<>(CompletableFuture.completedFuture(0L));

IntStream.range(rankUpEvent.getOldClubRank()+1, rankUpEvent.getNewClubRank()).forEach(
    rank -> {
        if (!clubs.containsKey(rank)) {
            finalAmountRef.updateAndGet(prevTotalFuture -> 
                prevTotalFuture.thenCompose(total -> 
                    configService.getConfig(id, rank - 1)
                        .thenApply(config -> total + getAmount(config, "value"))
                )
            );
        }
    }
);
return finalAmountRef.get();

方案2：并行计算（性能更高） #

如果允许所有配置请求并行发起，可以先收集所有符合条件的计算任务，全部完成后统一累加，整体耗时等于最慢的单个请求耗时：

import java.util.stream.Collectors;
import java.util.List;
import java.util.concurrent.CompletableFuture;

List<CompletableFuture<Long>> amountFutures = IntStream.range(rankUpEvent.getOldClubRank()+1, rankUpEvent.getNewClubRank())
    .filter(rank -> !clubs.containsKey(rank))
    .mapToObj(rank -> configService.getConfig(id, rank - 1)
        .thenApply(config -> getAmount(config, "value"))
    )
    .collect(Collectors.toList());

return CompletableFuture.allOf(amountFutures.toArray(new CompletableFuture[0]))
    .thenApply(ignored -> 
        amountFutures.stream()
            .mapToLong(CompletableFuture::join)
            .sum()
    );

选型建议 #

• 串行方案适合下游服务吞吐量低、要控制并发请求数的场景
• 并行方案适合请求量不大、下游服务可承受并发调用的场景，性能优势明显

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