RxJava 操做符系列二

時間  2019-12-05
標籤 rxjava 系列 欄目 Java 简体版
原文   原文鏈接

RxJava操做符源碼傳送門

在上篇文章RxJava操做符系列一咱們介紹的操做符幾乎都是建立被觀察者的操做符，那麼今天的這篇文章就介紹一下常常用到的轉換操做符。話很少說，開始上車。javascript

Map

該操做符是對原始Observable發射的每一項數據運用一個函數，而後返回一個發射這些結果的Observable。java

例如咱們有一個整形數組的數據，當大於5時輸出爲true，則代碼實現git

Integer[] integers = {0, 9, 6, 4, 8};
        Observable.from(integers).map(new Func1<Integer, Boolean>() {
            @Override
            public Boolean call(Integer integer) {
                Log.e(TAG, "call: "+integer);
                return (integer > 5);
            }
        }).subscribe(new Subscriber<Boolean>() {
            @Override
            public void onCompleted() {
                Log.e(TAG, "onCompleted: ");
            }

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

            @Override
            public void onNext(Boolean aBoolean) {
                Log.e(TAG, "onNext: "+aBoolean);
            }
        });複製代碼

日誌輸出信息github

call: 0
onNext: false
call: 9
onNext: true
call: 6
onNext: true
call: 4
onNext: false
call: 8
onNext: true
onCompleted:複製代碼

對於map,他能夠將將數據源變換爲你想要的類型，好比，你想獲取有一個Student對象（裏面age，name屬性）那麼咱們能夠經過map只獲取name。接下來。咱們再舉個例子，咱們根據一個圖片路徑獲取圖片並將圖片設置到ImageView,而後將ImageView加的咱們的佈局中。數組

String path = Environment.getExternalStorageDirectory()+ File.separator+"aaa.jpg";
        Observable.just(path)
                .subscribeOn(Schedulers.io())
                .map(new Func1<String, Bitmap>() {
                    @Override
                    public Bitmap call(String s) {
                        Bitmap bitmap = BitmapFactory.decodeFile(s);
                        Log.e(TAG, "call: Bitmap"+bitmap);
                        return bitmap;
                    }
                }).map(new Func1<Bitmap, ImageView>() {
            @Override
            public ImageView call(Bitmap bitmap) {
                Log.e(TAG, "call: ImageView");
                ImageView imageView = new ImageView(getActivity());
                LinearLayout.LayoutParams params = new LinearLayout.LayoutParams(LinearLayout.LayoutParams.WRAP_CONTENT, LinearLayout.LayoutParams.WRAP_CONTENT);
                imageView.setLayoutParams(params);
                imageView.setImageBitmap(bitmap);
                return imageView;
            }
        }).observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<ImageView>() {
                    @Override
                    public void onCompleted() {
                        Log.e(TAG, "onCompleted: ");
                    }
                    @Override
                    public void onError(Throwable e) {
                        Log.e(TAG, "onError: ");
                    }

                    @Override
                    public void onNext(ImageView imageView) {
                        Log.e(TAG, "onNext: ");
                        layout.addView(imageView);
                    }
                });複製代碼

Cast

該操做符就是作一些強制類型轉換操做的。例如，當咱們在頁面跳轉時數據對象每每是序列化的，當咱們在新的頁面收到數據後就要強制轉換爲咱們想要的類型。cast操做符也能夠實現這樣的功能。以下緩存

Observable.just(serializable).cast(FileInfo.class).subscribe(new Subscriber<FileInfo>() {
            @Override
            public void onCompleted() {
                Log.e(TAG, "onCompleted: " );
            }

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

            @Override
            public void onNext(FileInfo fileInfo) {
                Log.e(TAG, "onNext: "+fileInfo.toString());
                tv1.append("\n"+fileInfo.toString());
            }
        });複製代碼

不過在該操做符實際用途並無那麼的普遍，不多用到，固然這個操做符也能夠達到java 中instanceof相同的做用，用於類型檢查，當不是該類型就會執行onError()方法。併發

FlatMap

該操做符與map操做符的區別是它將一個發射數據的Observable變換爲多個Observables，而後將它們發射的數據合併後放進一個單獨的Observable.app

Integer[] integers = {1, 2, 3};
        Observable.from(integers).flatMap(new Func1<Integer, Observable<String>>() {
            @Override
            public Observable<String> call(final Integer integer) {
                return Observable.create(new Observable.OnSubscribe<String>() {
                    @Override
                    public void call(Subscriber<? super String> subscriber) {
                        Log.e(TAG, "call: FlatMap " + Thread.currentThread().getName());
                        try {
                            Thread.sleep(200);
                            subscriber.onNext(integer + 100 + " FlatMap");
                            subscriber.onCompleted();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                            subscriber.onError(e);
                        }
                    }
                }).subscribeOn(Schedulers.newThread());
            }
        }).observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onCompleted() {
                        Log.e(TAG, "onCompleted: FlatMap");
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.e(TAG, "onError: FlatMap");
                    }

                    @Override
                    public void onNext(String s) {
                        Log.e(TAG, "onNext: FlatMap " + s);
                    }
                });複製代碼

