使 filterWhen() 与 Mono 谓词并行

Posted 2023-02-22

【中文标题】使 filterWhen() 与 Mono 谓词并行

【英文标题】：making filterWhen() with Mono predicate parallel

【发布时间】：2021-01-22 19:25:03

【问题描述】：

使用 Reactor Flux 过滤器时，我看到了一些我需要克服的行为。

给定以下代码：

Flux.fromIterable(List.of(1,2))
  .filterWhen(it -> predicateMono(it))

地点：

Mono<boolean> predicateMono(int value)  ...  

我注意到predicateMono()s 是按顺序执行的，这意味着对于值 2，在第一个完成之前不会调用操作。

当我的代码中的predicateMono()s 是对我想并行执行的后端系统的http 调用时，这会成为一个问题。如何编写此代码以便可以并行过滤通量值？

predicateMono() 是非阻塞 http 调用，与响应式方法兼容。

【问题讨论】：

在使用之前阅读标签。单声道标签不适用于 Java。

【参考方案1】：

解决方案是使用flatmap 而不是filterwhen。如果您想过滤掉，只需映射到空。 Flux 上的空值会被后续操作忽略。

Flux.fromIterable(List.of(1, 2))
        .flatMap(it ->
            predicateMono(it).flatMap(result ->
                result ? Mono.just(it) : Mono.empty()));