R2DBC中save与doOnNext执行顺序不符的问题咨询

先看你贴的代码：

Flux.range(1, 10)
        .flatMap(id -> repository.save(new Student(id, "lxp")))
        .doOnNext(System.out::println)
        .subscribe();

问题出在flatMap的并发特性上 #

Reactor的flatMap默认会一次性订阅多个上游元素对应的异步任务——这里每个repository.save()返回的是Mono<Student>，flatMap默认并发上限为256，所以你的10个save请求会被同时发出去，不是逐个等待前一个完成再执行下一个。

至于doOnNext要等所有save都完成才执行，大概率和R2DBC的事务或数据库连接逻辑有关：

• 如果你的save操作是在同一个事务上下文里执行的，R2DBC会把所有数据库操作攒到事务提交时才真正完成并返回结果，这就导致所有save对应的Mono都得等事务提交后才会触发onNext，自然doOnNext就会一次性执行。
• 另外，数据库连接池如果允许同时处理多个请求，这些save会并行执行，但数据库本身可能对批量操作做了优化，或者因为事务原子性要求，必须等所有操作都完成才返回结果。

要逐个执行？换concatMap就行 #

如果你想实现“save完成一个就立刻执行doOnNext”，把flatMap换成concatMap就好：

Flux.range(1, 10)
        .concatMap(id -> repository.save(new Student(id, "lxp")))
        .doOnNext(System.out::println)
        .subscribe();

concatMap是串行处理逻辑，必须等前一个save的Mono完全完成，才会去处理下一个元素，这样就能保证每个save做完就立刻触发doOnNext。

再提一句R2DBC的异步逻辑 #

R2DBC是异步非阻塞的数据库驱动，但异步不代表串行。flatMap的并发设计是为了提升吞吐量，适合批量并行处理；而concatMap则适合必须严格按顺序执行的场景。

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