RxJava學習筆記操作符學習（4）

組合 / 合并操作符
  • 作用

組合 多個被觀察者（Observable） & 合并需要發(fā)送的事件

  • 常見類型
常見類型.png
  • .應用場景 & 對應操作符介紹
  1. concat（） / concatArray（）
    作用
    組合多個被觀察者一起發(fā)送數(shù)據(jù)，合并后按發(fā)送順序串行執(zhí)行。
    二者區(qū)別：組合被觀察者的數(shù)量，即concat（）組合被觀察者數(shù)量≤4個，而concatArray（）則可＞4個。
    事例：
 public void ConcatOperators(View view){
       //  concat（）：組合多個被觀察者（≤4個）一起發(fā)送數(shù)據(jù)  concatArray（）：組合多個被觀察者一起發(fā)送數(shù)據(jù)（可＞4個）
        // 注：串行執(zhí)行
        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（）
作用
組合多個被觀察者一起發(fā)送數(shù)據(jù)，合并后按時間線并行執(zhí)行。
二者區(qū)別：組合被觀察者的數(shù)量，即merge（）組合被觀察者數(shù)量≤4個，而mergeArray（）則可＞4個。
區(qū)別上述的concat（） / concatArray（）操作符：同樣是組合多個被觀察者一起發(fā)送數(shù)據(jù)，但concat（） / concatArray（）操作符合并后是按發(fā)送順序串行執(zhí)行。
事例

public void MergeOperators(View view){
        // merge（）：組合多個被觀察者（＜4個）一起發(fā)送數(shù)據(jù)  mergeArray（） = 組合4個以上的被觀察者一起發(fā)送數(shù)據(jù)
        // 注：合并后按照時間線并行執(zhí)行
        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事件作出響應");
                    }
                });
    }

  1. 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）發(fā)送的事件，生成一個新的事件序列（即組合過后的事件序列），并最終發(fā)送。
事件組合方式 = 嚴格按照原先事件序列進行對位合并；最終合并的事件數(shù)量 = 多個被觀察者（Observable）中數(shù)量最少的數(shù)量。


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發(fā)送了事件1");
                emitter.onNext(1);
                Thread.sleep(1000);
                Log.e("sss", "被觀察者1發(fā)送了事件2");
                emitter.onNext(2);
                Thread.sleep(1000);
                Log.e("sss", "被觀察者1發(fā)送了事件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發(fā)送了事件1");
                emitter.onNext("A");
                Thread.sleep(3000);
                Log.e("sss", "被觀察者2發(fā)送了事件2");
                emitter.onNext("B");
                Thread.sleep(3000);
                Log.e("sss", "被觀察者2發(fā)送了事件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中的任何一個發(fā)送了數(shù)據(jù)后，將先發(fā)送了數(shù)據(jù)的Observables 的最新（最后）一個數(shù)據(jù) 與 另外一個Observable發(fā)送的每個數(shù)據(jù)結合，最終基于該函數(shù)的結果發(fā)送數(shù)據(jù)。
與Zip（）的區(qū)別：Zip（） = 按個數(shù)合并，即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發(fā)送的最新（最后）1個數(shù)據(jù)
                        // aLong2 = 第2個Observable發(fā)送的每1個數(shù)據(jù)
                        Log.e("sss", "合并的數(shù)據(jù)是： "+ 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（）
作用
把被觀察者需要發(fā)送的事件聚合成1個事件 & 發(fā)送，聚合的邏輯根據(jù)需求撰寫，但本質(zhì)都是前2個數(shù)據(jù)聚合，然后與后1個數(shù)據(jù)繼續(xù)進行聚合，依次類推。
事例

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發(fā)送的數(shù)據(jù)事件收集到一個數(shù)據(jù)結構里。
事例

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","本次發(fā)送的數(shù)據(jù)是： "+integers);
            }
        });
    }

9.startWith（） / startWithArray（）
作用
在一個被觀察者發(fā)送事件前，追加發(fā)送一些數(shù)據(jù) / 一個新的被觀察者。
事例

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事件作出響應");
                    }
                });
    }
  1. count（）
    作用
    統(tǒng)計被觀察者發(fā)送事件的數(shù)量。
    事例
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","接受的數(shù)量為"+aLong);
                    }
                });
    }

11總結


i總結.png
  • 實際開發(fā)中的應用

1.合并數(shù)據(jù)源展示
merge的使用

String result = "數(shù)據(jù)源來自 = " ;
    public void combineDataOperators(View view){


        Observable<String> observable1=Observable.just("網(wǎng)絡");

        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", "數(shù)據(jù)源有： "+ s  );
                        result += s + "+";
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {
                        Log.e("sss", "獲取數(shù)據(jù)完成");
                        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","接受到的數(shù)據(jù)"+s);
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {

                    }
                });

    }

2.從磁盤 / 內(nèi)存緩存中 獲取緩存數(shù)據(jù)

 public void CacheOperators(View view){
        final String memoryCache=null;
        final String diskCache="從磁盤中獲取上數(shù)據(jù)";
        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("從網(wǎng)絡獲取");
        Observable.concat(memory,disk,netWork)
                // 通過firstElement()，從串聯(lián)隊列中取出并發(fā)送第1個有效事件（Next事件），即依次判斷檢查memory、disk、network
                // a. firstElement()取出第1個事件 = memory，即先判斷內(nèi)存緩存中有無數(shù)據(jù)緩存；由于memoryCache = null，即內(nèi)存緩存中無數(shù)據(jù)，所以發(fā)送結束事件（視為無效事件）
                // b. firstElement()繼續(xù)取出第2個事件 = disk，即判斷磁盤緩存中有無數(shù)據(jù)緩存：由于diskCache ≠ null，即磁盤緩存中有數(shù)據(jù)，所以發(fā)送Next事件（有效事件）
                // c. 即firstElement()已發(fā)出第1個有效事件（disk事件），所以停止判斷。
                .firstElement()
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        Log.e("sss","最終獲取的數(shù)據(jù)來源 =  "+ s);
                    }
                });
        
    }

3.聯(lián)合判斷多個事件
與RxBinding聯(lián)合使用


        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);
            }
        });
?著作權歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務。

相關閱讀更多精彩內(nèi)容

  • 本篇文章介主要紹RxJava中操作符是以函數(shù)作為基本單位，與響應式編程作為結合使用的，對什么是操作、操作符都有哪些...
    嘎啦果安卓獸閱讀 2,975評論 0贊 10
  • 一、RxJava操作符概述 RxJava中的操作符就是為了提供函數(shù)式的特性，函數(shù)式最大的好處就是使得數(shù)據(jù)處理簡潔易...
    無求_95dd閱讀 3,487評論 0贊 21
  • 一、RxJava操作符概述 RxJava中的操作符就是為了提供函數(shù)式的特性，函數(shù)式最大的好處就是使得數(shù)據(jù)處理簡潔易...
    測天測地測空氣閱讀 683評論 0贊 1
  • RxJava操作符圖譜 創(chuàng)建操作符 create 完整創(chuàng)建1個被觀察者對象（Observable） just 快速...
    yswheye閱讀 9,694評論 1贊 15
  • 注：只包含標準包中的操作符，用于個人學習及備忘參考博客：http://blog.csdn.net/maplejaw...
    小白要超神閱讀 2,366評論 2贊 8

友情鏈接更多精彩內(nèi)容