2. 程式人生 > >RxJava的操作符3

RxJava的操作符3

阿新 • • 發佈：2019-01-05

本篇文章繼續介紹以下型別的操作符

  • Combining Observables(Observable的組合操作符)
  • Error Handling Operators(Observable的錯誤處理操作符)

Combining Observables(Observable的組合操作符)

combineLatest操作符

combineLatest操作符把兩個Observable產生的結果進行合併，合併的結果組成一個新的Observable。這兩個Observable中任意一個Observable產生的結果，都和另一個Observable最後產生的結果，按照一定的規則進行合併。流程圖如下：
這裡寫圖片描述
呼叫例子如下:

//產生0,5,10,15,20數列
        Observable<Long> observable1 = Observable.timer(0, 1000, TimeUnit.MILLISECONDS)
                .map(new Func1<Long, Long>() {
                    @Override
                    public Long call(Long aLong) {
                        return aLong * 5;
                    }
                }).take(5
 
);

        //產生0,10,20,30,40數列
        Observable<Long> observable2 = Observable.timer(500, 1000, TimeUnit.MILLISECONDS)
                .map(new Func1<Long, Long>() {
                    @Override
                    public Long call(Long aLong) {
                        return aLong * 10;
                    }
                }).take(5
 
);


        Observable.combineLatest(observable1, observable2, new Func2<Long, Long, Long>() {
            @Override
            public Long call(Long aLong, Long aLong2) {
                return aLong+aLong2;
            }
        }).subscribe(new Subscriber<Long>() {
            @Override
            public void onCompleted() {
                System.out.println("Sequence complete.");
            }

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

            @Override
            public void onNext(Long aLong) {
                System.out.println("Next: " + aLong);
            }
        });
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
save_snippets.png
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40

執行結果如下：
Next: 0
Next: 5
Next: 15
Next: 20
Next: 30
Next: 35
Next: 45
Next: 50
Next: 60
Sequence complete.

join操作符

join操作符把類似於combineLatest操作符，也是兩個Observable產生的結果進行合併，合併的結果組成一個新的Observable，但是join操作符可以控制每個Observable產生結果的生命週期，在每個結果的生命週期內，可以與另一個Observable產生的結果按照一定的規則進行合併，流程圖如下：
這裡寫圖片描述

