rxjava怎么组合数据流

在 RxJava 中，“组合数据流”通常指把多个 Observable / Flowable / Single 等数据源，按某种规则合并、转换或协调后，输出一个新的数据流。下面按常见场景给你系统讲一下。

一、合并型（多个流 → 一个流）

1. merge / mergeWith

多个流同时发射，按时间顺序合并（不保证顺序）

Observable.merge(
    observableA,
    observableB
).subscribe(System.out::println);

observableA
    .mergeWith(observableB)
    .subscribe(...);

✅ 适合：多个独立数据源同时监听

2. concat / concatWith

顺序执行，前一个结束再发下一个

Observable.concat(
    observableA,
    observableB
)

✅ 适合：有先后依赖（如：先读缓存，再请求网络）

3. concatEager

同时订阅，但按 concat 顺序发射

Observable.concatEager(a, b)

二、组合型（按“关系”组合）

4. zip

一一对应合并多个流

Observable.zip(
    observableA,
    observableB,
    (a, b) -> a + "-" + b
)

✅ 适合：

• 多个接口同时请求
• 等所有结果都到了再处理

5. combineLatest

任一流变化，用最新值组合

Observable.combineLatest(
    a,
    b,
    (x, y) -> x + y
)

✅ 适合：

• 表单多输入框
• 实时计算（如金额 + 汇率）

6. withLatestFrom

主流动时，取另一个流最新值

main.withLatestFrom(other, (m, o) -> m + o)

✅ 适合：

• 点击事件 + 当前状态

三、扁平化（流里套流）

7. flatMap

无序、并发

observable.flatMap(item -> request(item))

8. concatMap

有序、串行

observable.concatMap(item -> request(item))

9. switchMap

只保留最新流，取消旧流

editTextChanges.switchMap(query -> search(query))

✅ 适合：搜索联想

四、条件 / 选择型

10. amb

谁先发射用谁

Observable.ambArray(a, b)

11. publish + merge

一个源被多个订阅者复用

ConnectableObservable<Integer> pm = source.publish();
pm.mergeWith(pm.map(i -> i * 2)).subscribe();
pm.connect();

五、实战示例（常见组合）

缓存 + 网络

Observable.concat(
    getCache(),
    getNetwork()
)

多接口并发

Observable.zip(
    api1(),
    api2(),
    (r1, r2) -> combine(r1, r2)
)

实时表单

Observable.combineLatest(
    name, age,
    (n, a) -> n + a
)

六、选型速查表

如果你有具体业务场景（比如：多个接口、搜索、表单、缓存），我可以直接帮你写 RxJava 代码。