Webflux中用skip()和take()分页时，如何获取过滤后的总条目数？

你当前的两种实现都有局限：第一种拿不到过滤后的总条数，第二种会把所有过滤数据加载到内存，数据量大时性能堪忧。要在流层面实现需求，可以通过共享过滤后的数据流，并行获取总条数和分页数据，再合并结果的方式解决，具体代码如下：

public Mono<CustomersResponse> getCustomers(Integer page, Integer pageSize, String filter) {
    // 先定义过滤后的数据流，并用cache()缓存结果，避免多次订阅重复执行上游查询和过滤
    Flux<Customer> filteredCustomers = customersService.getCustomers()
            .filter(customer -> someFilterFunction(filter))
            .cache();

    // 并行获取两个结果：过滤后的总条数、当前页的数据列表
    Mono<Long> totalCountMono = filteredCustomers.count();
    Mono<List<Customer>> pageItemsMono = filteredCustomers
            .skip((long) page * pageSize)
            .take(pageSize)
            .collectList();

    // 合并两个Mono的结果，构建响应对象
    return Mono.zip(totalCountMono, pageItemsMono)
            .map(tuple -> new CustomersResponse()
                    .items(tuple.getT2())
                    .totalCount(tuple.getT1().intValue()));
}

关键说明： #

• 共享数据流：使用cache()可以让filteredCustomers的两次订阅（count和分页）共享同一个上游数据流，避免重复调用customersService.getCustomers()和执行过滤逻辑，提升性能。如果需要更精细的控制，也可以用publish().autoConnect(2)，确保只有当两个订阅都存在时才触发上游数据流。
• 流层面处理：整个过程没有将所有过滤后的数据加载到内存，分页时依然是按需skip和take，兼顾了内存效率和功能需求。
• 并行合并：Mono.zip会等待两个异步操作都完成后再合并结果，保证响应对象中的分页数据和总条数是匹配的。

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