join方法的用法如下：
observableA.join(observableB,
observableA產生結果生命週期控制函式，
observableB產生結果生命週期控制函式，
observableA產生的結果與observableB產生的結果的合併規則）

呼叫例子如下：

//產生0,5,10,15,20數列
        Observable<Long> observable1 = Observable.timer(0, 1000, TimeUnit.MILLISECONDS)
                .map(new Func1<Long, Long>() {
                    @Override
                    public Long call(Long aLong) {
                        return aLong * 5;
                    }
                }).take(5);

        //產生0,10,20,30,40數列
        Observable<Long> observable2 = Observable.timer(500, 1000, TimeUnit.MILLISECONDS)
                .map(new Func1<Long, Long>() {
                    @Override
                    public Long call(Long aLong) {
                        return aLong * 10;
                    }
                }).take(5);

        observable1.join(observable2, new Func1<Long, Observable<Long>>() {
            @Override
            public Observable<Long> call(Long aLong) {
                //使Observable延遲600毫秒執行
                return Observable.just(aLong).delay(600, TimeUnit.MILLISECONDS);
            }
        }, new Func1<Long, Observable<Long>>() {
            @Override
            public Observable<Long> call(Long aLong) {
                //使Observable延遲600毫秒執行
                return Observable.just(aLong).delay(600, TimeUnit.MILLISECONDS);
            }
        }, new Func2<Long, Long, Long>() {
            @Override
            public Long call(Long aLong, Long aLong2) {
                return aLong + aLong2;
            }
        }).subscribe(new Subscriber<Long>() {
            @Override
            public void onCompleted() {
                System.out.println("Sequence complete.");
            }

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

            @Override
            public void onNext(Long aLong) {
                System.out.println("Next: " + aLong);
            }
        });
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
save_snippets.png
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51

執行結果如下：
Next: 0
Next: 5
Next: 15
Next: 20
Next: 30
Next: 35
Next: 45
Next: 50
Next: 60
Sequence complete.

groupJoin操作符

groupJoin操作符非常類似於join操作符，區別在於join操作符中第四個引數的傳入函式不一致，其流程圖如下：
這裡寫圖片描述

呼叫例子如下：

//產生0,5,10,15,20數列
        Observable<Long> observable1 = Observable.timer(0, 1000, TimeUnit.MILLISECONDS)
                .map(new Func1<Long, Long>() {
                    @Override
                    public Long call(Long aLong) {
                        return aLong * 5;
                    }
                }).take(5);

        //產生0,10,20,30,40數列
        Observable<Long> observable2 = Observable.timer(500, 1000, TimeUnit.MILLISECONDS)
                .map(new Func1<Long, Long>() {
                    @Override
                    public Long call(Long aLong) {
                        return aLong * 10;
                    }
                }).take(5);

        observable1.groupJoin(observable2, new Func1<Long, Observable<Long>>() {
            @Override
            public Observable<Long> call(Long aLong) {
                return Observable.just(aLong).delay(1600, TimeUnit.MILLISECONDS);
            }
        }, new Func1<Long, Observable<Long>>() {
            @Override
            public Observable<Long> call(Long aLong) {
                return Observable.just(aLong).delay(600, TimeUnit.MILLISECONDS);
            }
        }, new Func2<Long, Observable<Long>, Observable<Long>>() {
            @Override
            public Observable<Long> call(Long aLong, Observable<Long> observable) {
                return observable.map(new Func1<Long, Long>() {
                    @Override
                    public Long call(Long aLong2) {
                        return aLong + aLong2;
                    }
                });
            }
        }).subscribe(new Subscriber<Observable<Long>>() {
            @Override
            public void onCompleted() {
                System.out.println("Sequence complete.");
            }

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

            @Override
            public void onNext(Observable<Long> observable) {
                observable.subscribe(new Subscriber<Long>() {
                    @Override
                    public void onCompleted() {

                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onNext(Long aLong) {
                        System.out.println("Next: " + aLong);
                    }
                });
            }
        });
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
save_snippets.png
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69

執行結果如下：
Next: 0
Next: 5
Next: 10
Next: 15
Next: 20
Next: 25
Next: 30
Next: 35
Next: 40
Next: 45
Next: 50
Next: 60
Next: 55
Sequence complete.

merge操作符

merge操作符是按照兩個Observable提交結果的時間順序，對Observable進行合併，如ObservableA每隔500毫秒產生資料為0,5,10,15,20；而ObservableB每隔500毫秒產生資料0,10,20,30,40，其中第一個資料延遲500毫秒產生，最後合併結果為：0,0,5,10,10,20,15,30,20,40;其流程圖如下：
這裡寫圖片描述

呼叫例子如下：

//產生0,5,10,15,20數列
        Observable<Long> observable1 = Observable.timer(0, 1000, TimeUnit.MILLISECONDS)
                .map(new Func1<Long, Long>() {
                    @Override
                    public Long call(Long aLong) {
                        return aLong * 5;
                    }
                }).take(5);

        //產生0,10,20,30,40數列
        Observable<Long> observable2 = Observable.timer(500, 1000, TimeUnit.MILLISECONDS)
                .map(new Func1<Long, Long>() {
                    @Override
                    public Long call(Long aLong) {
                        return aLong * 10;
                    }
                }).take(5);

        Observable.merge(observable1, observable2)
                .subscribe(new Subscriber<Long>() {
                    @Override
                    public void onCompleted() {
                        System.out.println("Sequence complete.");
                    }

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

                    @Override
                    public void onNext(Long aLong) {
                        System.out.println("Next:" + aLong);
                    }
                });
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
save_snippets.png
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35

執行結果如下：
Next:0
Next:0
Next:5
Next:10
Next:10
Next:20
Next:15
Next:30
Next:20
Next:40
Sequence complete.

mergeDelayError操作符

從merge操作符的流程圖可以看出，一旦合併的某一個Observable中出現錯誤，就會馬上停止合併，並對訂閱者回調執行onError方法，而mergeDelayError操作符會把錯誤放到所有結果都合併完成之後才執行，其流程圖如下：
這裡寫圖片描述

呼叫例子如下：

//產生0,5,10數列,最後會產生一個錯誤
        Observable<Long> errorObservable = Observable.error(new Exception("this is end!"));
        Observable < Long > observable1 = Observable.timer(0, 1000, TimeUnit.MILLISECONDS)
                .map(new Func1<Long, Long>() {
                    @Override
                    public Long call(Long aLong) {
                        return aLong * 5;
                    }
                }).take(3).mergeWith(errorObservable.delay(3500, TimeUnit.MILLISECONDS));

        //產生0,10,20,30,40數列
        Observable<Long> observable2 = Observable.timer(500, 1000, TimeUnit.MILLISECONDS)
                .map(new Func1<Long, Long>() {
                    @Override
                    public Long call(Long aLong) {
                        return aLong * 10;
                    }
                }).take(5);

        Observable.mergeDelayError(observable1, observable2)
                .subscribe(new Subscriber<Long>() {
                    @Override
                    public void onCompleted() {
                        System.out.println("Sequence complete.");
                    }

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

                    @Override
                    public void onNext(Long aLong) {
                        System.out.println("Next:" + aLong);
                    }
                });
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
save_snippets.png
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36

執行結果如下：
Next:0
Next:0
Next:5
Next:10
Next:10
Next:20
Next:30
Next:40
Error: this is end!

startWith操作符

startWith操作符是在源Observable提交結果之前，插入指定的某些資料，其流程圖如下：
這裡寫圖片描述

呼叫例子如下：

Observable.just(10,20,30).startWith(2, 3, 4).subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
                System.out.println("Sequence complete.");
            }

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

            @Override
            public void onNext(Integer value) {
                System.out.println("Next:" + value);
            }
        });
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
save_snippets.png
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16

