如何使用状态过滤通量？

Question

我想根据从先前值计算的状态在我的Flux 上应用filter。不过根据the javadoc建议避免在算子中使用状态

请注意，应避免在 Flux 运算符中使用的 java.util.function / lambdas 中使用状态，因为这些可能在多个订阅者之间共享。

例如，Flux#distinct 过滤较早出现的项目。我们如何实现我们自己的distinct 版本？

Answer 1

我找到了我的问题的答案。 Flux#distinct 可以采用提供初始状态的Supplier 和执行“不同”检查的BiPredicate，因此我们可以在存储中存储任意状态并决定是否保留每个元素。

以下代码展示了如何在不改变顺序的情况下保留每个 mod2 组的前 3 个元素。

// Get first 3 elements per mod 2.
Flux<Integer> first3PerMod2 =
    Flux.fromIterable(ImmutableList.of(9, 3, 7, 4, 5, 10, 6, 8, 2, 1))
        .distinct(
            // Group by mod2
            num -> num % 2,
            // Counter to store how many elements have been processed for each group.
            () -> new HashMap<Integer, Integer>(),
            // Increment or set 1 to the counter,
            // and return whether 3 elements are published.
            (map, num) -> map.merge(num, 1, Integer::sum) <= 3,
            // Clean up the state.
            map -> map.clear());

StepVerifier.create(first3PerMod2).expectNext(9, 3, 7, 4, 10, 6).verifyComplete();