rx-java - 如果第二个 observable 开始发出项目，rxjava switch observable

Question

我有一些并行执行的可观察对象集，例如 localObservable 和 networkObservable。如果 networkObservable 开始发射项目(从这个时候开始，我只需要这些项目)，然后丢弃 localObservable 发射的项目(也许 localObservable 还没有尚未开始)。

Observable<Integer> localObservable =
            Observable.defer(() -> Observable.range(1, 10)).subscribeOn(Schedulers.io());
Observable<Integer> networkObservable =
            Observable.defer(() -> Observable.range(11, 20)).subscribeOn(Schedulers.io());

Answer 1

你可以这样做:

 Observable<Long> networkObservable =
            Observable.interval(1000, 500, TimeUnit.MILLISECONDS)
                    .subscribeOn(Schedulers.io())
                    .share();
    Observable<Long> localObservable =
            Observable.interval(500, TimeUnit.MILLISECONDS)                       
                    .subscribeOn(Schedulers.io())
                    .takeUntil(networkObservable);

    Observable.merge(networkObservable, localObservable)
            .subscribe(System.out::println);

这将输出:

0 // localObservable 
1 // localObservable 
0 // networkObservable from here on
1
2
...

takeUntil 将使 localObservable 在 networkObservable 的第一次发射发生时停止并取消订阅，因此合并的 Observable 只要 networkObservable 没有启动，就会从 localObservable 发射，当它启动时，它将停止从 localObservable 发射，并切换到仅发射来自 networkObservable。