執行結果如下：
Next:2
Next:3
Next:4
Next:10
Next:20
Next:30
Sequence complete.

switchOnNext操作符

switchOnNext操作符是把一組Observable轉換成一個Observable，轉換規則為：對於這組Observable中的每一個Observable所產生的結果，如果在同一個時間記憶體在兩個或多個Observable提交的結果，只取最後一個Observable提交的結果給訂閱者，其流程圖如下：
這裡寫圖片描述

呼叫例子如下：

//每隔500毫秒產生一個observable
        Observable<Observable<Long>> observable = Observable.timer(0, 500, TimeUnit.MILLISECONDS).map(new Func1<Long, Observable<Long>>() {
            @Override
            public Observable<Long> call(Long aLong) {
                //每隔200毫秒產生一組資料（0,10,20,30,40)
                return Observable.timer(0, 200, TimeUnit.MILLISECONDS).map(new Func1<Long, Long>() {
                    @Override
                    public Long call(Long aLong) {
                        return aLong * 10;
                    }
                }).take(5);
            }
        }).take(2);

        Observable.switchOnNext(observable).subscribe(new Subscriber<Long>() {
            @Override
            public void onCompleted() {
                System.out.println("Sequence complete.");
            }

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

            @Override
            public void onNext(Long aLong) {
                System.out.println("Next:" + aLong);
            }
        });
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
save_snippets.png
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30

執行結果如下：
Next:0
Next:10
Next:20
Next:0
Next:10
Next:20
Next:30
Next:40
Sequence complete.

zip操作符

zip操作符是把兩個observable提交的結果，嚴格按照順序進行合併，其流程圖如下：
這裡寫圖片描述

呼叫例子如下：

