深入理解Webflux repeat与then操作符的交互行为-java教程-PHP中文网

深入理解Webflux repeat与then操作符的交互行为

碧海醫心

发布： 2025-10-30 11:36:52

原创

473人浏览过

深入理解Webflux repeat与then操作符的交互行为

本文深入探讨了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操作符同时出现在一个流中时，它们的相对位置将极大地影响流的执行逻辑。

行者AI

行者AI绘图创作，唤醒新的灵感，创造更多可能

100

查看详情

场景一：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<Integer>
                .doOnNext(i -> System.out.println("next 1: " + i)) // 类型：Mono<Integer>
                .repeat(2) // 类型：Flux<Integer> (重复2次，总共执行3次)
                .doOnNext(i -> System.out.println("next 2: " + i)) // 类型：Flux<Integer>
                .then(Mono.just("hello")) // 类型：Mono<String> (在前面的Flux完成所有重复后执行)
                .doOnNext(i -> System.out.println("next 3: " + i)) // 类型：Mono<String>
                .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的相对位置，以及它们对流中数据和完成信号传播的影响，以避免出现预期之外的行为。