Webflux 累加与详情：将同步代码转换为响应式实现

你需要把同步的库存详情+总计查询代码改成WebFlux响应式实现，目前已经完成了库存到响应对象的映射，但不知道如何在保留详情数据的同时计算总计。

原同步代码 #

@Getter
public class Stock {
    private int value;
    private String productNumber;
    public Stock(int value, String productNumber){
        this.value = value;
        this.productNumber = productNumber;
    }
}


@Getter
public class StockResponse {
    private int value;
    private String productNumber;
    public StockResponse(int value, String productNumber){
        this.value = value;
        this.productNumber = productNumber;
    }
    public StockResponse(Stock stock){
        this.value = stock.getValue();
        this.productNumber = stock.getProductNumber();
    }

    public void add(int value){
        this.value += value;
    }
    
}

public List<StockResponse> findStockResponsesDetailsAndTotal() {
    List<StockResponse> stocksResponses = new ArrayList();
    List<Stock> stocks = stockRepository.findAll();
    StockResponse total = new StockResponse(0, "Total");
    stocks.forEach(stock -> {
        total.add(stock.getValue());
        stocksResponses.add(new StockResponse(stock));
    });
    stocksResponses.add(total);
    return stocksResponses;
}

你已完成的代码 #

stockRepository.findAll()
  .map(StockResponse::convert)

解决方案 #

核心问题是reduce会把Flux转换成单个结果的Mono，导致丢失详情数据。我们需要同时保留详情流和计算总计，再把两者合并。下面提供两种实现方式：

方式1：适合数据量较小的场景（一次性收集到内存）

先把所有详情收集到Mono<List<StockResponse>>，基于这个列表计算总计，最后把详情列表转成Flux并拼接总计结果：

public Flux<StockResponse> findStockResponsesDetailsAndTotal() {
    // 收集所有库存详情到List，仅订阅一次数据库查询
    Mono<List<StockResponse>> detailListMono = stockRepository.findAll()
            .map(StockResponse::new) // 对应你写的StockResponse::convert
            .collectList();

    // 基于详情列表计算总计
    Mono<StockResponse> totalMono = detailListMono
            .map(details -> {
                int totalValue = details.stream()
                        .mapToInt(StockResponse::getValue)
                        .sum();
                return new StockResponse(totalValue, "Total");
            });

    // 先返回所有详情，再返回总计
    return detailListMono.flatMapMany(Flux::fromIterable)
            .concatWith(totalMono);
}

方式2：适合数据量较大的场景（流式处理，不加载全量数据到内存）

通过publish().autoConnect(2)让详情流被两个订阅者（详情输出和总计计算）共享，避免重复查询数据库：

public Flux<StockResponse> findStockResponsesDetailsAndTotal() {
    // 共享流，确保数据库查询仅执行一次
    Flux<StockResponse> detailResponses = stockRepository.findAll()
            .map(StockResponse::new)
            .publish()
            .autoConnect(2);

    // 基于流式数据计算总计
    Mono<StockResponse> totalResponse = detailResponses
            .reduce(new StockResponse(0, "Total"), (total, current) -> {
                total.add(current.getValue());
                return total;
            });

    // 拼接详情流和总计结果
    return Flux.concat(detailResponses, totalResponse);
}

优化建议：使用不可变对象 #

原StockResponse的add方法是可变操作，在多线程场景下可能存在线程安全问题。建议改成不可变实现，更符合响应式函数式编程理念：

@Getter
public class StockResponse {
    private final int value;
    private final String productNumber;

    public StockResponse(int value, String productNumber){
        this.value = value;
        this.productNumber = productNumber;
    }

    public StockResponse(Stock stock){
        this.value = stock.getValue();
        this.productNumber = stock.getProductNumber();
    }

    // 返回新对象，而非修改当前对象
    public StockResponse add(int value) {
        return new StockResponse(this.value + value, this.productNumber);
    }
}

此时总计计算的代码可以简化为纯函数式操作：

Mono<StockResponse> totalResponse = detailResponses
        .reduce(new StockResponse(0, "Total"), (total, current) -> 
                total.add(current.getValue())
        );

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