RxJava学习笔记操作符学习（4）

组合 / 合并操作符

作用

组合多个被观察者（Observable） & 合并需要发送的事件

常见类型

常见类型.png

.应用场景 & 对应操作符介绍

concat（） / concatArray（）
作用
组合多个被观察者一起发送数据，合并后按发送顺序串行执行。
二者区别：组合被观察者的数量，即concat（）组合被观察者数量≤4个，而concatArray（）则可＞4个。
事例：

 public void ConcatOperators(View view){
       //  concat（）：组合多个被观察者（≤4个）一起发送数据  concatArray（）：组合多个被观察者一起发送数据（可＞4个）
        // 注：串行执行
        Observable.concat(Observable.just("1","2","3","4")
                ,Observable.just("5","6","7","8")
                ,Observable.just("9","10","11","12")
                ,Observable.just("13","14","15"))
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(String s) {
                        Log.e("sss", "接收到了事件"+ s);
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.e("sss", "对Error事件作出响应");
                    }

                    @Override
                    public void onComplete() {
                        Log.e("sss", "对Complete事件作出响应");
                    }
                });
    }

2.merge（） / mergeArray（）
作用
组合多个被观察者一起发送数据，合并后按时间线并行执行。
二者区别：组合被观察者的数量，即merge（）组合被观察者数量≤4个，而mergeArray（）则可＞4个。
区别上述的concat（） / concatArray（）操作符：同样是组合多个被观察者一起发送数据，但concat（） / concatArray（）操作符合并后是按发送顺序串行执行。
事例

public void MergeOperators(View view){
        // merge（）：组合多个被观察者（＜4个）一起发送数据  mergeArray（） = 组合4个以上的被观察者一起发送数据
        // 注：合并后按照时间线并行执行
        Observable.merge(Observable.intervalRange(0,3,1,1, TimeUnit.SECONDS)
        ,Observable.intervalRange(6,3,1,1,TimeUnit.SECONDS))
                .subscribe(new Observer<Long>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(Long aLong) {
                        Log.e("sss", "接收到了事件"+ aLong);
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.e("sss", "对Error事件作出响应");
                    }

                    @Override
                    public void onComplete() {
                        Log.e("sss", "对Complete事件作出响应");
                    }
                });
    }

concatDelayError（） / mergeDelayError（）
作用

示意图.png

事例

 Observable.concatArrayDelayError(Observable.create(new ObservableOnSubscribe<Integer>() {
                    @Override
                    public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                        emitter.onNext(1);
                        emitter.onNext(2);
                        emitter.onNext(3);
                        emitter.onError(new Throwable());
                    }
                }),
                Observable.just(4,5,6))
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.e("sss", "接收到了事件"+ integer);
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.e("sss", "对Error事件作出响应");
                    }

                    @Override
                    public void onComplete() {
                        Log.e("sss", "对Complete事件作出响应");
                    }
                });
    }

4.Zip（）
作用
合并多个被观察者（Observable）发送的事件，生成一个新的事件序列（即组合过后的事件序列），并最终发送。
事件组合方式 = 严格按照原先事件序列进行对位合并；最终合并的事件数量 = 多个被观察者（Observable）中数量最少的数量。

image.png

事例

 public void ZipOperators(View view) {
        Observable<Integer> observable1 = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.e("sss", "被观察者1发送了事件1");
                emitter.onNext(1);
                Thread.sleep(1000);
                Log.e("sss", "被观察者1发送了事件2");
                emitter.onNext(2);
                Thread.sleep(1000);
                Log.e("sss", "被观察者1发送了事件3");
                emitter.onNext(3);
                Thread.sleep(1000);
                emitter.onComplete();
            }
        }).subscribeOn(Schedulers.io());
        Observable<String> observable2 = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                Log.e("sss", "被观察者2发送了事件1");
                emitter.onNext("A");
                Thread.sleep(3000);
                Log.e("sss", "被观察者2发送了事件2");
                emitter.onNext("B");
                Thread.sleep(3000);
                Log.e("sss", "被观察者2发送了事件3");
                emitter.onNext("C");
                Thread.sleep(3000);
                emitter.onComplete();
            }
        }).subscribeOn(Schedulers.io());

        Observable.zip(observable1, observable2, new BiFunction<Integer, String, String>() {

            @Override
            public String apply(Integer integer, String s) throws Exception {
                return integer + s;
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(String s) {
                Log.e("sss", "最终接收到的事件 =  " +  s);
            }

            @Override
            public void onError(Throwable e) {
                Log.e("sss", "对Error事件作出响应");
            }

            @Override
            public void onComplete() {
                Log.e("sss", "对Complete事件作出响应");
            }
        });
    }