打印日誌信息ide

call: FlatMap RxNewThreadScheduler-2
call: FlatMap RxNewThreadScheduler-3
call: FlatMap RxNewThreadScheduler-4
onNext: FlatMap 101 FlatMap
onNext: FlatMap 102 FlatMap
onNext: FlatMap 103 FlatMap
onCompleted: FlatMap複製代碼

ConcatMap

該操做符是相似於最簡單版本的flatMap，可是它按次序鏈接而不是合併那些生成的Observables，而後產生本身的數據序列.將上述flatMap代碼更改以下函數

Integer[] integers = {1, 2, 3};
        Observable.from(integers).concatMap(new Func1<Integer, Observable<String>>() {
            @Override
            public Observable<String> call(final Integer integer) {
                return Observable.create(new Observable.OnSubscribe<String>() {
                    @Override
                    public void call(Subscriber<? super String> subscriber) {
                        Log.e(TAG, "call:2 ConcatMap " + Thread.currentThread().getName());
                        try {
                            Thread.sleep(200);
                            subscriber.onNext(integer + 100 + " ConcatMap");
                            subscriber.onCompleted();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                            subscriber.onError(e);
                        }
                    }
                }).subscribeOn(Schedulers.newThread());
            }
        }).observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onCompleted() {
                        Log.e(TAG, "onCompleted: ConcatMap");
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.e(TAG, "onError: ConcatMap");
                    }

                    @Override
                    public void onNext(String s) {
                        Log.e(TAG, "onNext: ConcatMap " +s);
                    }
                });複製代碼

輸出日誌信息

call:2 ConcatMap RxNewThreadScheduler-5
onNext: ConcatMap 101 ConcatMap 
call:2 ConcatMap RxNewThreadScheduler-6
onNext: ConcatMap 102 ConcatMap 
call:2 ConcatMap RxNewThreadScheduler-7
onNext: ConcatMap 103 ConcatMap 
onCompleted: ConcatMap複製代碼

經過該操做符和flatMap輸出的日誌信息，很容易看出flatMap並無保證數據源的順序性，可是ConcatMap操做符保證了數據源的順序性。在應用中，若是你對數據的順序性有要求的話，就須要使用ConcatMap。若沒有要求，兩者皆可以使用。

SwitchMap

當原始Observable發射一個新的數據（Observable）時，它將取消訂閱並中止監視產生執以前那個數據的Observable，只監視當前這一個.

Integer[] integers = {1, 2, 3};
 Observable.from(integers).switchMap(new Func1<Integer, Observable<String>>() {
            @Override
            public Observable<String> call(Integer integer) {
                Log.e(TAG, "call: SwitchMap" + Thread.currentThread().getName());
                //若是不經過subscribeOn(Schedulers.newThread())在在子線程模擬併發操做，全部數據源依然會所有輸出,也就是併發操做此操做符纔有做用
                //若在此經過Thread。sleep（）設置等待時間，則輸出信息會不同。至關於模擬併發程度
                return Observable.just((integer + 100) + "SwitchMap").subscribeOn(Schedulers.newThread());
            }
        }).observeOn(AndroidSchedulers.mainThread()).subscribe(new Subscriber<String>() {
            @Override
            public void onCompleted() {
                Log.e(TAG, "onCompleted: SwitchMap");
            }

            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "onError: SwitchMap");
            }

            @Override
            public void onNext(String s) {
                Log.e(TAG, "onNext: SwitchMap "+s);
            }
        });複製代碼

輸出日誌信息

call: SwitchMapmain
call: SwitchMapmain
call: SwitchMapmain
onNext: SwitchMap 106SwitchMap
onCompleted: SwitchMap複製代碼

當數據源較多時，並不必定是隻輸出最後一項數據，有可能輸出幾項數據，也多是所有。

GroupBy

看到這個詞你就應該想到了這個操做符的做用，就是你理解的含義，他將數據源按照你的約定進行分組。咱們經過groupBy實行將1到10的數據進行就劃分，代碼以下

Observable.range(1, 10).groupBy(new Func1<Integer, Boolean>() {
            @Override
            public Boolean call(Integer integer) {
                return integer % 2 == 0;
            }
        }).subscribe(new Subscriber<GroupedObservable<Boolean, Integer>>() {
            @Override
            public void onCompleted() {
                Log.e(TAG, "onCompleted:1 ");
            }

            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "onError:1 ");
            }

            @Override
            public void onNext(GroupedObservable<Boolean, Integer> booleanIntegerGroupedObservable) {
                booleanIntegerGroupedObservable.toList().subscribe(new Subscriber<List<Integer>>() {
                    @Override
                    public void onCompleted() {
                        Log.e(TAG, "onCompleted:2 " );
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.e(TAG, "onError:2 ");
                    }

                    @Override
                    public void onNext(List<Integer> integers) {
                        Log.e(TAG, "onNext:2 "+integers);
                    }
                });
            }
        });複製代碼

