本文介绍在 Spring WebFlux 中处理 FilePart 时，如何在完成流式数据处理后一次性、安全地删除本地临时文件，同时不中断或消耗原始 Flux 输出流。核心在于利用 doOnComplete 延迟执行清理逻辑，并确保 I/O 操作在合适线程池中运行。
本文介绍在 spring webflux 中处理 `filepart` 时，如何在完成流式数据处理后**一次性、安全地删除本地临时文件**，同时不中断或消耗原始 `flux` 输出流。核心在于利用 `dooncomplete` 延迟执行清理逻辑，并确保 i/o 操作在合适线程池中运行。
在响应式编程中，尤其是使用 Project Reactor 处理文件上传（如 FilePart）时，一个常见模式是：先将文件落地为临时磁盘文件（便于流式解析或大文件处理），再基于该文件构建 Flux 进行逐块/逐行处理，最后必须及时清理临时文件以避免磁盘泄漏。但需注意：不能在 flatMap 或 map 中直接触发删除操作——这会导致每次元素发射都尝试删一次文件（竞态+失败），也无法保证删除发生在整个流结束之后。
正确做法是将清理逻辑绑定到流的生命周期终点：使用 doOnComplete —— 它仅在 Flux 正常终止（无错误）时触发一次，且不改变上游数据流。由于 deleteFile() 返回的是 Mono<void></void>（异步 I/O），我们需显式订阅它（subscribe()），否则其内部逻辑不会执行；同时应指定 Schedulers.boundedElastic() 以避免阻塞事件循环线程。
以下是重构后的推荐实现：
Flux<ClassB> functionA(Mono<FilePart> filePartMono) {
return filePartMono.flatMapMany(filePart ->
fileHandlerService.saveFile(Mono.just(filePart)) // 注意：此处传入 Mono.just(filePart)，避免重复消费 filePartMono
.then(Mono.just(filePart))
.flatMapMany(this::getItems)
.doOnComplete(() ->
Mono.just(filePart)
.flatMap(fileHandlerService::deleteFile)
.subscribeOn(Schedulers.boundedElastic())
.subscribe(
null, // onComplete 不需要处理
error -> log.error("Failed to delete temp file: {}", filePart.filename(), error)
)
)
);
}⚠️ 关键注意事项：
-
避免多次消费
Mono<filepart></filepart>：原代码中filePartMono被多次 flatMap，而Mono是冷序列，不可重复订阅。应改为Mono.just(filePart)或使用cache()缓存结果。 -
doOnComplete不捕获错误：若getItems()抛出异常或上游发生错误，doOnComplete不会执行。如需无论成功或失败都清理，请改用doFinally(signal -> { if (signal == SignalType.ON_COMPLETE || signal == SignalType.ON_ERROR) ... })。 -
日志与错误处理：删除失败不应影响主流程（因
Flux已完成），但务必记录错误以便运维排查。 -
路径安全性：
filePart.filename()可能含恶意路径（如../etc/passwd），生产环境务必对文件名做白名单校验或使用Path.resolve()防止路径遍历。
总结：响应式文件处理的清理工作，本质是资源生命周期管理问题。doOnComplete + 显式 subscribe() 是 WebFlux 场景下兼顾语义清晰性与执行可靠性的标准解法。
