组合操作符

concat操作符

concat操作符是将多个Observable 按传入顺序进行输出,O_a输出完毕,O_b接着输出:

Observable<Integer> range = Observable.range(1, 5);  
      List<Integer> data = Arrays.asList(6, 7, 8, 9, 10);  
      Observable<Integer> from = Observable.from(data);  
  
      Observable.concat(from,range).subscribe(new Observer<Integer>() {  
          @Override  
          public void onCompleted() {  
              LogUtils.d("----->onCompleted");  
          }  
  
          @Override  
          public void onError(Throwable e) {  
              LogUtils.d("----->onError");  
          }  
  
          @Override  
          public void onNext(Integer o) {  
              LogUtils.d("----->onNext:" + o);  
          }  
      });

运行结果如下：
----->onNext:6
----->onNext:7
----->onNext:8
----->onNext:9
----->onNext:10
----->onNext:1
----->onNext:2
----->onNext:3
----->onNext:4
----->onNext:5
----->onCompleted

combineLastest操作符

combineLastest() 是将第一个Observable的【最新】数据与后面的Observable数据项按某种规则合并。

 Observable.combineLatest(range, from, new Func2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer integer, Integer integer2) {
                Log.d("rxandroidtest","----->i1:"+integer+"  i2:"+integer2);
                return integer + integer2;
            }
        }).subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
                Log.d("rxandroidtest","----->onCompleted");
            }

            @Override
            public void onError(Throwable e) {
                Log.d("rxandroidtest","----->onError");
            }

            @Override
            public void onNext(Integer o) {
                Log.d("rxandroidtest","----->onNext:" + o);
            }
        });

运行结果如下：
----->i1:5 i2:6
----->onNext:11
----->i1:5 i2:7
----->onNext:12
----->i1:5 i2:8
----->onNext:13
----->i1:5 i2:9
----->onNext:14
----->i1:5 i2:10
----->onNext:15
----->onCompleted

join操作符

在a的生命周期内：b输出的数据项与a输出的数据项每个合并,直到b输出下一项：
为什么要加延迟？是因为可能b的输出会因为a的生命周期结束了而b还未开始
rang 和 from 参考上面

        from.observeOn(Schedulers.newThread());
        range.observeOn(Schedulers.newThread());
        range.join(from, new Func1<Integer, Observable<Integer>>() {
            @Override
            public Observable<Integer> call(Integer integer) {
                Log.d("join","left:"+integer);
                return Observable.just(integer).delay(1, TimeUnit.SECONDS);
            }
        }, new Func1<Integer, Observable<Integer>>() {
            @Override
            public Observable<Integer> call(Integer integer) {
                Log.d("join","right:"+integer);
                return Observable.just(integer).delay(1, TimeUnit.SECONDS);
            }
        }, new Func2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer integer, Integer integer2) {
                Log.d("join","left:"+integer + " right:"+integer2);
                return integer + integer2;
            }

        }).subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
                Log.d("join","onCompleted！");
            }

            @Override
            public void onError(Throwable e) {
                Log.d("join","onError:"+e.getMessage());
            }

            @Override
            public void onNext(Integer o) {
                Log.d("join","onNext:"+o);
            }
        });

运行结果如下：
D/join: left:1
D/join: left:2
D/join: left:3
D/join: left:4
D/join: left:5
D/join: right:6
D/join: left:5 right:6
D/join: onNext:11
D/join: left:2 right:6
D/join: onNext:8
D/join: left:1 right:6
D/join: onNext:7
D/join: left:4 right:6
D/join: onNext:10
D/join: left:3 right:6
D/join: onNext:9
D/join: right:7
D/join: left:5 right:7
D/join: onNext:12
D/join: left:2 right:7
D/join: onNext:9
D/join: left:1 right:7
D/join: onNext:8
D/join: left:4 right:7
D/join: onNext:11
D/join: left:3 right:7
D/join: onNext:10
D/join: right:8
D/join: left:5 right:8
D/join: onNext:13
D/join: left:2 right:8
D/join: onNext:10
D/join: left:1 right:8
D/join: onNext:9
D/join: left:4 right:8
D/join: onNext:12
D/join: left:3 right:8
D/join: onNext:11
D/join: right:9
D/join: left:5 right:9
D/join: onNext:14
D/join: left:2 right:9
D/join: onNext:11
D/join: left:1 right:9
D/join: onNext:10
D/join: left:4 right:9
D/join: onNext:13
D/join: left:3 right:9
D/join: onNext:12
D/join: right:10
D/join: left:5 right:10
D/join: onNext:15
D/join: left:2 right:10
D/join: onNext:12
D/join: left:1 right:10
D/join: onNext:11
D/join: left:4 right:10
D/join: onNext:14
D/join: left:3 right:10
D/join: onNext:13
D/join: onCompleted！

