根据没有 GroupBy 的条件将 Flux 一分为二

Question

对于给定的数据流：

Flux<Integer> evenFlux = Flux.just(1, 2, 3, 4, 5, 6, 7)
                         .filter(i -> i % 2 == 0)

Flux<Integer> oddFlux = Flux.just(1, 2, 3, 4, 5, 6, 7)
                         .filter(i -> i % 2 != 0)

我怎样才能使用单个管道进行拆分？ 所有过滤的元素到一个Flux，丢弃的元素到另一个Flux。使用onDiscardHook 什么的？

注意：我需要对新的Flux 进行repeat() 操作。无法在 GroupedFlux 上执行重复。

Answer 1

视情况而定，您有两种选择：

选项 1 是从热输入源创建两个新管道：

ConnectableFlux<Integer> input = Flux.range(1,7).publish();

//Pipeline 1
input.filter(i -> i % 2== 0)
    .subscribe(e -> System.out.println("Even stream value: " + e));

//Pipeline 2
input.filter(i -> i % 2 != 0)
    .subscribe(o -> System.out.println("Odd stream value: " + o));

input.connect();

选项 2 是在单个订阅者中处理结果，例如按以下方式分组：

Flux.just(1,2,3,4,5,6,7)
    .subscribe(n -> {
      if (n % 2 == 0) {
        System.out.println("Even stream value: " + n);
      } else {
        System.out.println("Odd stream value: " + n);
      }
    });

Answer 2

正如@Michael Berry 所建议的，使用groupBy 是解决方案之一。
我们可以使用i % 2进行分组，返回一个Flux<GroupedFlux>，然后我们可以使用GroupedFlux#key()来区分奇数组或偶数组。

public class SplitFlux {
    public static void main(String[] args) {
        Flux.just(1, 2, 3, 4, 5, 6, 7).groupBy(i -> i % 2).toStream()
                .forEach(groupedFlux ->
                {
                    System.out.println(groupedFlux.key() == 0 ? "even" : "odd");
                    groupedFlux.toStream().forEach(System.out::println);
                });
    }
}