
本文深入探讨了webflux中`repeat`和`then`操作符的复杂交互行为。`repeat`操作符会使其上游的publisher进行多次订阅,而`then`操作符则在接收到上游publisher完成信号后才执行。理解这两个操作符的相对位置及其对流类型(mono/flux)的影响,对于正确构建响应式数据流至关重要,尤其是在涉及重复执行和序列化操作的场景中。
Webflux repeat 操作符的基础行为
repeat操作符用于使其上游的响应式流重新订阅指定次数。这意味着在repeat操作符之前的整个流会从头开始执行多次。
考虑以下示例:
import reactor.core.publisher.Mono;
public class RepeatExample {
public static void main(String[] args) {
Mono.just(5)
.doOnNext(i -> System.out.println("next 1: " + i))
.doOnNext(i -> System.out.println("next 2: " + i))
.doOnNext(i -> System.out.println("next 3: " + i))
.repeat(2) // 重复2次,总共执行3次
.subscribe();
}
}输出结果清晰地表明,doOnNext操作符打印了三次,每次都执行了所有三个语句:
next 1: 5 next 2: 5 next 3: 5 next 1: 5 next 2: 5 next 3: 5 next 1: 5 next 2: 5 next 3: 5
即使我们将repeat操作符的位置上移,只要它仍然在所有doOnNext操作符之前,其行为保持不变,因为它会触发整个上游流的重新订阅。
import reactor.core.publisher.Mono;
public class RepeatExample2 {
public static void main(String[] args) {
Mono.just(5)
.doOnNext(i -> System.out.println("next 1: " + i))
.repeat(2) // 重复2次,总共执行3次
.doOnNext(i -> System.out.println("next 2: " + i))
.doOnNext(i -> System.out.println("next 3: " + i))
.subscribe();
}
}输出与第一个示例相同:
next 1: 5 next 2: 5 next 3: 5 next 1: 5 next 2: 5 next 3: 5 next 1: 5 next 2: 5 next 3: 5
这表明repeat操作符会使其上游的整个Publisher重新订阅,而不影响其下游操作符的执行顺序。
then 操作符的作用
then操作符用于在当前Publisher完成(即发出onComplete信号)后,订阅并执行另一个Publisher。它会忽略上游Publisher发出的所有元素,只关注其完成状态。
repeat 与 then 的交互行为
当repeat和then操作符同时出现在一个流中时,它们的相对位置将极大地影响流的执行逻辑。
场景一:then 在 repeat 之前
如果then操作符位于repeat操作符之前,那么then操作符所引入的后续流也会被repeat操作符重复执行。
import reactor.core.publisher.Mono;
public class RepeatThenExample1 {
public static void main(String[] args) {
Mono.just(5)
.doOnNext(i -> System.out.println("next 1: " + i))
.doOnNext(i -> System.out.println("next 2: " + i))
.then(Mono.just("hello")) // Mono.just(5)完成后,执行Mono.just("hello")
.doOnNext(i -> System.out.println("next 3: " + i)) // 此时流的元素是"hello"
.repeat(2) // 重复整个上游流
.subscribe();
}
}输出结果:
next 1: 5 next 2: 5 next 3: hello next 1: 5 next 2: 5 next 3: hello next 1: 5 next 2: 5 next 3: hello
在这个例子中,Mono.just(5)完成,然后Mono.just("hello")被订阅并发出"hello"元素,接着doOnNext("next 3: " + i)打印"next 3: hello"。这个完整的序列(next 1: 5, next 2: 5, next 3: hello)被repeat(2)重复了两次,总共执行了三次。
场景二:then 在 repeat 之后
这是最容易引起混淆的场景。当repeat操作符将一个Mono转换为一个会多次完成的Flux之后,如果then操作符位于repeat之后,那么then操作符将只会在整个repeat序列完全完成之后才执行一次。
import reactor.core.publisher.Mono;
import reactor.core.publisher.Flux; // 引入Flux以便于理解类型转换
public class RepeatThenExample2 {
public static void main(String[] args) {
Mono.just(5) // 类型:Mono
.doOnNext(i -> System.out.println("next 1: " + i)) // 类型:Mono
.repeat(2) // 类型:Flux (重复2次,总共执行3次)
.doOnNext(i -> System.out.println("next 2: " + i)) // 类型:Flux
.then(Mono.just("hello")) // 类型:Mono (在前面的Flux完成所有重复后执行)
.doOnNext(i -> System.out.println("next 3: " + i)) // 类型:Mono
.subscribe();
}
} 输出结果:
next 1: 5 next 2: 5 next 1: 5 next 2: 5 next 1: 5 next 2: 5 next 3: hello
在这个例子中:
- Mono.just(5)执行doOnNext("next 1: " + i)。
- repeat(2)将这个Mono转换为一个Flux,它会重复执行其上游三次。
- doOnNext("next 2: " + i)作为repeat的下游,也会在每次重复中执行。
- 因此,next 1: 5和next 2: 5会重复打印三次。
- 关键在于then(Mono.just("hello"))。它等待的是其上游的Flux(由repeat(2)产生)完全完成,即所有三次重复都结束后,才会订阅并执行Mono.just("hello")。
- 一旦Mono.just("hello")执行,它会发出"hello"元素,然后doOnNext("next 3: " + i)才会打印next 3: hello。
这解释了为什么next 3: hello只出现了一次,因为它是在整个重复序列结束后才被触发的。
总结与注意事项
- repeat操作符:使其上游的整个Publisher进行多次订阅。它会将一个Mono转换为一个Flux,或者延长一个Flux的生命周期。
- then操作符:在收到上游Publisher的onComplete信号后,才订阅并执行其参数Publisher。它会忽略上游Publisher发出的所有元素。
-
操作符位置的重要性:
- 当then在repeat之前时,then所引入的后续流会成为repeat操作符的上游,因此也会被重复执行。
- 当then在repeat之后时,then操作符会等待repeat操作符所产生的整个Flux(包含所有重复)完全完成,才会执行一次。
- 流类型转换:repeat操作符通常会将一个Mono转换为一个Flux,因为它会产生多个完成事件(尽管每个事件只包含一个元素)。理解这种类型转换有助于预测后续操作符的行为。
- 在设计复杂的响应式流时,务必仔细考虑repeat和then的相对位置,以及它们对流中数据和完成信号传播的影响,以避免出现预期之外的行为。










