【反应式编程】三，使用常见的反应式操作

Flux和Mono是Reactor提供的最基础的构建块，而这两种反应式类型所提供的操作符则是组合使用它们以构建数据流动管线的黏合剂。
这些操作都可以大致归类为：

• 创建操作
• 组合操作
• 转换操作
• 逻辑操作

可以使用Flux或Mono上的静态 just()方法来创建一个反应式类型

@Test
public void createAFlux_just() {
  Flux fruitFlux = Flux
      .just("Apple", "Orange", "Grape", "Banana", "Strawberry");
}

此时我们已经创建了Flux，但是它还没有订阅者。如果没有任何的订阅者，那么数据将不会流动。
添加一个订阅者

fruitFlux.subscribe(
  f -> System.out.println("Here's some fruit: " + f)
);

可以根据数组、Iterable 或者 Java Stream创建Flux

String[] fruits = new String[]{"Apple", "Orange", "Banana", "Grape", "Strawberry"};
Flux flux = Flux.just(fruits);
flux.subscribe(f -> System.out.println(f));

List fruits = new ArrayList<>();
fruits.add("Apple");
fruits.add("Orange");
fruits.add("Banana");
fruits.add("Grape");
fruits.add("Strawberry");

Flux flux = Flux.fromIterable(fruits);
flux.subscribe(f -> System.out.println(f));

Stream fruits = Stream.of("Apple","Orange","Banana","Grape","Strawberry");
Flux flux = Flux.fromStream(fruits);
flux.subscribe(f -> System.out.println(f));

有时候我们根本没有可用的数据，而只是想要一个作为计数器的Flux，它会在每次发送新值时增加1。要创建一个计数器Flux，我们可以使用静态方法range()。

Flux flux = Flux.range(1, 5);
flux.subscribe(i -> System.out.println(i));

另一个与range()方法类似的Flux创建方法是interval()。与range()方法一样，interval()方法会创建一个发布递增值的Flux。interval()的特殊之处在于，我们不是给它设置一个起始值和结束值，而是指定一个应该每隔多长时间发出值的间隔时间。

Flux flux = Flux.interval(Duration.ofSeconds(1)).take(5);
flux.subscribe(i -> System.out.println(i));

有时候，我们会需要操作两种反应式类型，并以某种方式将它们合并在一起。或者，在其他情况下，我们可能需要将Flux拆分为多种反应式类型。
假设我们有两个Flux流，并且需要据此创建一个结果Flux，这个形成的Flux会在任意上游Flux流有数据时产生数据。要将一个Flux与另一个Flux合并，可以使用mergeWith()方法

Flux characterFlux = Flux.just("Garfield", "Kojak", "Barbossa").delayElements(Duration.ofMillis(500));
Flux foodFlux = Flux.just("Lasagna", "Lollipops", "Apples").delayElements(Duration.ofMillis(500)).delaySubscription(Duration.ofMillis(250));
Flux flux = characterFlux.mergeWith(foodFlux);

因为mergeWith()方法不能完美地保证源Flux之间的先后顺序，所以我们可以考虑使用zip()方法。当两个Flux对象压缩在一起的时候，它将会产生一个新的发布元组的Flux，其中每个元组中都包含了来自每个源Flux的数据项。

Flux characterFlux = Flux.just("Garfield", "Kojak", "Barbossa").delayElements(Duration.ofMillis(500));
Flux foodFlux = Flux.just("Lasagna", "Lollipops", "Apples").delayElements(Duration.ofMillis(500)).delaySubscription(Duration.ofMillis(250));
Flux> zip = Flux.zip(characterFlux, foodFlux);

可以为zip()方法提供一个合并函数来生成你想要的任何对象，合并函数会传入这两个数据项

Flux characterFlux = Flux.just("Garfield", "Kojak", "Barbossa").delayElements(Duration.ofMillis(500));
Flux foodFlux = Flux.just("Lasagna", "Lollipops", "Apples").delayElements(Duration.ofMillis(500)).delaySubscription(Duration.ofMillis(250));

Flux flux = Flux.zip(characterFlux, foodFlux, (a, b) -> a + "-" + b);

假设我们有两个Flux对象，此时我们不想将它们合并在一起，而是想要创建一个新的Flux，让这个新的Flux从第一个产生值的Flux中发布值。
first()操作会在两个Flux对象中选择第一个发布值的Flux，并再次发布它的值。

Flux slowFlux = Flux.just("tortoise", "snail", "sloth").delaySubscription(Duration.ofMillis(100));
Flux fastFlux = Flux.just("hare", "cheetah", "squirrel");

Flux first = Flux.first(slowFlux, fastFlux);

在数据流经一个流时，我们通常需要过滤掉某些值并对其他的值进行处理。

skip操作跳过指定数目的消息并将剩下的消息继续在结果Flux上进行传递

Flux skipFlux = Flux.just("one", "two", "skip a few", "ninety nine", "one hundred").skip(3);

你可能并不想跳过特定数量的条目，而是想要在一段时间之内跳过所有的第一批数据。这是skip()操作的另一种形式，将会产生一个新Flux，在发布来自源Flux的数据项之前等待指定的一段时间。

