Spring Reactive：更新嵌套列表对象后保存父对象

我明白你现在的困境——在Reactor中处理嵌套对象的异步更新时，很容易陷入嵌套subscribe()的陷阱，导致代码混乱且不符合响应式规范。下面是针对你需求的正确实现方案，同时我会解释为什么之前的写法有问题。

正确的实现代码 #

首先直接给出可以正常工作的响应式代码：

public Mono<Video> scanSingleVideo(@PathVariable final String id) {
    return videoRepository.findById(id)
            // 拿到Video后，处理所有Thumbnail的Face检测
            .flatMap(video -> {
                // 创建一个流，处理每个Thumbnail的Face更新，完成后返回Void
                Flux<Void> thumbnailUpdateStream = Flux.fromIterable(video.getThumbnails())
                        .flatMap(thumbnail -> 
                            // 调用业务方法获取该缩略图的所有Face
                            faceService.detectFaces(thumbnail.getPath())
                                    // 一次性收集所有Face，批量添加到Thumbnail的集合中
                                    .collectList()
                                    .doOnNext(faces -> thumbnail.getFaces().addAll(faces))
                                    // 标记当前Thumbnail的处理完成
                                    .then()
                        );
                
                // 等所有Thumbnail的Face都处理完成后，再保存更新后的Video
                return thumbnailUpdateStream.then(videoRepository.save(video));
            });
}

如果你更倾向于逐个处理Face（虽然批量收集更高效），也可以这样写：

public Mono<Video> scanSingleVideo(@PathVariable final String id) {
    return videoRepository.findById(id)
            .flatMap(video -> {
                Flux<Void> thumbnailUpdateStream = Flux.fromIterable(video.getThumbnails())
                        .flatMap(thumbnail -> 
                            faceService.detectFaces(thumbnail.getPath())
                                    // 逐个将Face添加到Thumbnail的集合中
                                    .doOnNext(thumbnail.getFaces()::add)
                                    .then()
                        );
                
                return thumbnailUpdateStream.then(videoRepository.save(video));
            });
}

代码解释 #

1. 获取目标Video：通过videoRepository.findById(id)得到Mono<Video>，这是整个序列的起点。
2. 编排Thumbnail的异步处理：
   • 用Flux.fromIterable(video.getThumbnails())将Thumbnail列表转换为响应式流。
   • 对每个Thumbnail，调用faceService.detectFaces()获取对应的Face流，然后通过collectList()批量收集所有Face，再用doOnNext()将这些Face添加到Thumbnail的faces集合中（因为Thumbnail是引用类型，原Video中的对象会被直接修改）。
   • 用then()将每个Thumbnail的处理流转换为Mono<Void>，表示这个Thumbnail的更新操作已完成。
3. 统一保存Video：thumbnailUpdateStream.then(videoRepository.save(video))会等待所有Thumbnail的更新操作都完成后，才执行保存操作，确保保存的是已经更新了所有Face的Video对象。

之前写法的问题 #

你之前的嵌套subscribe()写法存在几个关键问题：

• 脱离响应式上下文：嵌套的subscribe()会启动独立的异步流，和外层的videoRepository.save(video)没有编排关系，导致保存操作可能在Face检测完成前就执行，最终保存的Video没有更新后的Face。
• 不必要的重复保存：每个Face都触发一次videoRepository.save(video)，会产生大量冗余的数据库操作，还可能引发并发更新的问题。
• 违反响应式原则：这种写法没有利用Reactor的异步编排能力，手动管理订阅会导致代码难以维护，还可能出现线程安全问题。

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