RxJava2如何按条件拆分单流为双流后合并执行？

架构老林

2026-5-15

RxJava2 数据流拆分与依赖处理方案

作为RxJava2新手，我最近在处理带依赖的数据流执行时遇到了瓶颈：我通过Retrofit获取到Single<MyResponse>，其中包含Procedure列表，经过.map(MyResponse::getProcedures).flatMap(Observable::fromIterable).flatMap(Procedure::execute)处理后，普通Procedure能正常执行，但部分Procedure有依赖关系——比如文件上传必须在WiFi禁用前完成，否则会失败。

我一开始尝试用RxJS Marbles找合适的操作符，但大多是1或2流转1流的，找不到直接1流转2流的操作符，于是想到两个思路：

思路1：对Procedure集合排序后执行

这是最直接的方案，把有依赖优先级的Procedure提前，确保它们先执行。不过这种方式只适合简单的优先级标记，处理复杂的依赖关联会力不从心。

思路2：用`publish`操作符实现拆分后合并执行

后来在@Lukasz的建议下，我用publish操作符实现了类似"拆分后按顺序合并"的逻辑，先处理优先级高的流，再处理普通流。我做了一个简化测试：

Observable.just(1, 2, 3, 4, 5, 6)
    .publish(f -> Observable.concat(
        f.filter(this::isEven).map(this::printEven),
        f.filter(this::isOdd).map(this::printOdd)
    )).subscribe();

执行后输出：2 is even, 4 is even, 6 is even, 1 is odd, 3 is odd, 5 is odd，完全符合预期——先执行所有偶数（模拟优先级操作），再执行奇数（模拟普通操作）。

不过这个方案的局限性也很明显：仅靠简单的条件过滤拆分，没法处理Procedure之间复杂的依赖关联（比如A依赖B完成，B又依赖C这种链式依赖）。