本文详解如何在 project reactor 中安全、响应式地收集 flux 执行过程中的多个计数状态（如成功数、错误数、过滤记录），避免因过早读取原子变量或资源提前关闭导致的竞态问题。
本文详解如何在 project reactor 中安全、响应式地收集 flux 执行过程中的多个计数状态（如成功数、错误数、过滤记录），避免因过早读取原子变量或资源提前关闭导致的竞态问题。
在使用 Project Reactor 处理流式数据（如 CSV 解析与批处理）时，一个常见需求是：在 Flux 完全终止后，获取执行过程中累积的多个指标（如成功处理数、错误数、被过滤的记录集合）。但直接使用 AtomicInteger + collectList().flatMap(...) 往往失败——原因在于：Flux 是惰性（lazy）且异步的，其内部操作不会立即执行；而你试图在订阅前就访问原子变量的值，此时它们仍为初始值（如 0）。
更关键的是，原始代码中存在同步资源管理与响应式流的严重混用：
- try-with-resources 提前关闭了 CSVReader，但 Flux 尚未开始或尚未完成消费；
- doOnNext 和 onErrorContinue 的副作用发生在异步线程上，而 successCount.get() 在主线程调用时无法保证可见性；
- collectList() 收集的是元素本身，但你真正关心的是副作用产生的“元状态”，而非数据内容。
✅ 正确解法：将所有状态收集逻辑内聚到响应式链中，并确保资源生命周期由 Reactor 管理。
✅ 推荐方案：使用 Flux.using() + handle() + reduce()
return Flux.using(
// 1. 资源创建（延迟到订阅时）
() -> {
try {
return new CSVReader(new InputStreamReader(new ZipInputStream(is)));
} catch (IOException e) {
throw Exceptions.propagate(e);
}
},
// 2. 数据流生成（注意：fromIterable 是同步阻塞的，需确保 reader 可重复/及时关闭）
reader -> Flux.fromIterable(reader)
.map(this::toRecord)
.handle((record, sink) -> {
if (filterRecords(record, filteredRecords)) {
// 过滤逻辑已在此完成，无需额外 map/filter
RecordResult result = processRecord(record);
sink.next(result); // 继续下游
}
// 被过滤的 record 不发出，但 filteredRecords 已更新
})
.doOnNext(result -> successCount.incrementAndGet())
.onErrorContinue((ex, record) -> {
log.error(ERROR_RECORD, StructuredArguments.v("record", record), ex);
errorCount.incrementAndGet();
}),
// 3. 资源清理（保证在 Flux 完成/出错/取消时执行）
reader -> {
try {
reader.close();
} catch (IOException e) {
log.warn("Failed to close CSV reader", e);
}
}
)
// 4. 收集最终状态：忽略元素本身，只聚合计数器和上下文
.reduce(
Tuples.of(0, 0, new HashMap<string list>>()), // 初始状态
(acc, ignored) -> Tuples.of(
successCount.get(),
errorCount.get(),
new HashMap(filteredRecords) // 浅拷贝避免后续修改影响
)
)
.flatMap(tuple -> checkErrorsAndLog(tuple.getT1(), tuple.getT2(), tuple.getT3(), fileName));</string>
⚠️ 注意事项与最佳实践
- 绝不手动 subscribe()：subscribe() 是终端操作，应仅在应用边界（如 Web 控制器返回、测试断言）调用。在中间链路调用会破坏响应式链，导致不可预测的线程模型和资源泄漏。
- 避免共享可变状态（如 AtomicInteger）跨异步边界：虽然 AtomicInteger 是线程安全的，但在 reduce() 或 collectList() 后再读取它，仍可能因内存可见性或执行时机问题拿到旧值。推荐改用 reduce() 或 scan() 进行纯函数式状态累积。
- 优先使用 Flux.using() 管理资源：它能确保资源在 onComplete、onError 或 cancel 时自动释放，彻底规避 try-with-resources 与响应式流的生命周期冲突。
- 若必须保留 AtomicInteger（如性能敏感场景）：请用 Mono.usingWhen() 包裹整个流程，并在 resourceCleanup 阶段读取最终值：
return Mono.usingWhen(
Mono.fromCallable(() -> new AtomicInteger[] {new AtomicInteger(), new AtomicInteger()}),
atoms -> Flux.using(/* ... same as above ... */)
.then(Mono.fromCallable(() ->
checkErrorsAndLog(atoms[0].get(), atoms[1].get(), filteredRecords, fileName))),
atoms -> Mono.empty() // cleanup not needed for atoms, but required signature
);
✅ 总结
将 Flux 转换为携带多个计数器的 Mono，核心在于：让状态收集成为流的一部分，而非外部副作用的“事后读取”。通过 Flux.using() 管理资源、reduce() 或 handle() 聚合状态、并严格遵循响应式编程的惰性与异步契约，即可写出既正确又可维护的 Reactive 代码。记住：在 Reactor 中，“何时执行”比“如何执行”更重要——一切逻辑都应在订阅后、由流驱动完成。
