RxJava2：如何等待列表中所有Flowable执行完成？

这个需求在RxJava里其实很好实现，核心思路是先把每个Flowable<Model>转换成能输出自身完整结果列表的Single<List<Model>>，再将这些Single合并，等待全部执行完成后收集所有结果。下面是具体的实现方案和示例：

核心实现代码 #

你想要的blockUntilAllFlowablesAreFinished逻辑可以通过RxJava的操作符组合出来，直接替换你的代码即可：

List<Flowable<Model>> flowableList = getFlowableList();

// 实现等待所有Flowable完成并收集结果
List<List<Model>> results = Flowable.fromIterable(flowableList)
    // 选择1：保持原flowableList的顺序执行，前一个完成后才处理下一个
    .concatMapSingle(flowable -> flowable.toList())
    // 选择2：并发执行所有Flowable，效率更高但结果顺序可能和原列表不一致
    // .flatMapSingle(flowable -> flowable.toList())
    .toList()
    .blockingGet();

// 后续操作：使用results处理数据
results.forEach((list, idx) -> {
    System.out.println("第" + (idx+1) + "个Flowable的结果：");
    list.forEach(model -> System.out.println("- " + model.toString()));
});

代码解释 #

• flowable.toList()：把每个Flowable<Model>转换成Single<List<Model>>，这个操作符会收集当前Flowable发射的所有元素，直到Flowable完成，然后一次性发射包含所有元素的列表。
• concatMapSingle vs flatMapSingle：
  • concatMapSingle：严格按照原flowableList的顺序执行每个Flowable，前一个执行完成后才会启动下一个，最终结果列表的顺序和原列表完全一致。
  • flatMapSingle：会同时启动多个Flowable并发执行，执行效率更高，但最终结果列表的顺序不一定和原列表相同（取决于每个Flowable的执行速度）。
• toList()：收集所有Single的结果，转换成Single<List<List<Model>>>，当所有Single都完成后，发射最终的结果列表。
• blockingGet()：阻塞当前线程，直到所有Flowable都执行完成，获取最终的结果列表。注意：这个方法不能在Android主线程使用，会导致ANR，仅适合后台线程场景。

错误处理（可选） #

如果担心某个Flowable执行出错导致整个流程中断，可以给每个Flowable添加错误处理，比如出错时返回空列表：

List<List<Model>> results = Flowable.fromIterable(flowableList)
    .concatMapSingle(flowable -> flowable
        .toList()
        // 出错时返回空列表，避免整个流程失败
        .onErrorReturnItem(Collections.emptyList()))
    .toList()
    .blockingGet();

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