如何组合同步与异步Observable：登录后并行执行Retrofit请求

当然有可行的实现方式啦！这正是RxJava这类响应式框架最擅长搞定的场景——先完成前置的登录任务，再并行启动后续多个API调用，整个流程的控制非常丝滑。

核心思路 #

我们可以利用RxJava的操作符组合来实现：

1. 先订阅登录的Observable，确保登录流程完全结束（拿到登录成功的响应或处理登录失败）。
2. 登录成功后，通过操作符并行触发另外4个API调用的Observable，根据需求选择合并结果或单独处理每个请求的响应。

代码示例（合并所有结果） #

如果需要等4个API都执行完成后，统一处理它们的结果，推荐用flatMap + Observable.zip的组合：

// 1. 定义各个Observable（Retrofit API调用）
Observable<LoginResponse> loginObservable = apiService.login(yourLoginCredentials);
Observable<DataModelA> dataAObservable = apiService.fetchDataA();
Observable<DataModelB> dataBObservable = apiService.fetchDataB();
Observable<DataModelC> dataCObservable = apiService.fetchDataC();
Observable<DataModelD> dataDObservable = apiService.fetchDataD();

// 2. 组合执行流程
loginObservable
    .flatMap(loginResponse -> {
        // 登录成功后，并行启动4个API调用
        return Observable.zip(
            dataAObservable,
            dataBObservable,
            dataCObservable,
            dataDObservable,
            (dataA, dataB, dataC, dataD) -> {
                // 将4个接口的结果封装成一个组合对象返回
                return new CombinedData(dataA, dataB, dataC, dataD);
            }
        );
    })
    .subscribeOn(Schedulers.io()) // 网络请求放在IO线程
    .observeOn(AndroidSchedulers.mainThread()) // 结果回调切回主线程（Android场景）
    .subscribe(
        combinedData -> {
            // 处理合并后的所有结果
            handleCombinedData(combinedData);
        },
        throwable -> {
            // 处理错误：登录失败 / 任意一个API调用失败都会触发这里
            handleRequestError(throwable);
        }
    );

代码示例（单独处理每个API结果） #

如果不需要等待所有API完成，而是每个API返回结果就立即处理，可以用flatMap + Observable.merge的组合：

loginObservable
    .flatMap(loginResponse -> {
        // 并行启动4个API，每个结果单独发射
        return Observable.merge(
            dataAObservable,
            dataBObservable,
            dataCObservable,
            dataDObservable
        );
    })
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(
        response -> {
            // 根据返回数据的类型，分别处理
            if (response instanceof DataModelA) {
                handleDataA((DataModelA) response);
            } else if (response instanceof DataModelB) {
                handleDataB((DataModelB) response);
            }
            // ... 处理其他类型
        },
        throwable -> {
            handleRequestError(throwable);
        }
    );

关键操作符说明 #

• flatMap：负责将登录Observable的结果，转换为后续的并行任务Observable，确保登录完成后才启动后续请求。
• Observable.zip：会并行订阅所有传入的Observable，等待所有Observable都发射完数据后，将结果组合成一个新对象发射，适合需要统一处理所有结果的场景。
• Observable.merge：同样并行订阅所有Observable，但只要有一个Observable发射数据，就立即将该数据发射出去，适合需要实时处理单个请求结果的场景。

容错优化（可选） #

如果希望某个API调用失败不影响其他请求，可以给单个Observable添加容错处理，比如onErrorReturn：

// 给dataAObservable添加容错，失败时返回默认值
Observable<DataModelA> safeDataAObservable = dataAObservable
    .onErrorReturn(throwable -> {
        Log.e("API_ERROR", "DataA请求失败", throwable);
        return new DataModelA(); // 返回默认空对象或占位数据
    });

// 后续用safeDataAObservable替代原对象即可

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