如何不阻塞地利用Filter Flux过滤Range Flux中的值？

当然可以，完全不用阻塞的toStream()操作就能实现需求。核心是利用Reactor的非阻塞收集方法，结合响应式操作符完成过滤，全程保持异步非阻塞特性：

Flux<String> filterFlux = Flux.just("2", "3", "4", "5");
Flux<Integer> rangeFlux = Flux.range(1, 10);

StepVerifier.create(
        // 非阻塞收集filterFlux的所有元素到List
        filterFlux.collectList()
                // 用收集到的List过滤rangeFlux
                .flatMapMany(filterList -> 
                        rangeFlux.map(String::valueOf)
                                .filter(s -> !filterList.contains(s))
                )
)
.expectNext("1")
.expectNextSequence(List.of("6", "7", "8", "9", "10"))
.expectComplete()
.verify();

关键说明： #

• collectList()：异步收集filterFlux的所有元素到List，返回Mono<List<String>>，全程不会阻塞线程
• flatMapMany：将Mono中拿到的过滤列表，用来处理rangeFlux的过滤逻辑，最终返回过滤后的Flux
• 整个流程完全遵循响应式编程的非阻塞原则，避免了原代码中toStream()带来的线程阻塞问题

如果你的filterFlux是无限流，这种方式不适用（因为collectList()会一直等待流完成），但针对你给出的有限流场景，这个方案完全可行。

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