Spring WebFlux中如何并行调用两个Flux并完成后续过滤处理？

不需要用CompletableFuture，Spring WebFlux本身提供了更贴合反应式编程模型的方式来实现并行调用。你之前用zipWith没达到预期，是因为zipWith是按元素顺序一对一配对两个Flux，而你需要的是先拿到两个数据源的完整结果再做后续处理。

实现思路 #

1. 并行获取完整数据集：

   • 将getDatesToExclude()收集为Mono<Set<Date>>，用Set是为了后续快速判断日期是否需要排除
   • 对getAvailableUserDates()过滤出指定用户的数据，再转为Mono<UserDates>
   • 这两个Mono的订阅会自动并行执行（只要你的数据源是异步非阻塞的，比如外部API调用、非阻塞文件读取）

2. 组合结果并处理日期过滤：

   • 用Mono.zip()组合两个Mono的结果，拿到排除日期集合和指定用户的可用日期
   • 从用户可用日期中移除排除列表里的日期

完整代码实现 #

public Flux<UserDates> getAvailableUserDates(String user) {
    // 1. 并行获取排除日期集合（转成Set提升查询效率）
    Mono<Set<Date>> excludedDatesMono = getDatesToExclude()
            .collect(Collectors.toSet());

    // 2. 过滤指定用户的可用日期，转为Mono（因为每个用户对应一条UserDates数据）
    Mono<UserDates> userDatesMono = getAvailableUserDates()
            .filter(userDates -> user.equals(userDates.getUser()))
            .singleOrEmpty(); // 如果可能有多个匹配，用first()或者根据业务调整

    // 3. 组合两个并行结果，处理日期排除
    return Mono.zip(excludedDatesMono, userDatesMono)
            .map(tuple -> {
                Set<Date> excludedDates = tuple.getT1();
                UserDates userDates = tuple.getT2();

                if (userDates == null || userDates.getDates() == null) {
                    return userDates; // 无数据时直接返回
                }

                // 从可用日期中移除排除日期
                List<Date> filteredDates = userDates.getDates().stream()
                        .filter(date -> !excludedDates.contains(date))
                        .collect(Collectors.toList());

                userDates.setDates(filteredDates);
                return userDates;
            })
            .flux(); // 转成Flux符合方法返回值要求
}

关键细节说明 #

• 为什么不用zipWith：zipWith会把两个Flux的元素按顺序一一配对，比如第一个排除日期和第一个UserDates配对，这不符合你需要拿到全部排除日期和指定用户全部日期的需求。
• 并行性保证：只要getDatesToExclude()（比如用WebClient调用外部API）和getAvailableUserDates()（非阻塞读取JSON文件）本身是异步非阻塞的，Spring WebFlux会自动在不同的调度线程上并行执行这两个操作，不需要额外配置。
• 集合选择：把排除日期转成Set是因为contains()操作的时间复杂度是O(1)，比List的O(n)更高效，尤其是当排除日期数量较多时。

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