merge 操作符

按照两个Observable提交结果的时间顺序进行输出

 Observable.merge(range, from).subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
                Log.d("join","onCompleted！");
            }

            @Override
            public void onError(Throwable e) {
                Log.d("join","onError:"+e.getMessage());
            }

            @Override
            public void onNext(Integer integer) {
                Log.d("join","onNext:"+integer);
            }
        });

运行结果如下：
onNext:1
D/join: onNext:2
D/join: onNext:3
D/join: onNext:4
D/join: onNext:5
D/join: onNext:6
D/join: onNext:7
D/join: onNext:8
D/join: onNext:9
D/join: onNext:10
D/join: onCompleted！

例子2

  Observable<Long> interval_1 = Observable.interval(3, TimeUnit.SECONDS).map(aLong -> aLong * 1).take(5);
        Observable<Long> interval_100 = Observable.interval(3, TimeUnit.SECONDS).map(aLong -> aLong * 100).take(5);
        Observable.merge(interval_1,interval_100).subscribe(new Subscriber<Long>() {
            @Override
            public void onCompleted() {
                Log.d("join","onCompleted！");
            }

            @Override
            public void onError(Throwable e) {
                Log.d("join","onError:"+e.getMessage());
            }

            @Override
            public void onNext(Long aLong) {
                Log.d("join","onNext:"+aLong);
            }
        });

运行结果如下：

D/join: onNext:0
D/join: onNext:0
D/join: onNext:1
D/join: onNext:100
D/join: onNext:2
D/join: onNext:200
D/join: onNext:3
D/join: onNext:300
D/join: onNext:4
D/join: onNext:400
D/join: onCompleted！

mergeDelayError操作符

在merge过程中一旦某一个Observable中出现错误,就会停止合并,mergeDelayError操作符会把错误放在所有结果都和并完成后才执行:

Observable<Long> interval_1 = Observable.interval(3, TimeUnit.SECONDS).map(aLong -> {
            if(aLong > 2) throw new RuntimeException();
            return aLong * 1;
        }).take(5);
        Observable<Long> interval_100 = Observable.interval(3, TimeUnit.SECONDS).map(aLong -> aLong * 100).take(5);
        Observable.mergeDelayError(interval_1,interval_100).subscribe(new Subscriber<Long>() {
            @Override
            public void onCompleted() {
                Log.d("join","onCompleted！");
            }

            @Override
            public void onError(Throwable e) {
                Log.d("join","onError:"+ e);
            }

            @Override
            public void onNext(Long aLong) {
                Log.d("join","onNext:"+aLong);
            }
        });

运行结果如下：
onNext:0
onNext:0
onNext:1
onNext:100
onNext:2
onNext:200
onNext:300
onNext:400
onError:java.lang.RuntimeException

startWith操作符

在源Observable输出之前插入指定数据项:

  range.startWith(8,9).subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
                Log.d("join","onCompleted！");
            }

            @Override
            public void onError(Throwable e) {
                Log.d("join","onError:"+ e);
            }

            @Override
            public void onNext(Integer integer) {
                Log.d("join","onNext:"+integer);
            }

        });

运行结果如下：
onNext:8
onNext:9
onNext:1
onNext:2
onNext:3
onNext:4
onNext:5
onCompleted！

switchOnNext操作符

把一组Observable转换成一个Observable, 这组Observable中的每一个Observable所产生的结果,如果在同一个时间段内两个或者多个Observable提交结果,只取最后一个Observable提交的结果

 Observable<Observable<Long>> longObservable = Observable.interval(10, TimeUnit.SECONDS).map(new Func1<Long, Observable<Long>>() {
            @Override
            public Observable<Long> call(final Long aLong1) {
                return Observable.interval(1, TimeUnit.SECONDS).map(new Func1<Long, Long>() {
                    @Override
                    public Long call(Long aLong) {
                    //    Log.d("join","------>aLong1:" + aLong);
                    //    Log.d("join","------>aLong:" + aLong);
                        return aLong + (aLong1*100);
                    }
                }).take(3);
            }
        }).take(3);

        Observable.switchOnNext(longObservable).subscribe(new Observer<Long>() {
            @Override
            public void onCompleted() {
                Log.d("join","------>onCompleted()");
            }

            @Override
            public void onError(Throwable e) {
                Log.d("join","------>onError()" + e);
            }

            @Override
            public void onNext(Long integer) {
                Log.d("join","------>onNext()" + integer);
            }
        });

