如何在Flux中累加多个数值实现响应式收支汇总报表服务

你原有实现的核心问题有三个：

• 不应使用AtomicReference这类带副作用的容器在响应式流中做累加，一旦流出现重试、重订阅场景会出现数据计算错误，且doOnEach属于副作用钩子，不适合承载核心计算逻辑
• 累加逻辑写错了：原有代码直接getAndSet覆盖了累计值，没有做累加操作
• 没有触发响应式流的终端操作，仓库返回的Flux根本不会执行，直接返回空流也不符合返回值要求

正确的实现可以选择两种方式： #

方式1：先收集全量明细再计算（逻辑最直观，易维护） #

@Override
public Mono<IncomeExpenseSummaryReport> getIncomeExpenseReportByPeriod(GetIncomeExpenseReportRequest request) {
    return incomeExpenseRepository.findAllByDateContaining(request.getPeriod())
            // 先收集所有收支明细为列表
            .collectList()
            .map(incomeExpenseList -> {
                BigDecimal totalIncome = BigDecimal.ZERO;
                BigDecimal totalPaid = BigDecimal.ZERO;
                for (IncomeExpense item : incomeExpenseList) {
                    totalIncome = totalIncome.add(BigDecimal.valueOf(item.getAmount()));
                    if ("paid".equals(item.getPaymentStatus())) {
                        totalPaid = totalPaid.add(BigDecimal.valueOf(item.getAmount()));
                    }
                }
                // 组装汇总结果返回
                return new IncomeExpenseSummaryReport(incomeExpenseList, totalIncome, totalPaid);
            });
}

方式2：流式单次累加（适合数据量较大的场景，不需要全量缓存明细到内存再计算） #

我们可以自定义一个临时累加容器，用reduce操作符完成流式计算：

// 先定义一个仅内部使用的临时累加容器，不需要对外暴露
private static class SummaryAccumulator {
    List<IncomeExpense> details = new ArrayList<>();
    BigDecimal totalIncome = BigDecimal.ZERO;
    BigDecimal totalPaid = BigDecimal.ZERO;
}

@Override
public Mono<IncomeExpenseSummaryReport> getIncomeExpenseReportByPeriod(GetIncomeExpenseReportRequest request) {
    return incomeExpenseRepository.findAllByDateContaining(request.getPeriod())
            .reduce(
                    // 初始化累加器
                    new SummaryAccumulator(),
                    (accumulator, current) -> {
                        // 累计明细
                        accumulator.details.add(current);
                        // 累计总收入
                        accumulator.totalIncome = accumulator.totalIncome.add(BigDecimal.valueOf(current.getAmount()));
                        // 累计已支付金额
                        if ("paid".equals(current.getPaymentStatus())) {
                            accumulator.totalPaid = accumulator.totalPaid.add(BigDecimal.valueOf(current.getAmount()));
                        }
                        return accumulator;
                    }
            )
            // 把累加容器转换为最终返回的汇总对象
            .map(acc -> new IncomeExpenseSummaryReport(acc.details, acc.totalIncome, acc.totalPaid));
}

两种方式都完全符合响应式编程规范，不需要手动维护线程安全变量，也能正确触发数据库查询操作。

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