Reactor框架中如何从List<String>中过滤掉Flux<String>包含的元素

已知条件 #

• 普通字符串集合：List<String> list = List.of("A", "B", "C", "D")
• 响应式字符串流：Flux<String> flux = Flux.just("A", "B")
• 预期输出：过滤掉list中所有flux存在的元素，得到结果 ["C", "D"]

常见问题说明 #

你提到的filterWhen仅返回第一个符合条件元素的问题，本质是错误使用导致的：如果直接在filterWhen的判断逻辑中每次都重新消费Flux，既会带来成倍的性能开销，也会因为Flux的冷流特性出现匹配结果异常的问题。

最优实现方案 #

做差集运算的前提是拿到所有需要排除的元素，因此优先将Flux的元素预收集到内存Set中，再做过滤，实现代码如下：

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

// 预收集Flux元素为Set，缓存结果避免重复消费
Mono<Set<String>> excludeSetMono = flux.collect(Collectors.toSet());

// 基于排除Set过滤原始List的元素
Flux<String> diffResult = excludeSetMono.flatMapMany(excludeSet ->
        Flux.fromIterable(list)
                .filter(item -> !excludeSet.contains(item))
);

// 非响应式上下文可通过block拿到最终List（响应式上下文直接订阅diffResult即可）
// List<String> resultList = diffResult.collectList().block();

该方案仅消费一次原始Flux，过滤逻辑是普通的内存Set存在性判断，性能远高于逐元素匹配Flux的实现。

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