Observable<Integer> observable1 = Observable.just(10,20,30);
        Observable<Integer> observable2 = Observable.just(4, 8, 12, 16);
        Observable.zip(observable1, observable2, new Func2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer integer, Integer integer2) {
                return integer + integer2;
            }
        }).subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
                System.out.println("Sequence complete.");
            }

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

            @Override
            public void onNext(Integer value) {
                System.out.println("Next:" + value);
            }
        });
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
save_snippets.png
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23

執行結果如下：
Next:14
Next:28
Next:42
Sequence complete.

Error Handling Operators(Observable的錯誤處理操作符)

onErrorReturn操作符

onErrorReturn操作符是在Observable發生錯誤或異常的時候（即將回調oError方法時），攔截錯誤並執行指定的邏輯，返回一個跟源Observable相同型別的結果，最後回撥訂閱者的onComplete方法，其流程圖如下：
這裡寫圖片描述
呼叫例子如下：

Observable<Integer> observable = Observable.create(new Observable.OnSubscribe<Integer>() {
            @Override
            public void call(Subscriber<? super Integer> subscriber) {
                if (subscriber.isUnsubscribed()) return;
                //迴圈輸出數字
                try {
                    for (int i = 0; i < 10; i++) {
                        if (i == 4) {
                            throw new Exception("this is number 4 error！");
                        }
                        subscriber.onNext(i);
                    }
                    subscriber.onCompleted();
                } catch (Exception e) {
                    subscriber.onError(e);
                }
            }
        });

        observable.onErrorReturn(new Func1<Throwable, Integer>() {
            @Override
            public Integer call(Throwable throwable) {
                return 1004;
            }
        }).subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
                System.out.println("Sequence complete.");
            }

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

            @Override
            public void onNext(Integer value) {
                System.out.println("Next:" + value);
            }
        });
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
save_snippets.png
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40

執行結果如下：
Next:0
Next:1
Next:2
Next:3
Next:1004
Sequence complete.

onErrorResumeNext操作符

onErrorResumeNext操作符跟onErrorReturn類似，只不過onErrorReturn只能在錯誤或異常發生時只返回一個和源Observable相同型別的結果，而onErrorResumeNext操作符是在錯誤或異常發生時返回一個Observable，也就是說可以返回多個和源Observable相同型別的結果，其流程圖如下：
這裡寫圖片描述
呼叫例子如下：

Observable<Integer> observable = Observable.create(new Observable.OnSubscribe<Integer>() {
            @Override
            public void call(Subscriber<? super Integer> subscriber) {
                if (subscriber.isUnsubscribed()) return;
                //迴圈輸出數字
                try {
                    for (int i = 0; i < 10; i++) {
                        if (i == 4) {
                            throw new Exception("this is number 4 error！");
                        }
                        subscriber.onNext(i);
                    }
                    subscriber.onCompleted();
                } catch (Exception e) {
                    subscriber.onError(e);
                }
            }
        });

        observable.onErrorResumeNext(new Func1<Throwable, Observable<? extends Integer>>() {
            @Override
            public Observable<? extends Integer> call(Throwable throwable) {
                return Observable.just(100,101, 102);
            }
        }).subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
                System.out.println("Sequence complete.");
            }

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

            @Override
            public void onNext(Integer value) {
                System.out.println("Next:" + value);
            }
        });
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
save_snippets.png
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40

執行結果如下：
Next:0
Next:1
Next:2
Next:3
Next:100
Next:101
Next:102
Sequence complete.

onExceptionResumeNext操作符

onExceptionResumeNext操作符和onErrorResumeNext操作符類似，不同的地方在於onErrorResumeNext操作符是當Observable發生錯誤或異常時觸發，而onExceptionResumeNext是當Observable發生異常時才觸發。

這裡要普及一個概念就是，Java的異常分為錯誤（error）和異常（exception）兩種，它們都是繼承於Throwable類。