Flux skipFlux = Flux.just("one", "two", "skip a few", "ninety nine", "one hundred")
        .delayElements(Duration.ofSeconds(1))
        .skip(Duration.ofSeconds(4));

take操作只发布第一批指定数量的数据项，然后将取消订阅

Flux nationalParkFlux = Flux.just("Yellowstone", "Yosemite", "Grand Canyon", "Zion", "Grand Teton")
        .take(3);

take()方法也有另一种替代形式，基于间隔时间

Flux nationalParkFlux = Flux.just("Yellowstone", "Yosemite", "Grand Canyon", "Zion", "Grand Teton")
        .delayElements(Duration.ofSeconds(1))
        .take(Duration.ofMillis(3500));

skip操作和take操作都可以被认为是过滤操作，其过滤条件是基于计数或者持续时间的，而Flux值的更通用过滤则是filter操作。
我们需要指定一个Predicate，用于决定数据项是否能通过Flux，filter操作允许我们根据任何条件进行选择性地发布。

Flux nationalParkFlux = Flux.just("Yellowstone", "Yosemite", "Grand Canyon", "Zion", "Grand Teton")
        .filter(np -> !np.contains(" "));

想要过滤掉已经接收过的数据条目，可以采用distinct操作，形成的Flux将只会发布源Flux中尚未发布过的数据项。

Flux animalsFlux = Flux.just("dog", "cat", "bird", "dog", "bird", "anteater")
        .distinct();

在Flux或Mono上最常见的操作之一就是将已发布的数据项转换为其他的形式或类型。Reactor的反应式类型（Flux和Mono）为此提供了map和flatMap操作。
map操作会创建一个新的Flux，只是在重新发布它所接收的每个对象之前会执行给定Function指定的转换。

Flux stringFlux = Flux.just("Michael Jordan", "Scottie Pippen", "Steve Kerr");
Flux playerFlux = stringFlux.map(s -> {
    String[] s1 = s.split(" ");
    return new Player(s1[0], s1[1]);
});

map操作是同步执行的，如果你想要异步地转换过程，那么你应该考虑使用flatMap操作。
flatMap并不像map操作那样简单地将一个对象转换到另一个对象，而是将对象转换为新的Mono或Flux。结果形成的Mono或Flux会扁平化为新的Flux。当与subscribeOn()方法结合使用时，flatMap操作可以释放Reactor反应式的异步能力。

Flux stringFlux = Flux.just("Michael Jordan", "Scottie Pippen", "Steve Kerr");
Flux playerFlux = stringFlux.flatMap(s -> Mono.just(s)
        .map(p -> {
            String[] s1 = p.split(" ");
            return new Player(s1[0], s1[1]);
        }).subscribeOn(Schedulers.parallel()));

在处理流经Flux的数据时，你可能会发现将数据流拆分为小块会带来一定的收益

Flux fruitFlux = Flux.just("apple", "orange", "banana", "kiwi", "strawberry");
Flux> buffer = fruitFlux.buffer(3);

Flux.just("apple", "orange", "banana", "kiwi", "strawberry")
        .buffer(3)
        .flatMap(l ->
            Flux.fromIterable(l)
                    .map(s -> s.toUpperCase())
                    .subscribeOn(Schedulers.parallel())
                    .log()
        ).subscribe();

如果由于某些原因需要将Flux发布的所有数据项都收集到一个List中，那么可以使用不带参数的buffer()方法：

Flux> bufferedFlux = fruitFlux.buffer();

可以使用collectList操作实现相同的功能

Flux stringFlux = Flux.just("apple", "orange", "banana", "kiwi", "strawberry");
Mono> listMono = stringFlux.collectList();

一种更加有趣的收集Flux发出的数据项的方法是将它们收集到Map中。
collectMap 操作将会产生一个发布Map的Mono，这个Map中填充了由给定Function计算key值所生成的条目。

Flux animalFlux = Flux.just("aardvark", "elephant", "koala", "eagle", "kangaroo");

Mono> mapMono = animalFlux.collectMap(a -> a.charAt(0));

有时候我们想要知道由Mono或者Flux发布的条目是否满足某些条件，那么all()和any()方法可以实现这样的逻辑。
可以使用all()方法来确保Flux中的所有消息都满足某些条件

可以使用any()方法来确保Flux中至少有一个消息满足某些条件

Flux animalFlux = Flux.just(
        "aardvark", "elephant", "koala", "eagle", "kangaroo");

Mono hasAMono = animalFlux.all(a -> a.contains("a"));
StepVerifier.create(hasAMono)
        .expectNext(true)
        .verifyComplete();

Mono hasKMono = animalFlux.all(a -> a.contains("k"));
StepVerifier.create(hasKMono)
        .expectNext(false)
        .verifyComplete();

Flux animalFlux = Flux.just(
        "aardvark", "elephant", "koala", "eagle", "kangaroo");

Mono hasAMono = animalFlux.any(a -> a.contains("t"));

StepVerifier.create(hasAMono)
        .expectNext(true)
        .verifyComplete();

Mono hasZMono = animalFlux.any(a -> a.contains("z"));
StepVerifier.create(hasZMono)
        .expectNext(false)
        .verifyComplete();