如何计算 Flux 中的项目数，如果计数大于 X，则返回错误，否则继续使用 Pipeline

Question

我是 Spring 项目 Reactor 的新手，我不完全确定如何执行某些操作：

我有我的管道管道返回记录。都好。

但我想计算这些记录，然后做一些事情（比如 if else），如果返回的记录 > X 则出错，否则继续。

知道 Count 返回 a Mono<Long>，那么之后我会丢失记录，我该怎么办？

我在想：

以某种方式使用flatMap并执行此平面图内的某些操作。不知何故，我发现 Flux 中有一种reduce方法可能会有所帮助。

关键是，我不知道如何继续。

Answer 1

不完全确定您想要什么，因此将根据假设提供两个建议

1..你想要收集所有元素，然后评估是否有超过n个，如果是则抛出错误。您可以使用collectList，对元素进行计数，然后在任何情况下转换回通量。doStuff仅当总数低于限制时，这才适用于任何元素。

    Flux.range(1,10)
            .collectList()
            .flatMap(s -> 
                s.size()>7 
                    ? Mono.error(new RuntimeException("TOO MANY!")) 
                    : Mono.just(s))
            .flatMapMany(Flux::fromIterable)
            .map(this::doStuff)

2.. 您想要动态评估元素的数量，您可以使用外部原子计数器。这将doStuff涉及到有问题的每一个元素。

    AtomicLong count = new AtomicLong();

    Flux.range(1,10)
            .flatMap(s -> 
                count.incrementAndGet() > 7 
                    ? Flux.error(new RuntimeException("TOO MANY!")) 
                    : Flux.just(s))
             .map(this::doStuff);