5.combineLatest（）
作用
当两个Observables中的任何一个发送了数据后，将先发送了数据的Observables 的最新（最后）一个数据与另外一个Observable发送的每个数据结合，最终基于该函数的结果发送数据。
与Zip（）的区别：Zip（） = 按个数合并，即1对1合并；CombineLatest（） = 按时间合并，即在同一个时间点上合并。
事例

 public void combineLatestOperators(View view){
        
        Observable.combineLatest(Observable.just(1L, 2L, 3L),
                Observable.intervalRange(0, 3, 1, 1, TimeUnit.SECONDS),
                new BiFunction<Long, Long, Long>() {
                    @Override
                    public Long apply(Long aLong, Long aLong2) throws Exception {
                        // aLong = 第1个Observable发送的最新（最后）1个数据
                        // aLong2 = 第2个Observable发送的每1个数据
                        Log.e("sss", "合并的数据是： "+ aLong + " "+ aLong2);
                        return aLong+aLong2;
                    }
                }).subscribe(new Consumer<Long>() {
            @Override
            public void accept(Long aLong) throws Exception {
                Log.e("sss", "合并的结果是： "+aLong);
            }
        });
    }

6.combineLatestDelayError（）
作用
类似于concatDelayError（） / mergeDelayError（），即错误处理，此处不作过多描述。
7.reduce（）
作用
把被观察者需要发送的事件聚合成1个事件 & 发送，聚合的逻辑根据需求撰写，但本质都是前2个数据聚合，然后与后1个数据继续进行聚合，依次类推。
事例

public void reduceOperators(View view){
        Observable.just(1,2,3,4)
                .reduce(new BiFunction<Integer, Integer, Integer>() {
                    @Override
                    public Integer apply(Integer integer, Integer integer2) throws Exception {
                        return integer*integer2;
                    }
                }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.e("sss", "最终计算的结果是： "+integer);
            }
        });
    }

8.collect（）
作用
将被观察者Observable发送的数据事件收集到一个数据结构里。
事例

public void collectOperators(View view){
        Observable.just(1,2,3,4,5,6)
                .collect(new Callable<ArrayList<Integer>>() {
                    @Override
                    public ArrayList<Integer> call() throws Exception {
                         return new ArrayList<>();
                    }
                }, new BiConsumer<ArrayList<Integer>, Integer>() {
                    @Override
                    public void accept(ArrayList<Integer> integers, Integer integer) throws Exception {
                        integers.add(integer);
                    }
                }).subscribe(new Consumer<ArrayList<Integer>>() {
            @Override
            public void accept(ArrayList<Integer> integers) throws Exception {
                Log.e("sss","本次发送的数据是： "+integers);
            }
        });
    }

9.startWith（） / startWithArray（）
作用
在一个被观察者发送事件前，追加发送一些数据 / 一个新的被观察者。
事例

public void startOperators(View view){
        Observable.just(7,8,9)
                .startWith(6)
                .startWithArray(1,2,3)
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.e("sss", "接收到了事件"+ integer );
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.e("sss", "对Error事件作出响应");
                    }

                    @Override
                    public void onComplete() {
                        Log.e("sss", "对Complete事件作出响应");
                    }
                });
    }

count（）
作用
统计被观察者发送事件的数量。
事例

public void countOperators(View view){
        Observable.just(1,2,3,4)
                .count()
                .subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                      Log.e("sss","接受的数量为"+aLong);
                    }
                });
    }

11总结

i总结.png

实际开发中的应用

1.合并数据源展示
merge的使用

String result = "数据源来自 = " ;
    public void combineDataOperators(View view){


        Observable<String> observable1=Observable.just("网络");

        Observable<String> observable2=Observable.just("本地");

        Observable.merge(observable1,observable2)
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(String s) {
                        Log.e("sss", "数据源有： "+ s  );
                        result += s + "+";
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {
                        Log.e("sss", "获取数据完成");
                        Log.e("sss",  result  );
                    }
                });
    }

