如何从Flux中精确获取n个成功操作结果且不使用block()方法

你可以完全依赖Reactor内置的操作符实现需求，不需要手动维护计数器，也不需要调用block()做阻塞调用，实现代码如下：

// 假设你要处理的元素列表是items，n是需要拿到的成功结果数量
return Flux.fromIterable(items)
        // concurrency设为1，符合逐一执行的需求，如果允许并发可以调整为更大的值
        .flatMap(item -> operation(item)
                // 操作返回true视为成功，向下游发射当前item，失败则不发射元素
                .filter(Boolean::booleanValue)
                .thenReturn(item), 1)
        // 拿到n个成功元素后自动终止流，向上游发送取消信号
        .take(n)
        // 如果需要返回n个成功的结果，把then()替换为collectList()即可
        .then();

方案说明 #

• 消除了block()阻塞调用，全程是异步非阻塞的响应式逻辑，不会浪费线程资源
• 不需要手动维护AtomicLong计数器，避免并发场景下的计数线程安全问题
• flatMap的concurrency参数控制处理并发度：如果你的需求要求严格逐一处理，保持为1即可；如果允许并行处理元素，调大该参数可以提升处理效率，流依然会在拿到n个成功结果后立即终止
• 如果你需要保留成功处理的n个元素的结果，将末尾的then()替换为collectList()，返回值就是Mono<List<Integer>>，可以直接拿到成功结果集合

小提示：如果你的operation(item)可能抛出异常，且你希望忽略异常继续处理后续元素，可以在operation(item)后面追加.onErrorResume(e -> Mono.empty())，把异常也视为操作失败，不向下游发射元素。

内容的提问来源于stack exchange，提问作者Andy