輸出日誌信息

onNext:2 [1, 3, 5, 7, 9]
onCompleted:2 
onNext:2 [2, 4, 6, 8, 10]
onCompleted:2 
onCompleted:1複製代碼

在上面代碼中booleanIntegerGroupedObservable變量有一個getKey（）方法,該方法返回的是分組的key，他的值就是groupBy方法call回調所用函數的值，在上面也就是integer % 2 == 0的值，及true和false。有幾個分組也是有此值決定的。

Scan

操做符對原始Observable發射的第一項數據應用一個函數，而後將那個函數的結果做爲本身的第一項數據發射。它將函數的結果同第二項數據一塊兒填充給這個函數來產生它本身的第二項數據。它持續進行這個過程來產生剩餘的數據序列。
例如計算1+2+3+4的和

Observable.range(1,4).scan(new Func2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer integer, Integer integer2) {
                Log.e(TAG, "call: integer:"+integer+" integer2 "+integer2);
                return integer+integer2;
            }
        }).subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
                Log.e(TAG, "onCompleted: ");
            }

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

            @Override
            public void onNext(Integer integer) {
                Log.e(TAG, "onNext: "+integer );
            }
        });複製代碼

輸出日誌信息

onNext: 1
call: integer:1  integer2 2
onNext: 3
call: integer:3  integer2 3
onNext: 6
call: integer:6  integer2 4
onNext: 10
onCompleted:複製代碼

對於scan有一個重載方法，能夠設置一個初始值，如上面代碼，初始值設置爲10，只需將scan加個參數scan（10，new Func2）。

Buffer

操做符將一個Observable變換爲另外一個，原來的Observable正常發射數據，變換產生的Observable發射這些數據的緩存集合，若是原來的Observable發射了一個onError通知，Buffer會當即傳遞這個通知，而不是首先發射緩存的數據，即便在這以前緩存中包含了原始Observable發射的數據。
示例代碼

Observable.range(10, 6).buffer(2).subscribe(new Subscriber<List<Integer>>() {
            @Override
            public void onCompleted() {
                Log.e(TAG, "onCompleted: ");
            }

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

            @Override
            public void onNext(List<Integer> integers) {
                Log.e(TAG, "onNext: " + integers);
            }
        });複製代碼

輸出日誌信息

onNext: [10, 11]
onNext: [12, 13]
onNext: [14, 15]
onCompleted:複製代碼

上面一次性訂閱兩個數據，若是設置參數爲6，就一次性訂閱。buffer的另外一重載方法buffer(count, skip)從原始Observable的第一項數據開始建立新的緩存（長度count），此後每當收到skip項數據，用count項數據填充緩存：開頭的一項和後續的count-1項，它以列表(List)的形式發射緩存，取決於count和skip的值，這些緩存可能會有重疊部分（好比skip < count時），也可能會有間隙（好比skip > count時）。具體執行結果，你能夠設置不一樣的skip和count觀察輸出日誌，查看執行結果及流程。

Window

Window和Buffer相似，但不是發射來自原始Observable的數據包，它發射的是Observables，這些Observables中的每個都發射原始Observable數據的一個子集，最後發射一個onCompleted通知。

Observable.range(10, 6).window(2).subscribe(new Subscriber<Observable<Integer>>() {
            @Override
            public void onCompleted() {
                Log.e(TAG, "onCompleted1: ");
            }

            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "onError1: ");
            }

            @Override
            public void onNext(Observable<Integer> integerObservable) {
                Log.e(TAG, "onNext1: ");
                tv1.append("\n");
                integerObservable.subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onCompleted() {
                        Log.e(TAG, "onCompleted2: ");
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.e(TAG, "onError2: ");
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.e(TAG, "onNext2: "+integer);
                    }
                });
            }
        });複製代碼

輸出日誌信息

onNext2: 10
onNext2: 11
onCompleted2: 
onNext2: 12
onNext2: 13
onCompleted2: 
onNext2: 14
onNext2: 15
onCompleted2: 
onCompleted1:複製代碼

window和buffer同樣也有不一樣的重載方法。這兩個操做符相對其餘操做符不太容易理解，能夠去RxJava GitHub理解，裏面有圖示解析。固然最好的理解方式就是經過更改變量的值，去觀察輸出的日誌信息。

好了，這篇文章就介紹到這裏。若文中有錯誤的地方，歡迎指正。謝謝。

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程