如何计数流量中的项目，如果计数大于X，返回错误，否则继续流水线

Question

我在想：

以某种方式使用flatmap并在此flatmap中执行某些操作。不知何故，我发现Flux中有一个reduce方法可能会有所帮助。

关键是，我不知道如何进行。

Answer 1

我不完全确定你想要什么，所以在假设的基础上提供两个建议

1...您希望收集所有元素，然后评估是否有超过n的元素，如果是，抛出错误。您可以使用collectList，计数元素，然后转换回flux（如果在任何情况下）。如果总数低于限制，这将只对任何元素执行dostufle。

    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..您想要评估动态元素的数量，您可以使用外部原子计数器对其进行评估。这将对每个元素dostufle直到有问题的元素。

    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);