錯誤（error）一般是比較嚴重的系統問題，比如我們經常遇到的OutOfMemoryError、StackOverflowError等都是錯誤。錯誤一般繼承於Error類，而Error類又繼承於Throwable類，如果需要捕獲錯誤，需要使用try..catch(Error e)或者try..catch(Throwable e)句式。使用try..catch(Exception e)句式無法捕獲錯誤

異常（Exception）也是繼承於Throwable類，一般是根據實際處理業務丟擲的異常，分為執行時異常（RuntimeException）和普通異常。普通異常直接繼承於Exception類，如果方法內部沒有通過try..catch句式進行處理，必須通過throws關鍵字把異常丟擲外部進行處理（即checked異常）；而執行時異常繼承於RuntimeException類，如果方法內部沒有通過try..catch句式進行處理，不需要顯式通過throws關鍵字丟擲外部，如IndexOutOfBoundsException、NullPointerException、ClassCastException等都是執行時異常，當然RuntimeException也是繼承於Exception類，因此是可以通過try..catch(Exception e)句式進行捕獲處理的。
onExceptionResumeNext流程圖如下：
這裡寫圖片描述

呼叫例子如下：

 Observable<Integer> observable = Observable.create(new Observable.OnSubscribe<Integer>() {
            @Override
            public void call(Subscriber<? super Integer> subscriber) {
                if (subscriber.isUnsubscribed()) return;
                //迴圈輸出數字
                try {
                    for (int i = 0; i < 10; i++) {
                        if (i == 4) {
                            throw new Exception("this is number 4 error！");
                        }
                        subscriber.onNext(i);
                    }
                    subscriber.onCompleted();
                } catch (Throwable e) {
                    subscriber.onError(e);
                }
            }
        });

        observable.onExceptionResumeNext(Observable.just(100, 101, 102)).subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
                System.out.println("Sequence complete.");
            }

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

            @Override
            public void onNext(Integer value) {
                System.out.println("Next:" + value);
            }
        });
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
save_snippets.png
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35

執行結果如下：
Next:0
Next:1
Next:2
Next:3
Next:100
Next:101
Next:102
Sequence complete.

retry操作符

retry操作符是當Observable發生錯誤或者異常時，重新嘗試執行Observable的邏輯，如果經過n次重新嘗試執行後仍然出現錯誤或者異常，則最後回撥執行onError方法；當然如果源Observable沒有錯誤或者異常出現，則按照正常流程執行。其流程圖如下：
這裡寫圖片描述

呼叫例子如下：

Observable<Integer> observable = Observable.create(new Observable.OnSubscribe<Integer>() {
            @Override
            public void call(Subscriber<? super Integer> subscriber) {
                if (subscriber.isUnsubscribed()) return;
                //迴圈輸出數字
                try {
                    for (int i = 0; i < 10; i++) {
                        if (i == 4) {
                            throw new Exception("this is number 4 error！");
                        }
                        subscriber.onNext(i);
                    }
                    subscriber.onCompleted();
                } catch (Throwable e) {
                    subscriber.onError(e);
                }
            }
        });

        observable.retry(2).subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
                System.out.println("Sequence complete.");
            }

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

            @Override
            public void onNext(Integer value) {
                System.out.println("Next:" + value);
            }
        });
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
save_snippets.png
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35

執行結果如下：
Next:0
Next:1
Next:2
Next:3

Next:0
Next:1
Next:2
Next:3

Next:0
Next:1
Next:2
Next:3
Error: this is number 4 error！

retryWhen操作符

retryWhen操作符類似於retry操作符，都是在源observable出現錯誤或者異常時，重新嘗試執行源observable的邏輯，不同在於retryWhen操作符是在源Observable出現錯誤或者異常時，通過回撥第二個Observable來判斷是否重新嘗試執行源Observable的邏輯，如果第二個Observable沒有錯誤或者異常出現，則就會重新嘗試執行源Observable的邏輯，否則就會直接回調執行訂閱者的onError方法。其流程圖如下：
這裡寫圖片描述

