一、描述

将一个发射多个Observables的Observable转换成另一个单独的Observable，后者发射那些Observables最近发射的数据项。

Switch订阅一个发射多个Observables的Observable。它每次观察那些Observables中的一个，Switch返回的这个Observable取消订阅前一个发射数据的Observable，开始发射最近的Observable发射的数据。注意：当原始Observable发射了一个新的Observable时（不是这个新的Observable发射了一条数据时），它将取消订阅之前的那个Observable。这意味着，在后来那个Observable产生之后到它开始发射数据之前的这段时间里，前一个Observable发射的数据将被丢弃（就像图例上的那个黄色圆圈一样）。

switch操作符在RxJava上的实现为switchOnNext,用来将一个发射多个小Observable的源Observable转化为一个Observable，然后发射这多个小Observable所发射的数据。

需要注意的就是，如果一个小的Observable正在发射数据的时候，源Observable又发射出一个新的小Observable，则前一个Observable发射的数据会被抛弃，直接发射新
的小Observable所发射的数据。可以看示意图中的黄色圆圈就被丢弃了。

二、示意图

switch

三、示例代码

private Observable<String> createObserver(final int index) {
    return Observable.create(new Observable.OnSubscribe<String>() {
        @Override
        public void call(Subscriber<? super String> subscriber) {
            for (int i = 1; i < 5; i++) {
                subscriber.onNext(index + "-" + i);
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }).subscribeOn(Schedulers.newThread());
}

@Override
protected void runSampleCode() {
    Observable.switchOnNext(Observable.create(
            new Observable.OnSubscribe<Observable<String>>() {
                @Override
                public void call(Subscriber<? super Observable<String>> subscriber) {
                    for (int i = 1; i < 3; i++) {
                        subscriber.onNext(createObserver(i));
                        try {
                            Thread.sleep(2000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }
    ))
            .subscribe(new Subscriber<String>() {
                @Override
                public void onCompleted() {
                    System.out.println("onCompleted.");
                }

                @Override
                public void onError(Throwable e) {
                    System.out.println("onError: " + e.getMessage());
                }

                @Override
                public void onNext(String s) {
                    System.out.println("onNext: " + s);
                }
            });
}

四、运行结果

onNext: 1-1
onNext: 1-2
onNext: 2-1
onNext: 2-2
onNext: 2-3
onNext: 2-4

可以看到使用siwtch的时候，第一个小Observable只发射出了两个数据，第二个小Observable就被源Observable发射出来了，所以其接下来的两个数据被丢弃。

项目代码已上传到Github：https://github.com/SherlockShi/RxJavaBestPractise

五、参考资料

ReactiveX官方文档

ReactiveX文档中文翻译

PS：欢迎关注 SherlockShi 个人博客