基于RxJava实现分页API循环调用直至获取全部数据

刚好做过类似的需求，我来分享两种实用的实现方式，核心都是利用RxJava的操作符来递归处理分页请求，直到拉取完所有数据。

核心思路 #

我们的目标是：从第一页开始请求，每拿到一页数据后判断是否还有下一页（通过totalCount和当前页的索引、每页大小计算），如果有就自动请求下一页，直到所有数据都获取完毕，最终返回完整的数据集。

方法一：递归+FlatMap（直观易懂） #

这种方式用递归调用结合flatMap来处理每一页的请求，逻辑清晰，适合刚接触RxJava的同学：

// 假设你的QueryResult里包含List<DataItem> dataList字段，DataItem是你的数据模型
public Observable<List<DataItem>> fetchAllData(QueryParams initialParams) {
    return yourApiService.queryData(initialParams)
            .flatMap(queryResult -> {
                // 取出当前页的数据
                List<DataItem> currentPage = queryResult.getDataList();
                
                // 判断是否还有下一页：这里假设pageIndex从0开始，根据你的API调整
                boolean hasMore = (queryResult.getPageIndex() + 1) * queryResult.getPageSize() < queryResult.getTotalCount();
                
                if (hasMore) {
                    // 构建下一页的请求参数，一定要保留初始的查询条件！
                    QueryParams nextParams = new QueryParams();
                    nextParams.setPageIndex(queryResult.getPageIndex() + 1);
                    nextParams.setPageSize(queryResult.getPageSize());
                    // 复制其他查询条件，比如关键词、筛选条件等
                    nextParams.setKeyword(initialParams.getKeyword());
                    nextParams.setCategory(initialParams.getCategory());
                    
                    // 递归调用，合并当前页和后续页的数据
                    return fetchAllData(nextParams)
                            .map(nextPageData -> {
                                List<DataItem> allData = new ArrayList<>(currentPage);
                                allData.addAll(nextPageData);
                                return allData;
                            });
                } else {
                    // 没有更多数据，直接返回当前页
                    return Observable.just(currentPage);
                }
            });
}

方法二：用Expand操作符（更简洁的RxJava风格） #

RxJava的expand操作符专门用于递归展开Observable，非常适合分页场景，代码更简洁：

public Observable<List<DataItem>> fetchAllData(QueryParams initialParams) {
    return Observable.defer(() -> {
        // 复制初始参数，避免修改原对象
        QueryParams startParams = new QueryParams();
        startParams.setPageIndex(initialParams.getPageIndex());
        startParams.setPageSize(initialParams.getPageSize());
        startParams.setKeyword(initialParams.getKeyword());
        startParams.setCategory(initialParams.getCategory());
        
        return yourApiService.queryData(startParams)
                // expand：把每个发射的QueryResult展开成新的请求Observable，直到返回empty
                .expand(queryResult -> {
                    boolean hasMore = (queryResult.getPageIndex() + 1) * queryResult.getPageSize() < queryResult.getTotalCount();
                    if (hasMore) {
                        QueryParams nextParams = new QueryParams();
                        nextParams.setPageIndex(queryResult.getPageIndex() + 1);
                        nextParams.setPageSize(queryResult.getPageSize());
                        nextParams.setKeyword(initialParams.getKeyword());
                        nextParams.setCategory(initialParams.getCategory());
                        return yourApiService.queryData(nextParams);
                    } else {
                        // 返回empty表示停止递归
                        return Observable.empty();
                    }
                })
                // reduce：把所有页的数据合并成一个完整的List
                .reduce(new ArrayList<>(), (allData, currentResult) -> {
                    allData.addAll(currentResult.getDataList());
                    return allData;
                })
                // 把Single转成Observable（reduce返回的是Single）
                .toObservable();
    });
}

使用示例 #

调用的时候记得处理线程调度，网络请求一定要放在IO线程：

// 构造初始请求参数
QueryParams params = new QueryParams();
params.setPageIndex(0); // 根据你的API调整起始索引（0或1）
params.setPageSize(20);
params.setKeyword("example");
params.setCategory("android");

fetchAllData(params)
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(
                allData -> {
                    // 拿到所有数据了，做你需要的处理
                    Log.d("Pagination", "Total data fetched: " + allData.size());
                },
                error -> {
                    // 处理请求失败的情况
                    Log.e("Pagination", "Fetch all data failed", error);
                }
        );

重要注意事项 #

1. 分页索引适配：一定要确认你的API的pageIndex是从0还是1开始的，调整判断hasMore的条件（比如如果从1开始，条件就是queryResult.getPageIndex() * queryResult.getPageSize() < queryResult.getTotalCount()）
2. 线程调度：必须用subscribeOn(Schedulers.io())指定网络请求在IO线程执行，否则会阻塞主线程
3. 异常处理：如果某一页请求失败，整个Observable会触发onError，你可以根据需求添加retry(3)来重试几次，或者用onErrorResumeNext来优雅处理错误
4. 内存优化：如果总数据量极大，一次性合并成一个List可能会占用过多内存，这种情况下可以考虑去掉reduce，直接在expand的下游每收到一页数据就处理，而不是等全部加载完

内容的提问来源于stack exchange，提问作者Pai-Hsiang Huang