Flux：当某些元素具有具体价值时，如何防止元素处理？

Question

以下测试方法：

@Test
void testMe() {
    Flux.just(1, 2, 3, 4, 5)
            .map(this::saveInDb)
            .toStream().count();
}

int saveInDb(int element) {
    System.out.println(element + " successfully stored in DB.");
    return element;
}

总是打印：

1 successfully stored in DB.
2 successfully stored in DB.
3 successfully stored in DB.
4 successfully stored in DB.
5 successfully stored in DB.

我的问题：当 Flux 中有某些特定元素可用时，如何防止任何保存到 DB？

例如：当 Flux 中存在元素 5 时，我不想将任何内容存储到 DB。

是否有可能以非阻塞方式实现此要求？

Answer 1

只有将整个Flux 放入内存中才有可能：

Flux.just(1, 2, 3, 4, 5)
    .collectList()
    .filter(items -> !items.contains(5))
    .flatMapIterable(x -> x)
    .map(this::saveInDb)

如果 Flux 无法装入内存，您可以：

• 缓冲 Flux 并仅检查缓冲子集中是否存在（这可能不适合您的用例）
• 使用Flux 两次，首先使用过滤器检查是否存在，如果该项目不存在然后再次订阅（这显然只有在Flux 保证在两种情况下都具有相同的输出时才有效）李>