如何在统计多项指标后将Flux转换为Mono？

你遇到的核心问题是Reactor冷流的执行时机：你定义的successCount、errorCount等外部变量，只有在Flux被订阅后才会被更新，但你之前的代码错误地在流未执行时就尝试读取这些变量的值。另外，使用外部可变对象（如AtomicInteger、HashMap）在Reactor异步流程中存在线程安全风险（如果后续改为并行处理），也不符合响应式编程的最佳实践。

不要依赖外部可变变量，而是通过Reactor的reduce操作在流处理过程中累积统计数据，最后直接使用累积的结果调用后续方法。

步骤1：定义统计结果类 #

首先创建一个类来封装所有统计信息，避免使用外部变量：

private static class ProcessingStats {
    private int successCount = 0;
    private int errorCount = 0;
    private final Map<String, List<String>> filteredRecords = new HashMap<>();

    public int getSuccessCount() { return successCount; }
    public int getErrorCount() { return errorCount; }
    public Map<String, List<String>> getFilteredRecords() { return filteredRecords; }

    public void incrementSuccess() { successCount++; }
    public void incrementError() { errorCount++; }
}

步骤2：重构Flux处理逻辑 #

使用reduce操作替代外部变量，在每个元素处理时更新统计信息：

var stats = new ProcessingStats();
try (var reader = new CSVReader(new InputStreamReader(new ZipInputStream(is)))) {
    return Flux.fromIterable(reader)
            .map(this::toRecord)
            .reduce(stats, (currentStats, record) -> {
                // 处理过滤逻辑，将被过滤的记录存入统计对象
                if (!filterRecords(record, currentStats.getFilteredRecords())) {
                    return currentStats;
                }
                // 处理记录，捕获异常并更新错误统计
                try {
                    this.processRecord(record);
                    currentStats.incrementSuccess();
                } catch (Exception ex) {
                    log.error(ERROR_RECORD, StructuredArguments.v("record", record), ex);
                    currentStats.incrementError();
                }
                return currentStats;
            })
            // 流处理完成后，使用统计结果调用checkErrorsAndLog
            .flatMap(finalStats -> checkErrorsAndLog(
                    finalStats.getSuccessCount(),
                    finalStats.getErrorCount(),
                    finalStats.getFilteredRecords(),
                    fileName
            ));
} catch (CsvValidationException | IOException exception) {
    log.error(ERROR_LOG_MSG, StructuredArguments.value(FILE_NAME, fileName),
            StructuredArguments.value(RECORD_COUNT, 0), exception);
    return Mono.error(new ProcessingException(String.format(ERROR_PROCESSING_FAILED, fileName), exception));
}

为什么这个方案有效？ #

1. 冷流执行时机对齐：reduce操作会等待Flux所有元素处理完成后，才会将最终的统计对象传递给flatMap，确保统计值已经完全更新。
2. 线程安全：所有统计操作都在流的处理流程中完成，避免了外部可变对象的线程安全问题（即使后续改为并行处理，只需调整reduce为线程安全的累积方式即可）。
3. 符合响应式编程风格：保持链式调用，避免手动subscribe()带来的异步流程割裂问题。

你之前的代码中，虽然用了collectList().flatMap和Mono.defer，但核心问题是：

• 外部变量的更新依赖于Flux的订阅执行，而如果返回的Mono未被订阅，流根本不会执行，统计值自然为0。
• onErrorContinue的异常处理逻辑与doOnNext的成功统计是分离的，虽然逻辑上正确，但依赖外部变量的方式容易导致时机误解。

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