Webfluxrepeat与then用法解析

本文深入解析了Webflux中`repeat`和`then`操作符的交互机制，旨在帮助开发者更好地理解和运用响应式编程。`repeat`操作符会触发上游Publisher的多次订阅，而`then`操作符则需等待上游Publisher发出完成信号后才会执行。文章通过实例详细阐述了这两个操作符在不同位置对数据流的影响，以及流类型的转换（Mono到Flux）。重点强调了`then`操作符在`repeat`之前与之后的不同行为：前者会被重复执行，后者仅在整个`repeat`序列完成后执行一次。掌握`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操作符同时出现在一个流中时，它们的相对位置将极大地影响流的执行逻辑。

场景一：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

在这个例子中：

1. Mono.just(5)执行doOnNext("next 1: " + i)。
2. repeat(2)将这个Mono转换为一个Flux，它会重复执行其上游三次。
3. doOnNext("next 2: " + i)作为repeat的下游，也会在每次重复中执行。
4. 因此，next 1: 5和next 2: 5会重复打印三次。
5. 关键在于then(Mono.just("hello"))。它等待的是其上游的Flux（由repeat(2)产生）完全完成，即所有三次重复都结束后，才会订阅并执行Mono.just("hello")。
6. 一旦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的相对位置，以及它们对流中数据和完成信号传播的影响，以避免出现预期之外的行为。

到这里，我们也就讲完了《Webfluxrepeat与then用法解析》的内容了。个人认为，基础知识的学习和巩固，是为了更好的将其运用到项目中，欢迎关注golang学习网公众号，带你了解更多关于的知识点！