Spring WebFlux如何并发执行多分页请求并按序收集结果

你的原有递归实现存在3个核心问题：

• 完全串行执行：必须等上一页请求返回才会发起下一页，没有并行能力
• 共享可变List线程不安全：高并发下会出现数据覆盖、丢失的问题
• 递归逻辑没有利用总页数信息，存在栈溢出风险，也无法做并发控制

要实现分页请求并行执行、结果严格按页码顺序输出的需求，直接利用Reactor内置的mergeSequential算子即可，这个算子的特性就是异步订阅所有上游源并行执行，但会严格按照上游源的顺序发射结果，完全匹配需求。

完整实现代码 #

public Flux<Item> listAllItemsInOrder() {
    // 首先请求第1页，获取总页数信息
    return getAPageOfItems(1)
            .flatMapMany(firstPage -> {
                int totalPage = firstPage.totalPage;
                // 总页数为1时直接返回第一页数据
                if (totalPage <= 1) {
                    return Flux.fromIterable(firstPage.items);
                }

                // 生成剩余页码的所有请求，和第一页合并为完整的分页流
                return Flux.range(2, totalPage - 1)
                        // 将每个页码转为对应分页请求的Mono
                        .map(pageNum -> getAPageOfItems(pageNum)
                                // 可在此处配置单请求重试逻辑，应对临时网络异常
                                .retry(3)
                        )
                        // mergeSequential: 并行发起所有请求，严格按源顺序输出结果
                        // 第二个参数为最大并发数，可根据下游服务承载能力调整，避免请求打满
                        .as(pageFlux -> Flux.mergeSequential(pageFlux, 10))
                        // 第一页放在结果最前端
                        .startWith(firstPage)
                        // 可选：增加按页号排序的兜底逻辑，避免接口异常返回乱序页
                        // .sort(Comparator.comparingInt(pageOfItem -> pageOfItem.page))
                        // 逐页展开条目，输出最终的Item流
                        .flatMapIterable(pageOfItem -> pageOfItem.items);
            });
}

关键说明 #

• 并行控制：mergeSequential的第二个参数指定最大在途请求数，比如示例中设为10，代表最多同时发起10个分页请求，不会一次性把所有分页请求打出去把下游服务压垮
• 顺序保证：mergeSequential会缓存提前返回的高页码数据，等前面所有页码的数据返回后才会向后发射，严格保证输出顺序和页码顺序一致
• 线程安全：全程没有使用共享可变集合，所有数据转换都在响应式流内完成，不存在并发安全问题
• 背压支持：实现完全遵循Reactive Streams规范，天然支持背压，不会出现数据生产速度快于消费速度导致的内存溢出问题

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