zip的使用

public void togetherOperators(View view){
        Retrofit retrofit=new Retrofit.Builder()
                .baseUrl("http://fy.iciba.com/")
                .addConverterFactory(GsonConverterFactory.create())
                .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
                .build();

        GetRequest_Interface getRequest_interface=retrofit.create(GetRequest_Interface.class);

        Observable<Translation> observable1=getRequest_interface.getCall_1().subscribeOn(Schedulers.io());
        Observable<Translation> observable2=getRequest_interface.getCall_2().subscribeOn(Schedulers.io());


        Observable.zip(observable1, observable2, new BiFunction<Translation, Translation, String>() {
            @Override
            public String apply(Translation translation, Translation translation2) throws Exception {
                return translation.getContent().getOut()+translation2.getContent().getOut();
            }
        }).observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        Log.e("sss","接受到的数据"+s);
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {

                    }
                });

    }

2.从磁盘 / 内存缓存中获取缓存数据

 public void CacheOperators(View view){
        final String memoryCache=null;
        final String diskCache="从磁盘中获取上数据";
        Observable<String> memory=Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                if (memoryCache!=null){
                    emitter.onNext(memoryCache);
                }else {
                    emitter.onComplete();
                }
            }
        });
        Observable<String> disk=Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                if (diskCache!=null){
                    emitter.onNext(diskCache);
                }else {
                    emitter.onComplete();
                }
            }
        });
        Observable<String> netWork=Observable.just("从网络获取");
        Observable.concat(memory,disk,netWork)
                // 通过firstElement()，从串联队列中取出并发送第1个有效事件（Next事件），即依次判断检查memory、disk、network
                // a. firstElement()取出第1个事件 = memory，即先判断内存缓存中有无数据缓存；由于memoryCache = null，即内存缓存中无数据，所以发送结束事件（视为无效事件）
                // b. firstElement()继续取出第2个事件 = disk，即判断磁盘缓存中有无数据缓存：由于diskCache ≠ null，即磁盘缓存中有数据，所以发送Next事件（有效事件）
                // c. 即firstElement()已发出第1个有效事件（disk事件），所以停止判断。
                .firstElement()
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        Log.e("sss","最终获取的数据来源 =  "+ s);
                    }
                });
        
    }

3.联合判断多个事件
与RxBinding联合使用


        final EditText name,pwd,verification ;
        final Button list;

        name=findViewById(R.id.name);
        pwd=findViewById(R.id.pwd);
        verification=findViewById(R.id.verification);

        list=findViewById(R.id.list);

        Observable<CharSequence> nameObservable= RxTextView.textChanges(name).skip(1);
        Observable<CharSequence> pwdObservable= RxTextView.textChanges(pwd).skip(1);
        Observable<CharSequence> verificationObservable= RxTextView.textChanges(verification).skip(1);

        Observable.combineLatest(nameObservable, pwdObservable, verificationObservable, new Function3<CharSequence, CharSequence, CharSequence, Boolean>() {
            @Override
            public Boolean apply(CharSequence charSequence, CharSequence charSequence2, CharSequence charSequence3) throws Exception {


                boolean isUserNameValid = !TextUtils.isEmpty(name.getText()) ;
                // 除了设置为空，也可设置长度限制
                // boolean isUserNameValid = !TextUtils.isEmpty(name.getText()) && (name.getText().toString().length() > 2 && name.getText().toString().length() < 9);

                boolean isUserPwdValid = !TextUtils.isEmpty(pwd.getText());
                boolean isUserVerValid = !TextUtils.isEmpty(verification.getText()) ;
                return isUserNameValid&isUserPwdValid&isUserVerValid;
            }
        }).subscribe(new Consumer<Boolean>() {
            @Override
            public void accept(Boolean aBoolean) throws Exception {
                list.setEnabled(aBoolean);
            }
        });

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

RxJava学习笔记操作符学习（4）

组合 / 合并操作符

作用

常见类型

.应用场景 & 对应操作符介绍

实际开发中的应用

推荐阅读更多精彩内容