呼叫例子如下：

Observable<Integer> observable = Observable.create(new Observable.OnSubscribe<Integer>() {
            @Override
            public void call(Subscriber<? super Integer> subscriber) {
                System.out.println("subscribing");
                subscriber.onError(new RuntimeException("always fails"));
            }
        });

        observable.retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
            @Override
            public Observable<?> call(Observable<? extends Throwable> observable) {
                return observable.zipWith(Observable.range(1, 3), new Func2<Throwable, Integer, Integer>() {
                    @Override
                    public Integer call(Throwable throwable, Integer integer) {
                        return integer;
                    }
                }).flatMap(new Func1<Integer, Observable<?>>() {
                    @Override
                    public Observable<?> call(Integer integer) {
                        System.out.println("delay retry by " + integer + " second(s)");
                        //每一秒中執行一次
                        return Observable.timer(integer, TimeUnit.SECONDS);
                    }
                });
            }
        }).subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
                System.out.println("Sequence complete.");
            }

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

            @Override
            public void onNext(Integer value) {
                System.out.println("Next:" + value);
            }
        });
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
save_snippets.png
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41

執行結果如下：
subscribing
delay retry by 1 second(s)
subscribing
delay retry by 2 second(s)
subscribing
delay retry by 3 second(s)
subscribing
Sequence complete.

好了，先介紹這麼多，下回繼續介紹其他的操作符，敬請期待！

相關推薦

RxJava系列3(轉換操作符)

圖片 distinct when return span 訂閱 結合 block latest 前面兩篇文章中我們介紹了RxJava的一些基本概念和RxJava最簡單的用法。從這一章開始，我們開始聊聊RxJava中的操作符Operators，RxJava中的操作符主要分

3章 RxJava操作符

CSDN學院課程地址 RxJava2從入門到精通-初級篇:https://edu.csdn.net/course/detail/10036 RxJava2從入門到精通-中級篇:https://edu.csdn.net/course/detail/10037

RxJava的操作符3

本篇文章繼續介紹以下型別的操作符 Combining Observables(Observable的組合操作符)Error Handling Operators(Observable的錯誤處理操作符)Combining Observables(Observable的組合操作符) combineLatest操

RxJava操作符總結之過濾

println 能夠 seconds 什麽 range tor array 延遲 lines RxJava操作符總結之過濾 jsut() just(T t1, T t2, T t3 ....) ，just能夠傳入多個同樣類型的參數，並將當前參數一

JavaScript操作符-3

返回值 == 邏輯操作符 否則 pan 第一個 cal 不一定 align 學習目標 1.掌握什麽是表達式 2.掌握javascript操作符的分類 3.掌握算數操作符 什麽是表達式 將類型的數據（如常量、變量、函數等），用運算符號按一定的規則鏈接起來的、有意義的式子稱為表

Android RxJava操作符詳解系列： 變換操作符

urn 原因 轉換 需要 生產 依賴 reat 入門 所有 Rxjava，由於其基於事件流的鏈式調用、邏輯簡潔 & 使用簡單的特點，深受各大 Android開發者的歡迎。Github截圖 如果還不了解 RxJava，請看文章：Android：這是一篇 清晰 &

Android RxJava操作符的學習---過濾操作符----聯想搜尋優化

1. 需求場景  2. 功能說明 3. 具體實現 佈局檔案：activity_filter.xml <?xml version="1.0" encoding="utf-8"?> <LinearLayout xmlns:android="ht

Android RxJava操作符的學習---過濾操作符----功能防抖

1. 需求場景   2. 功能說明   3. 具體實現 // 註冊控制元件 Button button; button = (Button)findViewById(R.id.button); /*

Android RxJava操作符的學習---過濾操作符

3.5 過濾操作符 作用 過濾 / 篩選 被觀察者（Observable）傳送的事件 & 觀察者 （Observer）接收的事件 應用場景 根據 指定條件 過濾事件 根據 指定事件數量 過濾事件 根據 指定時間 過濾事件

