RXJava中Zip方法实现：多并行API请求结果合并为列表

嘿，这事儿RxJava简直是为它量身定做的！我来给你捋清楚怎么实现：

你已经把所有要发的请求放进了List<Observable<ForkliftDetail>>里，接下来只需要用RxJava的操作符让这些请求并行执行，最后把所有结果打包成一个List<ForkliftDetail>就行。

假设你已经通过Retrofit拿到了API服务实例apiService，并且已经构建好了请求列表requests，直接用下面的代码就能搞定：

import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.schedulers.Schedulers;

// 并行执行所有请求，合并结果为List
Observable.merge(requests)
    .subscribeOn(Schedulers.io()) // 所有请求在IO线程并行执行
    .toList() // 等待所有请求完成，把结果打包成List
    .observeOn(AndroidSchedulers.mainThread()) // Android场景下切回主线程处理结果
    .subscribe(
        forkliftDetails -> {
            // 这里拿到最终合并好的List！
            // 比如更新UI、处理业务逻辑
            System.out.println("成功拿到" + forkliftDetails.size() + "条叉车详情");
        },
        throwable -> {
            // 处理错误：如果任何一个请求失败，会走到这里
            throwable.printStackTrace();
            System.out.println("请求出错了：" + throwable.getMessage());
        }
    );

• Observable.merge(requests)：把你列表里的所有Observable合并成一个流，所有请求会并行发射数据，不会互相等待
• toList()：这个操作符会耐心等所有请求都完成，然后把所有返回的ForkliftDetail打包成一个List发射出来
• subscribeOn(Schedulers.io())：确保请求在IO线程执行，不会阻塞主线程（Android里必须加，否则会ANR）
• observeOn(AndroidSchedulers.mainThread())：把最终结果的回调切回主线程，方便更新UI（后端项目可以去掉这一行）

上面的代码里，如果有一个请求失败，整个流程会直接触发onError，其他请求的结果也拿不到。如果你想忽略失败的请求，只收集成功的结果，可以给每个请求单独加错误处理：

// 构建请求列表的时候，给每个Observable加上错误兜底
List<Observable<ForkliftDetail>> requests = new ArrayList<>();
for (Forklift item : yourForkliftList) {
    Observable<ForkliftDetail> singleRequest = apiService.getForkliftDetail("zh-CN", item.getId())
        .onErrorResumeNext(Observable.empty()); // 请求失败就发射空流，相当于忽略这个失败的请求
    requests.add(singleRequest);
}

或者用onErrorReturn返回一个默认对象：

.onErrorReturn(throwable -> {
    // 打印错误日志，然后返回一个默认的ForkliftDetail
    System.err.println("单个叉车详情请求失败：" + throwable.getMessage());
    return new ForkliftDetail(); // 或者根据业务需求返回null（注意toList会包含null）
});

如果你更喜欢用fromIterable+flatMap的写法，也可以这么写：

Observable.fromIterable(requests)
    .flatMap(observable -> observable.subscribeOn(Schedulers.io())) // 每个请求并行执行
    .toList()
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(...);

和merge的写法效果完全一致，选你看着顺眼的就行~

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