RxJava2如何合并多个Observable并返回单次发射的单一结果？

这问题我熟！要实现多页数据合并后单次发射，咱们可以用RxJava的操作符组合来搞定，刚好适配你这个Retrofit分页请求的场景。

核心思路 #

1. 先发起第一页请求，拿到初始帖子列表和下一页ID
2. 自动递归请求后续页面，直到没有下一页（根据nextPage判断）
3. 把所有页面的帖子列表合并成一个大列表
4. 最终仅发射一次合并后的完整列表

代码实现 #

假设你已经有了MyService的实例，我们可以封装一个方法来实现这个逻辑：

public Observable<List<UserPost>> getAllUserPosts(int userId) {
    // 初始请求第一页（如果你的接口第一页不需要page参数，就调用getUserPosts(userId)）
    return service.getUserPosts(userId, 1)
            // 递归获取后续分页：expand会自动订阅每个新生成的Observable，直到返回empty终止
            .expand(result -> {
                // 根据nextPage判断是否有下一页，这里假设有效页码>0
                if (result.mPageId > 0) {
                    return service.getUserPosts(userId, result.mPageId);
                } else {
                    // 没有下一页，返回空Observable结束递归
                    return Observable.empty();
                }
            })
            // 提取每个请求结果中的帖子列表
            .map(result -> result.mPosts)
            // 合并所有列表为一个大列表：reduce只会在所有请求完成后发射一次最终结果
            .reduce(new ArrayList<>(), (totalPosts, currentPagePosts) -> {
                totalPosts.addAll(currentPagePosts);
                return totalPosts;
            })
            // 处理无数据的情况：确保即使没有任何帖子，也会发射一个空列表
            .defaultIfEmpty(new ArrayList<>());
}

关键操作符解释 #

• expand：专门用于递归场景的操作符，它会先处理第一个请求的结果，然后根据结果生成下一个请求的Observable，直到返回Observable.empty()才停止，完美适配分页递归请求。
• map：把每个UserPostsResult转换成对应的帖子列表，简化后续的合并逻辑。
• reduce：从初始空列表开始，将每一页的帖子列表累加进去，只有当所有源Observable都完成时，才会发射最终合并后的大列表，刚好满足你“仅发射一次”的需求。
• defaultIfEmpty：避免因用户无帖子导致Observable没有任何发射，保证订阅者总能收到回调。

额外注意事项 #

• 确保UserPostsResult的nextPage有明确的终止标记（比如无下一页时返回0），这样expand的判断逻辑才能正常工作。
• 可以添加异常处理，比如用onErrorResumeNext处理网络请求失败的情况：

.onErrorResumeNext(throwable -> {
    // 这里可以根据需求处理错误，比如返回空列表或自定义异常
    Log.e("PostsRequest", "获取帖子失败", throwable);
    return Observable.just(new ArrayList<>());
})

• 如果你的第一页请求不需要page参数，只需要把初始请求换成service.getUserPosts(userId)即可，后续逻辑保持不变。

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