Android RxJava操作符的學習---功能性操作符--網路請求出錯重連（結合Retrofit）

1. 需求場景   2. 功能說明 功能需求說明     功能邏輯  例項說明 在本例子中：採用Get方法對 金山詞霸API 傳送網路請求 通過 斷開網路連線 模擬 網路異常錯誤（恢復網路即可

Android RxJava操作符的學習---功能性操作符--（有條件）網路請求輪詢（結合Retrofit）

1. 需求場景   2. 功能說明 採用Get方法對 金山詞霸API 按規定時間重複傳送網路請求，從而模擬 輪詢 需求實現 停止輪詢的條件 = 當輪詢到第4次時 採用 Gson 進行資料解析   3.

Android RxJava操作符的學習---功能性操作符

3.4 功能性操作符 作用 輔助被觀察者（Observable） 在傳送事件時實現一些功能性需求 實際應用場景 連線（訂閱） 觀察者 & 被觀察者 執行緒排程（切換） 錯誤處理 事件生命週期操作 延時操作 重複傳送操作

Android RxJava操作符的學習---組合合併操作符---聯合判斷多個事件

1. 需求場景 需要同時對多個事件進行聯合判斷 如，填寫表單時，需要表單裡所有資訊（姓名、年齡、職業等）都被填寫後，才允許點選 "提交" 按鈕 2. 功能說明 此處採用 填寫表單 作為聯合判斷功能展示 即，表單裡所有資訊（姓名、年齡、職業等）都被填寫後，

Android RxJava操作符的學習---組合合併操作符---從磁碟或記憶體快取中獲取快取資料

1. 需求場景     2. 功能說明 對於從磁碟 / 記憶體快取中 獲取快取資料 的功能邏輯如下： 3. 具體實現 詳細請看程式碼註釋 // 該2變數用於模擬記憶體快取 & 磁碟快取中的資料 String me

Android RxJava操作符的學習---組合合併操作符---合併資料來源並展示

1. 需求場景     2. 功能說明 即，同時向2個數據源獲取資料 -> 合併資料 -> 統一展示到客戶端 3. 具體實現 此處採用Merge（） & Zip（）操作符進行講解，其中： Merge（）例子 ：實現較為簡單

Android RxJava操作符的學習---組合 / 合併操作符

3.3 組合 / 合併操作符 3.3.1. 作用 組合 多個被觀察者（Observable） & 合併需要傳送的事件 應用場景 組合多個被觀察者 合併多個事件 傳送事件前追加發送事件 統計傳送事件數量 3.3.

Android RxJava操作符的學習---變換操作符---網路請求巢狀回撥

變換操作符的主要開發需求場景 = 巢狀回撥（Callback hell） 下面，我將採用一個實際應用場景例項來講解巢狀回撥（Callback hell） 1. 需求場景 1.1 背景 需要進行巢狀網路請求：即在第1個網路請求成功後，繼續再進行一次網路請求 如

Android RxJava操作符的學習---變換操作符

3.2 變換操作符 3.2.1.作用 對事件序列中的事件 / 整個事件序列 進行加工處理（即變換），使得其轉變成不同的事件 / 整個事件序列 具體原理    3.2.2.作用型別 應用場景 巢狀回撥（Callback hell） 根

Android RxJava操作符的學習---建立操作符--（無條件）網路請求輪詢

1. 需求場景   2. 功能說明 本文將採用Get方法對 金山詞霸API 按規定時間 重複傳送網路請求，從而模擬 輪詢 需求實現 採用 Gson 進行資料解析   3. 具體實現 下面結合 Ret

Android RxJava操作符的學習---建立操作符

RxJava如此受歡迎的原因，在於其提供了豐富 & 功能強大的操作符，幾乎能完成所有的功能需求 1. 簡介 RxJava 操作符的具體簡介如下： 2. 型別 RxJava功能強大，所以其對應的操作符的型別也非常多 根據功能作用的不同，Rx