如何在所有并行HTTP请求完成后调用方法？（RxJava实现需求）

没问题，我给你整理一套不用lambda的RxJava实现方案，完美匹配你的需求——先批量获取分类，再逐个拉取对应频道，等所有数据都拿到后再执行最终方法：

第一步：改造Retrofit接口 #

首先把原来的Call返回类型改成Observable，这样才能和RxJava无缝配合，同时补充获取频道的接口：

public interface VideoAPI {
    @GET("/api/{categoryId}")
    Observable<Category> getCategory(@Path("categoryId") String categoryId);

    // 假设你的频道接口是这样的，根据实际情况调整
    @GET("/api/categories/{categoryId}/channels")
    Observable<List<Channel>> getChannels(@Path("categoryId") String categoryId);
}

第二步：核心实现代码 #

接下来写具体的业务逻辑，全程用匿名内部类，完全不用lambda：

private void loadAllCategoriesAndChannels() {
    // 1. 把tabs列表转成Observable，逐个处理每个Tab
    Observable.fromIterable(tabs)
            // 发起分类请求，将Tab转换为Category的Observable
            .flatMap(new Function<Tab, Observable<Category>>() {
                @Override
                public Observable<Category> apply(Tab tab) throws Exception {
                    return videoAPI.getCategory(tab.getId())
                            // 处理单个分类请求失败的情况，避免整个流中断
                            .onErrorReturn(new Function<Throwable, Category>() {
                                @Override
                                public Category apply(Throwable throwable) throws Exception {
                                    Log.i(TAG, "获取分类失败: " + throwable.getLocalizedMessage());
                                    return null; // 也可以返回空Category对象，根据需求调整
                                }
                            });
                }
            })
            // 过滤掉获取失败的分类（如果返回null的话）
            .filter(new Predicate<Category>() {
                @Override
                public boolean test(Category category) throws Exception {
                    return category != null;
                }
            })
            // 对每个分类，发起频道请求，绑定分类和频道数据
            .flatMap(new Function<Category, Observable<CategoryWithChannels>>() {
                @Override
                public Observable<CategoryWithChannels> apply(final Category category) throws Exception {
                    return videoAPI.getChannels(category.getId())
                            // 将频道列表和分类绑定成一个对象
                            .map(new Function<List<Channel>, CategoryWithChannels>() {
                                @Override
                                public CategoryWithChannels apply(List<Channel> channels) throws Exception {
                                    return new CategoryWithChannels(category, channels);
                                }
                            })
                            // 处理单个频道请求失败的情况
                            .onErrorReturn(new Function<Throwable, CategoryWithChannels>() {
                                @Override
                                public CategoryWithChannels apply(Throwable throwable) throws Exception {
                                    Log.i(TAG, "获取频道失败: " + throwable.getLocalizedMessage());
                                    // 即使频道获取失败，也保留分类信息
                                    return new CategoryWithChannels(category, new ArrayList<Channel>());
                                }
                            });
                }
            })
            // 等待所有请求完成，收集所有结果为一个List
            .toList()
            // 指定网络请求在IO线程执行
            .subscribeOn(Schedulers.io())
            // 指定结果处理在主线程（Android环境下必须）
            .observeOn(AndroidSchedulers.mainThread())
            // 订阅结果
            .subscribe(new Observer<List<CategoryWithChannels>>() {
                @Override
                public void onSubscribe(Disposable d) {
                    // 可选：保存Disposable，用于在页面销毁时取消请求
                    // disposable = d;
                }

                @Override
                public void onNext(List<CategoryWithChannels> allData) {
                    // 这里已经拿到所有分类和对应的频道数据了
                    categories.clear();
                    for (CategoryWithChannels item : allData) {
                        categories.add(item.getCategory());
                        // 可以在这里保存频道数据，比如：
                        // channelMap.put(item.getCategory().getId(), item.getChannels());
                    }
                    // 调用你需要的最终方法
                    onAllDataLoadedSuccessfully();
                }

                @Override
                public void onError(Throwable e) {
                    Log.i(TAG, "整体加载出错: " + e.getLocalizedMessage());
                    // 可选：处理整体出错的情况，比如提示用户
                }

                @Override
                public void onComplete() {
                    // onNext执行后会触发这个方法，一般不需要额外逻辑
                }
            });
}

// 辅助类：用来绑定分类和对应的频道数据，方便后续统一处理
private static class CategoryWithChannels {
    private final Category category;
    private final List<Channel> channels;

    public CategoryWithChannels(Category category, List<Channel> channels) {
        this.category = category;
        this.channels = channels;
    }

    public Category getCategory() {
        return category;
    }

    public List<Channel> getChannels() {
        return channels;
    }
}

// 你需要的最终回调方法：所有数据加载完成后调用
private void onAllDataLoadedSuccessfully() {
    // 这里写所有数据准备好后的业务逻辑
}

第三步：配置依赖和Retrofit #

确保你的项目中添加了必要的依赖：

// RxJava核心库
implementation 'io.reactivex.rxjava2:rxjava:2.2.21'
// RxAndroid：适配Android线程
implementation 'io.reactivex.rxjava2:rxandroid:2.1.1'
// Retrofit的RxJava适配器
implementation 'com.squareup.retrofit2:adapter-rxjava2:2.9.0'
// 别忘了Retrofit和Gson转换器（如果用的话）
implementation 'com.squareup.retrofit2:retrofit:2.9.0'
implementation 'com.squareup.retrofit2:converter-gson:2.9.0'

创建Retrofit实例时，要添加RxJava2CallAdapterFactory：

Retrofit retrofit = new Retrofit.Builder()
        .baseUrl("你的服务器基础URL")
        .addConverterFactory(GsonConverterFactory.create())
        .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
        .build();
VideoAPI videoAPI = retrofit.create(VideoAPI.class);

关键逻辑说明 #

• Observable.fromIterable(tabs)：把tabs集合转换成Observable，逐个发射每个Tab对象；
• flatMap：用来将一个Observable转换成另一个Observable，这里用来发起网络请求，把Tab转换成Category，再把Category转换成带频道的对象；
• onErrorReturn：处理单个请求失败的情况，保证整个数据流不会因为某一个请求失败而中断；
• toList()：等待所有Observable都发射完数据，然后把结果收集成一个List，这样就能确保所有数据都加载完成后才执行后续逻辑；
• subscribeOn和observeOn：分别指定网络请求在IO线程，结果处理在主线程，符合Android的线程规范。

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