Spring Boot WebFlux上传文件存ES时文件ID未关联实体问题

问题现象 #

基于Spring Boot WebFlux实现文件上传逻辑，预期链路为：接收前端上传文件→将文件上传至第三方存储服务→把第三方返回的文件ID关联存储到Elasticsearch中对应的Order实体。当前异常表现：

• 文件可成功上传至第三方存储服务
• 返回的文件ID无法绑定关联到对应Order实体，addImages方法内的doOnNext、flatMap代码块未执行
• 直接从Controller层调用addImages方法时，方法可正常运行

原问题代码 #

Controller层 #

@PostMapping(value = "upload", consumes = MediaType.MULTIPART_FORM_DATA_VALUE, produces = MediaType.APPLICATION_JSON_VALUE)
public Flux<String> store(@RequestParam(required = false) String orderId, @RequestPart("file") Flux<FilePart> files){
    return imageService.store(orderId, files);
}

ImageService层 #

public Flux<String> store(String orderId, Flux<FilePart> files) {
    return marketService.findById(orderId)
            .filter(Objects::nonNull)
            .flatMapMany(order -> {
                return files.ofType(FilePart.class).flatMap(file -> save(orderId, file));
            });
}

private Mono<String> save(String orderId, FilePart file) {
    return file.content()
            .flatMap(dataBuffer -> {
                byte[] bytes = new byte[dataBuffer.readableByteCount()];
                dataBuffer.read(bytes);
                String image = storeApi.upload(bytes, file.filename());
                DataBufferUtils.release(dataBuffer);
                return Mono.just(image);
            })
            .doOnNext(image -> marketService.addImages(orderId, image))
            .last();
}

MarketService层addImages方法 #

public Mono<Order> addImages(String id, String image){
    log.info("addImages: id={}, image={}", id, image);
    return orderRepository
            .findById(id)
            .doOnNext(order -> {
                if(order.getProduct().getImages() == null){
                    order.getProduct().setImages(new ArrayList<>());
                }
                order.getProduct().getImages().add(image);
            })
            .flatMap(this::create);
}

根因分析 #

核心错误是响应式流未被纳入主链路、未触发订阅，违反了Reactor响应式编程的基本规则：

• 响应式编程中，Mono/Flux是惰性执行的，只有被订阅时才会触发内部逻辑。你在doOnNext回调中调用marketService.addImages(orderId, image)时，仅获取到了方法返回的Mono<Order>对象，既没有将这个Mono串联到整个接口的响应式主链路中，也没有主动订阅，因此Mono内部的数据库查询、更新逻辑永远不会执行。
• 你能看到addImages方法入口的日志打印，是因为方法调用是同步执行的，日志语句在构建Mono对象时就会运行，和Mono后续是否被订阅无关。
• 直接从Controller调用addImages能正常执行，是因为Controller返回的Mono会被WebFlux核心框架自动订阅，触发整个流的执行。
• 原代码还存在两个隐藏bug：一是直接遍历file.content()的DataBuffer时，大文件会被拆分为多个DataBuffer分片，原逻辑仅会处理最后一个分片，导致上传到第三方的文件损坏；二是doOnNext是用于日志、埋点等轻量副作用的peek操作，不适合承载返回响应式流的数据库更新操作。

修复代码 #

修正save方法，将更新订单的逻辑串联到主响应式链路 #

private Mono<String> save(String orderId, FilePart file) {
    // 用join聚合文件所有DataBuffer分片，避免大文件读取损坏
    return DataBufferUtils.join(file.content())
            .flatMap(dataBuffer -> {
                byte[] bytes = new byte[dataBuffer.readableByteCount()];
                dataBuffer.read(bytes);
                // 手动释放DataBuffer避免内存泄漏
                DataBufferUtils.release(dataBuffer);
                String fileId = storeApi.upload(bytes, file.filename());
                // 将订单更新操作串联到流中，确保执行时被订阅
                return marketService.addImages(orderId, fileId)
                        .thenReturn(fileId);
            });
}

修正store方法逻辑 #

public Flux<String> store(String orderId, Flux<FilePart> files) {
    return marketService.findById(orderId)
            // 订单不存在时直接抛出异常，避免空流无响应
            .filter(Objects::nonNull)
            .switchIfEmpty(Mono.error(() -> new IllegalArgumentException("对应订单不存在")))
            .flatMapMany(order -> files.flatMap(file -> save(orderId, file)));
}

响应式开发注意：所有返回Mono/Flux的业务方法，都不能直接丢弃返回值，必须将其串联到整个响应流链路中，或明确触发订阅，否则内部逻辑不会执行。doOnNext、doOnError这类peek操作仅适合做无返回值的轻量副作用，不能用来承载需要执行响应式流的业务逻辑。

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