如何在RxJava中根据发射项的索引将一个Observable/Flowable拆分为多个Observable/Flowable？

这个需求其实用RxJava的groupBy结合index()操作符就能轻松搞定，核心思路是先给每个发射项绑定索引，再根据索引对目标拆分数量取模的结果分组，最后提取每个分组即可。

核心实现步骤 #

• 给原始数据源的每个元素绑定从0开始的索引（用index()操作符）
• 用groupBy按照「索引 % 拆分数量」的结果作为分组key，把元素分配到对应的组
• 从分组中提取出每个单独的Observable/Flowable

Flowable示例代码 #

假设我们要把一个发射1-9的Flowable拆分成3个Flowable，分别对应第1、4、7项，第2、5、8项，第3、6、9项：

int splitCount = 3;

// 模拟原始数据源
Flowable<Integer> sourceFlowable = Flowable.range(1, 9);

// 给元素绑定索引并分组
Flowable<GroupedFlowable<Integer, Integer>> groupedFlowables = sourceFlowable
    .index() // 发射Pair<Long, Integer>，Long是从0开始的索引
    .groupBy(pair -> (int) (pair.getFirst() % splitCount));

// 提取每个分组
Flowable<Integer> group1 = groupedFlowables
    .filter(group -> group.getKey() == 0) // 对应索引0、3、6 → 原始第1、4、7项
    .flatMap(group -> group.map(Pair::getSecond));

Flowable<Integer> group2 = groupedFlowables
    .filter(group -> group.getKey() == 1) // 对应索引1、4、7 → 原始第2、5、8项
    .flatMap(group -> group.map(Pair::getSecond));

Flowable<Integer> group3 = groupedFlowables
    .filter(group -> group.getKey() == 2) // 对应索引2、5、8 → 原始第3、6、9项
    .flatMap(group -> group.map(Pair::getSecond));

// 测试订阅
group1.subscribe(i -> System.out.println("Group 1: " + i));
group2.subscribe(i -> System.out.println("Group 2: " + i));
group3.subscribe(i -> System.out.println("Group 3: " + i));

Observable版本 #

如果用的是Observable，代码逻辑完全一致，只需要替换对应的类即可：

int splitCount = 3;

Observable<Integer> sourceObservable = Observable.range(1, 9);

Observable<GroupedObservable<Integer, Integer>> groupedObservables = sourceObservable
    .index()
    .groupBy(pair -> (int) (pair.getFirst() % splitCount));

Observable<Integer> group1 = groupedObservables
    .filter(group -> group.getKey() == 0)
    .flatMap(group -> group.map(Pair::getSecond));
// group2、group3的写法和Flowable版本一致

灵活生成所有分组 #

如果拆分数量不固定，或者想批量生成分组，可以用循环把所有分组存入List：

List<Flowable<Integer>> flowableGroups = new ArrayList<>();
for (int i = 0; i < splitCount; i++) {
    final int targetKey = i;
    flowableGroups.add(
        groupedFlowables
            .filter(group -> group.getKey() == targetKey)
            .flatMap(group -> group.map(Pair::getSecond))
    );
}
// flowableGroups.get(0)就是第一个分组，以此类推

注意事项 #

• 索引计数：index()返回的索引是从0开始的，如果你的业务习惯从1开始计数，可以把分组key改成(pair.getFirst() + 1) % splitCount，再调整filter的条件即可。
• 无限流支持：这种方案对无限数据源同样有效，groupBy会实时分组，不会阻塞数据流。
• 背压处理：如果用Flowable，groupBy会自动处理背压，但要确保下游订阅者能处理各自的流量，避免背压异常。

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