运行结果如下：

onNext()0
onNext()1
onNext()2
onNext()100
onNext()101
onNext()102
onNext()200
onNext()201
onNext()202
onCompleted()
如果同一时间有多个Observable发射，只取最后一个

  Observable<Observable<Long>> longObservable = Observable.interval(3, TimeUnit.SECONDS).map(new Func1<Long, Observable<Long>>() {
            @Override
            public Observable<Long> call(final Long aLong1) {
                return Observable.interval(1, TimeUnit.SECONDS).map(new Func1<Long, Long>() {
                    @Override
                    public Long call(Long aLong) {
                        //    Log.d("join","------>aLong1:" + aLong);
                        //    Log.d("join","------>aLong:" + aLong);
                        return aLong + (aLong1*100);
                    }
                }).take(3);
            }
        }).take(3);

        Observable.switchOnNext(longObservable).subscribe(new Observer<Long>() {
            @Override
            public void onCompleted() {
                Log.d("join","------>onCompleted()");
            }

            @Override
            public void onError(Throwable e) {
                Log.d("join","------>onError()" + e);
            }

            @Override
            public void onNext(Long integer) {
                Log.d("join","------>onNext()" + integer);
            }
        });

运行结果如下：
少了2，102
onNext()0
onNext()1
onNext()100
onNext()101
onNext()200
onNext()201
onNext()202
onCompleted()

zip 操作符

是将同索引位置的数据项进行合并,放单的数据项丢弃

 Observable.zip(Observable.just(1, 2, 3), Observable.just(10, 20, 30), new Func2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer integer, Integer integer2) {
                return integer + integer2;
            }
        }).subscribe(
                new Observer<Integer>() {
                    @Override
                    public void onCompleted() {
                        Log.d("join","------>onCompleted()");
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.d("join","------>onError()" + e);
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.d("join","------>onNext()" + integer);
                    }
                }
        );

运行结果如下：

onNext()11
onNext()22
onNext()33
onCompleted()

最后编辑于：2017.12.04 15:22:32

人面猴
序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
沈念sama阅读 206,839评论 6赞 482
死咒
序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
沈念sama阅读 88,543评论 2赞 382
救了他两次的神仙让他今天三更去死
文/潘晓璐我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
开封第一讲书人阅读 153,116评论 0赞 344
道士缉凶录：失踪的卖姜人
文/不坏的土叔我叫张陵，是天一观的道长。经常有香客问我，道长，这世上最难降的妖魔是什么？我笑而不...
开封第一讲书人阅读 55,371评论 1赞 279
港岛之恋（遗憾婚礼）
正文为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
茶点故事阅读 64,384评论 5赞 374
恶毒庶女顶嫁案：这布局不是一般人想出来的
文/花漫我一把揭开白布。她就那样静静地躺着，像睡着了一般。火红的嫁衣衬着肌肤如雪。梳的纹丝不乱的头发上，一...
开封第一讲书人阅读 49,111评论 1赞 285
城市分裂传说
那天，我揣着相机与录音，去河边找鬼。笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。我是一名探鬼主播，决...
沈念sama阅读 38,416评论 3赞 400
双鸳鸯连环套：你想象不到人心有多黑
文/苍兰香墨我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
开封第一讲书人阅读 37,053评论 0赞 259
万荣杀人案实录
序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
沈念sama阅读 43,558评论 1赞 300
护林员之死
正文独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋以下内容为张勋视角年9月15日...
茶点故事阅读 36,007评论 2赞 325
白月光启示录
正文我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
茶点故事阅读 38,117评论 1赞 334
活死人
序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
沈念sama阅读 33,756评论 4赞 324
日本核电站爆炸内幕
正文年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
茶点故事阅读 39,324评论 3赞 307
男人毒药：我在死后第九天来索命
文/蒙蒙一、第九天我趴在偏房一处隐蔽的房顶上张望。院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
开封第一讲书人阅读 30,315评论 0赞 19
一桩弑父案，背后竟有这般阴谋
文/苍兰香墨我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。一阵脚步声响...
开封第一讲书人阅读 31,539评论 1赞 262
情欲美人皮
我被黑心中介骗来泰国打工，没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。一个月前我还...
沈念sama阅读 45,578评论 2赞 355
代替公主和亲
正文我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
茶点故事阅读 42,877评论 2赞 345

组合操作符

concat操作符

combineLastest操作符

join操作符

merge 操作符

mergeDelayError操作符

startWith操作符

switchOnNext操作符

zip 操作符

推荐阅读更多精彩内容