如何在Spring WebFlux服务中非阻塞处理数据并返回Mono<BigDecimal>

问题背景 #

使用collectList().block()获取Flux中的列表会触发阻塞异常，Reactor非阻塞线程禁止调用阻塞方法，需用非阻塞操作符实现汇率相除逻辑并返回Mono<BigDecimal>。

方案一：基于批量查询的非阻塞处理 #

利用collectList()将Flux元素收集为Mono包裹的列表，通过flatMap处理计算逻辑，同时添加校验避免数据异常：

Mono<BigDecimal> convertedRateMono = currencyRepository.findByCurrencyNameIn(List.of(from, to))
        .map(CurrencyService::fromModel)
        .collectList()
        .flatMap(currencyRates -> {
            // 校验是否获取到两条有效记录
            if (currencyRates.size() != 2) {
                return Mono.error(new IllegalArgumentException("未获取到足够的汇率记录"));
            }
            // 匹配对应货币的汇率记录（避免数据库返回顺序不一致）
            CurrencyRate fromRate = currencyRates.stream()
                    .filter(rate -> rate.getCurrencyName().equals(from))
                    .findFirst()
                    .orElseThrow(() -> new IllegalArgumentException("未找到[" + from + "]对应的汇率"));
            CurrencyRate toRate = currencyRates.stream()
                    .filter(rate -> rate.getCurrencyName().equals(to))
                    .findFirst()
                    .orElseThrow(() -> new IllegalArgumentException("未找到[" + to + "]对应的汇率"));
            // 计算汇率比值，保留3位小数并使用HALF_EVEN舍入模式
            BigDecimal convertedRate = fromRate.getRate()
                    .divide(toRate.getRate(), 3, RoundingMode.HALF_EVEN);
            return Mono.just(convertedRate);
        });

方案二：基于独立查询的合并处理 #

如果允许拆分查询，用Mono.zip()合并两个独立的货币查询Mono，逻辑更简洁：

Mono<BigDecimal> convertedRateMono = Mono.zip(
        // 查询源货币汇率
        currencyRepository.findByCurrencyName(from).map(CurrencyService::fromModel),
        // 查询目标货币汇率
        currencyRepository.findByCurrencyName(to).map(CurrencyService::fromModel)
)
.map(tuple -> {
    CurrencyRate fromRate = tuple.getT1();
    CurrencyRate toRate = tuple.getT2();
    return fromRate.getRate()
            .divide(toRate.getRate(), 3, RoundingMode.HALF_EVEN);
});

方案对比 #

• 方案一：仅触发一次数据库批量查询，适合需要减少DB请求的场景，但需处理列表元素的匹配与校验
• 方案二：逻辑更直观，无需处理顺序问题，但会触发两次数据库查询

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