Rxjava讲解（2）

2019-03-06 本文已影响1人 Simplelove_f033

上面文章讲过创建操作符，转换操作符，过滤操作符，这篇文字介绍组合操作符，条件操作符，功能操作符。

组合操作符

在Rxjava中组合操作符主要包含： Merge,StartWith,Concat,Zip,CombineLatest,SwitchOnNext,Join等等。

Merge

merge(Observable, Observable)将两个Observable发射的事件序列组合并成一个事件序列，就像是一个Observable发射的一样。你可以简单的将它理解为两个Obsrvable合并成了一个Observable，合并后的数据是无序的。

image

我们看下面的例子，一共有两个Observable：一个用来发送字母，另一个用来发送数字；现在我们需要两连个Observable发射的数据合并。

String[] letters = new String[]{"A", "B", "C", "D", "E", "F", "G", "H"};
Observable<String> letterSequence = Observable.interval(300, TimeUnit.MILLISECONDS)
        .map(new Func1<Long, String>() {
            @Override
            public String call(Long position) {
                return letters[position.intValue()];
            }
        }).take(letters.length);

Observable<Long> numberSequence = Observable.interval(500, TimeUnit.MILLISECONDS).take(5);

Observable.merge(letterSequence, numberSequence)
        .subscribe(new Observer<Serializable>() {
            @Override
            public void onCompleted() {
                System.exit(0);
            }

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

            @Override
            public void onNext(Serializable serializable) {
                System.out.print(serializable.toString()+" ");
            }
        });

程序输出：

A 0 B C 1 D E 2 F 3 G H 4

merge(Observable[])将多个Observable发射的事件序列组合并成一个事件序列，就像是一个Observable发射的一样。

image

StartWith
startWith(T)用于在源Observable发射的数据前插入数据。使用startWith(Iterable<T>)我们还可以在源Observable发射的数据前插入Iterable。官方示意图：

image

startWith(Observable<T>)用于在源Observable发射的数据前插入另一个Observable发射的数据（这些数据会被插入到源Observable发射数据的前面）。官方示意图：

image

Concat
concat(Observable<? extends T>, Observable<? extends T>) concat(Observable<？ extends Observable<T>>)用于将多个obserbavle发射的的数据进行合并发射，concat严格按照顺序发射数据，前一个Observable没发射玩是不会发射后一个Observable的数据的。它和merge、startWitch和相似，不同之处在于：

merge:合并后发射的数据是无序的；
startWitch:只能在源Observable发射的数据前插入数据。

image

这里我们将前面Merge操作符的例子拿过来，并将操作符换成Concat，然后我们看看执行结果：

String[] letters = new String[]{"A", "B", "C", "D", "E", "F", "G", "H"};
Observable<String> letterSequence = Observable.interval(300, TimeUnit.MILLISECONDS)
        .map(new Func1<Long, String>() {
            @Override
            public String call(Long position) {
                return letters[position.intValue()];
            }
        }).take(letters.length);

Observable<Long> numberSequence = Observable.interval(500, TimeUnit.MILLISECONDS).take(5);

Observable.concat(letterSequence, numberSequence)
        .subscribe(new Observer<Serializable>() {
            @Override
            public void onCompleted() {
                System.exit(0);
            }

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

            @Override
            public void onNext(Serializable serializable) {
                System.out.print(serializable.toString() + " ");
            }
        });

程序输出：

A B C D E F G H 0 1 2 3 4

Zip
zip(Observable, Observable, Func2)用来合并两个Observable发射的数据项，根据Func2函数生成一个新的值并发射出去。当其中一个Observable发送数据结束或者出现异常后，另一个Observable也将停在发射数据。

image

和前面的例子一样，我们将操作符换成了zip:

String[] letters = new String[]{"A", "B", "C", "D", "E", "F", "G", "H"};
Observable<String> letterSequence = Observable.interval(120, TimeUnit.MILLISECONDS)
        .map(new Func1<Long, String>() {
            @Override
            public String call(Long position) {
                return letters[position.intValue()];
            }
        }).take(letters.length);

Observable<Long> numberSequence = Observable.interval(200, TimeUnit.MILLISECONDS).take(5);

Observable.zip(letterSequence, numberSequence, new Func2<String, Long, String>() {
    @Override
    public String call(String letter, Long number) {
        return letter + number;
    }
}).subscribe(new Observer<String>() {
    @Override
    public void onCompleted() {
        System.exit(0);
    }

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

    @Override
    public void onNext(String result) {
        System.out.print(result + " ");
    }
});

程序输出：

A0 B1 C2 D3 E4

CombineLatest
combineLatest(Observable, Observable, Func2)用于将两个Observale最近发射的数据已经Func2函数的规则进展组合。下面是官方提供的原理图：

image

下面这张图应该更容易理解：

image

List<String> communityNames = DataSimulator.getCommunityNames();
List<Location> locations = DataSimulator.getLocations();

Observable<String> communityNameSequence = Observable.interval(1, TimeUnit.SECONDS)
        .map(new Func1<Long, String>() {
            @Override
            public String call(Long position) {
                return communityNames.get(position.intValue());
            }
        }).take(communityNames.size());
Observable<Location> locationSequence = Observable.interval(1, TimeUnit.SECONDS)
        .map(new Func1<Long, Location>() {
            @Override
            public Location call(Long position) {
                return locations.get(position.intValue());
            }
        }).take(locations.size());

Observable.combineLatest(
        communityNameSequence,
        locationSequence,
        new Func2<String, Location, String>() {
            @Override
            public String call(String communityName, Location location) {
                return "小区名:" + communityName + ", 经纬度:" + location.toString();
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onCompleted() {
                System.exit(0);
            }

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

            @Override
            public void onNext(String s) {
                System.out.println(s);
            }
        });

程序输出：

小区名:竹园新村, 经纬度:(21.827, 23.323)
小区名:康桥半岛, 经纬度:(21.827, 23.323)
小区名:康桥半岛, 经纬度:(11.923, 16.309)
小区名:中粮·海景壹号, 经纬度:(11.923, 16.309)
小区名:中粮·海景壹号, 经纬度:(22.273, 53.623)
小区名:浦江名苑, 经纬度:(22.273, 53.623)
小区名:南辉小区, 经纬度:(22.273, 53.623)

SwitchOnNext
switchOnNext(Observable<? extends Observable<? extends T>>用来将一个发射多个小Observable的源Observable转化为一个Observable，然后发射这多个小Observable所发射的数据。如果一个小的Observable正在发射数据的时候，源Observable又发射出一个新的小Observable，则前一个Observable发射的数据会被抛弃，直接发射新的小Observable所发射的数据。

结合下面的原理图大家应该很容易理解，我们可以看到下图中的黄色圆圈就被丢弃了。

image

Join

join(Observable, Func1, Func1, Func2)我们先介绍下join操作符的4个参数：

Observable：源Observable需要组合的Observable,这里我们姑且称之为目标Observable；
Func1：接收从源Observable发射来的数据，并返回一个Observable，这个Observable的声明周期决定了源Obsrvable发射出来的数据的有效期；
Func1：接收目标Observable发射来的数据，并返回一个Observable，这个Observable的声明周期决定了目标Obsrvable发射出来的数据的有效期；
Func2：接收从源Observable和目标Observable发射出来的数据，并将这两个数据组合后返回。

所以Join操作符的语法结构大致是这样的：onservableA.join(observableB, 控制observableA发射数据有效期的函数，控制observableB发射数据有效期的函数，两个observable发射数据的合并规则)

join操作符的效果类似于排列组合，把第一个数据源A作为基座窗口，他根据自己的节奏不断发射数据元素，第二个数据源B，每发射一个数据，我们都把它和第一个数据源A中已经发射的数据进行一对一匹配；举例来说，如果某一时刻B发射了一个数据“B”,此时A已经发射了0，1，2，3共四个数据，那么我们的合并操作就会把“B”依次与0,1,2,3配对，得到四组数据： [0, B][1, B] [2, B] [3, B]

再看看下面的图是不是好理解了呢？！

image

读懂了上面的文字，我们再来写段代码加深理解。

final List<House> houses = DataSimulator.getHouses();//模拟的房源数据，用于测试

//用来每秒从houses总取出一套房源并发射出去
Observable<House> houseSequence =
        Observable.interval(1, TimeUnit.SECONDS)
                .map(new Func1<Long, House>() {
                    @Override
                    public House call(Long position) {
                        return houses.get(position.intValue());
                    }
                }).take(houses.size());//这里的take是为了防止houses.get(position.intValue())数组越界

//用来实现每秒发送一个新的Long型数据
Observable<Long> tictoc = Observable.interval(1, TimeUnit.SECONDS);

houseSequence.join(tictoc,
        new Func1<House, Observable<Long>>() {
            @Override
            public Observable<Long> call(House house) {
                return Observable.timer(2, TimeUnit.SECONDS);
            }
        },
        new Func1<Long, Observable<Long>>() {
            @Override
            public Observable<Long> call(Long aLong) {
                return Observable.timer(0, TimeUnit.SECONDS);
            }
        },
        new Func2<House, Long, String>() {
            @Override
            public String call(House house, Long aLong) {
                return aLong + "-->" + house.getDesc();
            }
        }
).subscribe(new Observer<String>() {
    @Override
    public void onCompleted() {
        System.exit(0);
    }

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

    @Override
    public void onNext(String s) {
        System.out.println(s);
    }
});

程序输出：

0-->中粮海景壹号新出大平层！总价4500W起
1-->中粮海景壹号新出大平层！总价4500W起
1-->满五唯一，黄金地段
2-->中粮海景壹号新出大平层！总价4500W起
2-->满五唯一，黄金地段
2-->一楼自带小花园
3-->一楼自带小花园
3-->毗邻汤臣一品
4-->毗邻汤臣一品
4-->顶级住宅，给您总统般尊贵体验
5-->顶级住宅，给您总统般尊贵体验
5-->顶层户型，两室一厅
6-->顶层户型，两室一厅
6-->南北通透，豪华五房
7-->南北通透，豪华五房

条件操作符

条件操作符：通过设置函数，判断被观察者(Observable) 发送的事件是否符合条件。包括： contains(),exist(),isEmpty(), amb(),all(),takeWhile(),takeUntil(),skipUntil(),skipWhile(),defaultEmpty(),sequenceEqual()。

all
all() 操作符是判断发送到数据是否都满足指定的条件
public static void all() {

Observable
        .range(1, 5)
        .all(new Predicate<Integer>() {
            @Override
            public boolean test(Integer integer) throws Exception {
                return integer < 5;
            }
        })
        .subscribe(new Consumer<Boolean>() {
            @Override
            public void accept(Boolean aBoolean) throws Exception {
                Log.d(TAG + "all", aBoolean ? "发送数据都小于5" : "发送的数据不满足全小于5");
            }
        });

}
输出如下：

repeatUntil
repeatUntil() 操作符是repeat操作符的升级版。可以动态控制是否继续重复发射事件序列。 return 则停止重复，return 则继续重复发射

static int count = 0;

public static void repeatUntil() {
    Observable
            .just(1, 2, 3)
            .repeatUntil(new BooleanSupplier() {
                @Override
                public boolean getAsBoolean() throws Exception {
                    count++;
                    if (count >= 2)
                        return true;
                    return false;
                }
            })
            .subscribe(new Consumer<Integer>() {
                @Override
                public void accept(Integer integer) throws Exception {
                    Log.d(TAG + "repeat", String.valueOf(integer));
                }
            });
}

输出如下：

takeUntil
takeUntil() 操作符是发送complete的结束条件，当然发送结束之前也会包含这个值. return true 时结束，false继续
public static void takeUntil() {
Observable
.range(0, 10)
.takeUntil(new Predicate<Integer>() {
@Override
public boolean test(Integer integer) throws Exception {
if (integer.equals(5))
return true;
return false;
}
})
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.d(TAG + "takeUntil", String.valueOf(integer));
}
});
}
输出如下：

image.png

/**
* takeUntil 也能传入一个被观察者Observable,当该Obervable开始发送数据时(注意：观察者Observer不会接收事件)，那么原始的Observable则停止发送
*/
public static void takeUntil2() {
Observable
.interval(1, TimeUnit.SECONDS)
.takeUntil(Observable.timer(5, TimeUnit.SECONDS))
.subscribe(new Consumer<Long>() {
@Override
public void accept(Long aLong) throws Exception {
Log.d(TAG + "takeUntil2", String.valueOf(aLong));
}
});
}

输出如下：

image.png
takeWhile
takeWhile() 操作符表示满足条件则发送，不满足则停止发送

public static void takeWhile() {
    Observable.range(0, 10)
            .takeWhile(new Predicate<Integer>() {
                @Override
                public boolean test(Integer integer) throws Exception {
                    if (integer < 6)
                        return true;
                    return false;
                }
            })
            .subscribe(new Consumer<Integer>() {
                @Override
                public void accept(Integer integer) throws Exception {
                    Log.d(TAG + "takeWhile", String.valueOf(integer));
                }
            });
}

输出如下：

skipWhile
skipWhile操作符主要是判断发送的每项数据是否满足指定函数条件。直到该判断条件为false时，才开始发送observable的数据(前面的实际会丢弃)
public static void skipWhile() {
Observable
.interval(1, TimeUnit.SECONDS)
.skipWhile(new Predicate<Long>() {
@Override
public boolean test(Long aLong) throws Exception {
if (aLong > 5)
return false;
else return true;
}
})
.subscribe(new Consumer<Long>() {
@Override
public void accept(Long aLong) throws Exception {
Log.d(TAG + "skipWhile", String.valueOf(aLong));
}
});
}
输出如下：

sequenceEqual
sequenceEqual操作符主要是判断两个obervable需要发送的数据是否相等，如果相同则返回true，否则返回false

 public static void sequenceEqual() {
    Observable
            .sequenceEqual(Observable.just(4, 5, 6), Observable.just(4, 5, 6))
            .subscribe(new Consumer<Boolean>() {
                @Override
                public void accept(Boolean aBoolean) throws Exception {
                    Log.d(TAG + "sequenceEq", "两个Obervable是否相等：" + aBoolean);
                }
            });
}

输出如下：

contains
contains操作符是判断发送的数据是否包含指定数据

public static void contains() {
    Observable
            .just(1, 2, 3, 4, 5)
            .contains(3)
            .subscribe(new Consumer<Boolean>() {
                @Override
                public void accept(Boolean aBoolean) throws Exception {
                    Log.d(TAG + "contains", "发送的数据是否包含3：" + (aBoolean ? "是" : "否"));
                }
            });
}

输出如下：

isEmpty
isEmpty操作符是判断被观察者发送的数据是否为空
public static void isEmpty() {
Observable
.just(1)
.isEmpty()
.subscribe(new Consumer<Boolean>() {
@Override
public void accept(Boolean aBoolean) throws Exception {
Log.d(TAG + "isEmpty", "发送的数据是否为空：" + (aBoolean ? "是" : "否"));
}
});
}

输出如下：

amb
amb操作符主要是当需要发送多个Observable时，只发送先发送数据的Observerable 的数据，其余Observable会被丢弃

public static void amb() {

    List<ObservableSource<Integer>> list = new ArrayList<>();

    Observable observable1 = Observable.just(1, 2, 3);
    Observable observable2 = Observable.just(4, 5, 6).delay(2, TimeUnit.SECONDS);

    list.add(observable1);
    list.add(observable2);

    Observable
            .amb(list)
            .subscribe(new Observer<Integer>() {
                @Override
                public void onSubscribe(Disposable d) {

                }

                @Override
                public void onNext(Integer integer) {
                    //只能接收到observable1 的发送的数据，而observable2会被丢弃
                    Log.d(TAG + "amb", String.valueOf(integer));
                }

                @Override
                public void onError(Throwable e) {

                }

                @Override
                public void onComplete() {
                }
            });
}

输出如下：

defaultEmpty
defaultEmpty操作符主要是在不发送一个有效事件(next事件)、仅发送了complete事件的前提下，发送一个默认值

public static void defaultEmpty() {
 
    Observable
            .create(new ObservableOnSubscribe<Integer>() {
                @Override
                public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                    emitter.onComplete();
                }
            })
            .defaultIfEmpty(6)
            .subscribe(new Consumer<Integer>() {
                @Override
                public void accept(Integer integer) throws Exception {
                    Log.d(TAG + "defaultIfEmp", String.valueOf(integer));
                }
            });
}

输出如下：

[图片上传中...(image-a94761-1551849184066-0)]

功能操作符

功能操作符：辅助被观察者(Observable) 发送事件时实现一些功能性需求，如错误处理，线程调度
subscribe
subscribe主要是负责Observable与Observer进行订阅，

 public static void all() {

    Observable
            .range(1, 5)
            .all(new Predicate<Integer>() {
                @Override
                public boolean test(Integer integer) throws Exception {
                    return integer < 5;
                }
            })
            .subscribe(new Consumer<Boolean>() {
                @Override
                public void accept(Boolean aBoolean) throws Exception {
                    Log.d(TAG + "all", aBoolean ? "发送数据都小于5" : "发送的数据不满足全小于5");
                }
            });
}

输出如下：

**repeatUntil() **
repeatUntil主要是延迟发送事件，repeatUntil有多个重载方法：

1、delay(long delay,TimeUnit unit) :指定延迟时间。参数一：时间；参数二：时间单位
2、 delay(long delay, TimeUnit unit, Scheduler scheduler) 指定延迟时间&线程调度器。参数一：时间；参数二：时间单位；参数三：线程调度器
3、delay(long delay, TimeUnit unit, boolean delayError) 指定延迟时间&线程调度器。参数一：时间；参数二：时间单位；参数三：是否错误延迟
4、delay(long delay, TimeUnit unit, Scheduler scheduler, boolean delayError) 指定延迟时间&线程调度器&错误延迟参数一：时间；参数二：时间单位；
参数三：线程调度器; 参数四：是否错误延迟(若中间发生错误，是否如常执行，执行完在执行onError())

 public static void delay() {
    Observable
            .just(1, 2)
            .delay(10, TimeUnit.SECONDS)
            .subscribe(new Consumer<Integer>() {
                @Override
                public void accept(Integer integer) throws Exception {
                    Log.d(TAG + "delay", String.valueOf(integer));
                }
            });
}

输出如下：

[图片上传中...(image-7a5f2a-1551849868197-0)]

**do **
do操作符主要在在事件发送&接收的整个周期过程中进行操作。如发送事件前的操作，发送事件后的回调请求

do系列操作符包含以下：

doOnEach() :当Observable每发送一次事件就会调用一次(包含onNext()，onError()，onComplete())
doOnNext(): 执行 onNext()前调用
doAfterNext()：执行onNext()后调用
doOnComplete()：执行onComplete()前调用
doOnError():执行 onError()前调用
doOnTerminate(): 执行终止(无论正常发送完毕/异常终止)
doFinally(): 最后执行
doOnSubscribe() ：观察者订阅是调用

doOnUnScbscribe()：观察者取消订阅时调用

 public static void dos() {
     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 NullPointerException());
                 }
             })
             .doOnEach(new Consumer<Notification<Integer>>() {
                 @Override
                 public void accept(Notification<Integer> integerNotification) throws Exception {
                     Log.d(TAG + "doOnEach", "doOnEach:  " + String.valueOf(integerNotification.getValue()));
                 }
             })
             .doOnNext(new Consumer<Integer>() {
                 @Override
                 public void accept(Integer integer) throws Exception {
                     Log.d(TAG + "doOnNext", "doOnNext:  " + String.valueOf(integer));
                 }
             })
             .doAfterNext(new Consumer<Integer>() {
                 @Override
                 public void accept(Integer integer) throws Exception {
                     Log.d(TAG + "doAfterNext", "doAfterNext:  " + String.valueOf(integer));
                 }
             })
             .doOnComplete(new Action() {
                 @Override
                 public void run() throws Exception {
                     Log.d(TAG + "doOnComplete", "doOnComplete");
                 }
             })
             .doOnError(new Consumer<Throwable>() {
                 @Override
                 public void accept(Throwable throwable) throws Exception {
                     Log.d(TAG + "doOnError", "doOnError");
                 }
             })
             .doOnTerminate(new Action() {
                 @Override
                 public void run() throws Exception {
                     Log.d(TAG + "doOnTerminate", "doOnTerminate");
                 }
             })
             .doAfterTerminate(new Action() {
                 @Override
                 public void run() throws Exception {
                     Log.d(TAG + "doAfterTermi", "doAfterTerminate");
                 }
             })
             .doOnSubscribe(new Consumer<Disposable>() {
                 @Override
                 public void accept(Disposable disposable) throws Exception {
                     Log.d(TAG + "doOnSubscribe", "doOnSubscribe");
                 }
             })
             .doFinally(new Action() {
                 @Override
                 public void run() throws Exception {
                     Log.d(TAG + "doFinally", "doFinally");
                 }
             })
             .subscribe(new Observer<Integer>() {
                 @Override
                 public void onSubscribe(Disposable d) {
                     Log.d(TAG, "onSubscribe");
                 }
 
                 @Override
                 public void onNext(Integer integer) {
                     Log.d(TAG, "收到的数据：  " + integer);
                 }
 
                 @Override
                 public void onError(Throwable e) {
                     Log.d(TAG, "onError");
                 }
 
                 @Override
                 public void onComplete() {
                     Log.d(TAG, "onComplete");
                 }
             });

 }

输出如下：

onErrorReturn
onErrorReturn操作符可以捕获错误，遇到错误时，发送一个特需的事件，并且正常终止注意后面的事件将不会发送

      public static void onErrorReturn() {

        Observable
                .create(new ObservableOnSubscribe<Integer>() {
                    @Override
                    public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                        emitter.onNext(1);
                        emitter.onNext(2);
                        emitter.onError(new Throwable("Throwable"));
                        emitter.onNext(3);

                    }
                })
                .onErrorReturn(new Function<Throwable, Integer>() {
                    @Override
                    public Integer apply(Throwable throwable) throws Exception {
                        Log.e(TAG, "发生了错误：  " + throwable.getMessage());
                        return 404;
                    }
                })
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.d(TAG, String.valueOf(integer));
                    }

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

                    @Override
                    public void onComplete() {
                        Log.d(TAG, "onComplete");
                    }
                });

    }

输出如下：

onExceptionResumeNext/onErrorResumeNext
该两个操作符主要是在遇到错误时发送一个新的Observable,并且正常终止原Observable后面的事件将不会发送，如果捕获Exception的话是要onExceptionResumeNext，捕获错误的用onErrorResumeNext()

    public static void onExceptionResumeNext() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onError(new NullPointerException("NullPointerException"));
                emitter.onNext(3);
            }
        }).onExceptionResumeNext(new Observable<Integer>() {
            @Override
            protected void subscribeActual(Observer<? super Integer> observer) {
                observer.onNext(4);
                observer.onNext(5);

            }
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Integer integer) {
                Log.d(TAG, String.valueOf(integer));
            }

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

            @Override
            public void onComplete() {
                Log.d(TAG, "onComplete");
            }
        });
    }

输出如下：

retry
retry操作符作用是出现错误时让被观察者重新发送数据
注：若发送错误，则一直重新发送
有几个重载方法：

retry() ：出现错误时，让被观察者重新发送数据。若错误一直发生，则一直重新发送
retry(long time)：与retry不同的书，若错误一直发生，被观察者则一直重新发送数据，但这持续重新发送有次数限制
retry(Predicate predicate) ：出现错误时，根据指定逻辑(可以捕获到发生的错误)决定是否让被观察者重新发送数据
retry(new BiPredicate<Integer, Throwable>)：出现错误时，根据指定逻辑(可以捕获重发的次数和发生的错误)决定是否让被观察者重新发送数据

retry（long time,Predicate predicate）：出现错误时，根据指定逻辑(可以捕获到发生的错误)决定是否让被观察者重新发送数据。并且有持续重发的次数限制

 public static void retry() {

     Observable
             .create(new ObservableOnSubscribe<Integer>() {
                 @Override
                 public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                     emitter.onNext(1);
                     emitter.onNext(2);
                     emitter.onError(new Throwable("发生错误了"));
                     emitter.onNext(3);
                 }
             })
             .retry(new BiPredicate<Integer, Throwable>() {
                 @Override
                 public boolean test(Integer integer, Throwable throwable) throws Exception {

                     // interger 为重试次数 ，throwable 为捕获到的异常

                     Log.e(TAG + "retry", throwable.getMessage());
                     Log.e(TAG + "integer", "重试次数： " + integer);

                     //return true ： 重新发送请求(若持续遇到错误，就持续重新发送)
                     //return false ：    不重新发送数据 并且调用观察者的onError()方法结束

                     if (integer > 2)
                         return false;
                     return true;
                 }
             })
             .subscribe(new Observer<Integer>() {
                 @Override
                 public void onSubscribe(Disposable d) {

                 }

                 @Override
                 public void onNext(Integer integer) {
                     Log.e(TAG + "retry", String.valueOf(integer));
                 }

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

                 @Override
                 public void onComplete() {
                     Log.d(TAG, "onComplete");
                 }
             });
 }

输出如下：

retryUntil
retryUntil的作用是在发送事件遇到错误，指定规则是否重新发送。retry(Predicate predicate)
return true ：不重新发送请求，并且调用观察者的onError()方法结束
return false ：重新发送数据(若持续遇到错误，就持续重新发送)

    public static void retryUntil() {

        Observable
                .create(new ObservableOnSubscribe<Integer>() {
                    @Override
                    public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                        emitter.onNext(1);
                        emitter.onNext(2);
                        emitter.onError(new Throwable("发生错误了"));
                        emitter.onNext(3);
                    }
                })
                .retryUntil(new BooleanSupplier() {
                    @Override
                    public boolean getAsBoolean() throws Exception {

                         //return true ： 不重新发送请求，并且调用观察者的onError()方法结束
                        // return false ： 重新发送数据(若持续遇到错误，就持续重新发送)
                        return false;
                    }
                })
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.e(TAG + "retryUntil", String.valueOf(integer));
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {
                        Log.d(TAG + "retryUntil", "onComplete");
                    }
                });
    }

输出如下：

** retryWhen**
retryWhen操作符作用是在遇到错误时，将发生的错误传递给一个新的被观察者(Observable)，并决定是否需要重新订阅原始被观察者(Observable) & 发送事件

    public static void retryWhen() {
        Observable
                .create(new ObservableOnSubscribe<Integer>() {
                    @Override
                    public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                        emitter.onNext(1);
                        emitter.onNext(2);
                        emitter.onError(new Throwable("发送了错误"));
                        emitter.onNext(3);
                    }
                })
                //遇到Error时会回调
                .retryWhen(new Function<Observable<Throwable>, ObservableSource<?>>() {
                    @Override
                    public ObservableSource<?> apply(Observable<Throwable> throwableObservable) throws Exception {
                        return throwableObservable.flatMap(new Function<Throwable, ObservableSource<?>>() {
                            @Override
                            public ObservableSource<?> apply(Throwable throwable) throws Exception {

                                //1、若返回的Observable发送的事件 = Error ，则原始的Observable则不重新发送事件。该异常信息可在观察者的onError中获得
                                //return Observable.error(throwable);

                                //2、若返回的Observable发送的事件= Next事件(和next的内容无关)，则原始的Observable重新发送事件(若持续遇到错误，则持续发送)
                                return Observable.just(5); //仅仅是作为一个触发重新订阅原被观察者的通知，什么数据并不重要，只有不是onComplete/onError事件
                            }
                        });

                    }
                })
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.d(TAG + "retryWhen", String.valueOf(integer));
                    }

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

                    @Override
                    public void onComplete() {
                        Log.d(TAG + "retryWhen", "onComplete");
                    }
                });
    }

输出如下：

repeat
repeat操作符的作用是重复发射 observable的数据序列，可以使无限次也可以是指定次数.不传时为重复无限次

public static void repeat() {
    Observable
            .just(1, 2, 3)
            .repeat(3)
            .subscribe(new Consumer<Integer>() {
                @Override
                public void accept(Integer integer) throws Exception {
                    Log.d(TAG + "repeat", String.valueOf(integer));
                }
            });
}

输出如下：

repeatWhen
repeatWhen作用是将原始 Observable 停止发送事件的标识（Complete（） / Error（））转换成1个 Object 类型数据传递给1个新被观察者（Observable）以此决定是否重新订阅 & 发送原来的 Observable

    public static void repeatWhen() {

        Observable
                .just(1, 2, 4)
                .repeatWhen(new Function<Observable<Object>, ObservableSource<?>>() {
                    @Override
                    public ObservableSource<?> apply(Observable<Object> objectObservable) throws Exception {
                        return  objectObservable.flatMap(new Function<Object, ObservableSource<?>>() {
                            @Override
                            public ObservableSource<?> apply(Object o) throws Exception {

                                //若新被观察者（Observable）返回1个Complete()/  Error()事件，则不重新订阅 & 发送原来的 Observable
                                //Observable.empty() = 发送Complete事件，但不会回调观察者的onComplete（）
                                return Observable.empty();

                                // return Observable.error(new Throwable("不再重新订阅事件"));
                                // 返回Error事件 = 回调onError（）事件，并接收传过去的错误信息。

                                // 情况2：若新被观察者（Observable）返回其余事件，则重新订阅 & 发送原来的 Observable
                                // return Observable.just(1);
                                // 仅仅是作为1个触发重新订阅被观察者的通知，发送的是什么数据并不重要，只要不是Complete（） /  Error（）事件
                            }
                        });
                    }
                })
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(Integer integer) {

                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }

输出如下：

debounce
debounce作用是一定的时间内没有操作就会发送事件（只会发送最后一次操作的事件）。
以下的例子：发送5个事件，每个事件间隔1秒。但是debounce限定了2秒内没有任何操作才会真正发送事件。所以只有最后一次满足条件，只能接收到事件 5

 public static void debounce() {
    Observable.intervalRange(1, 5, 0, 1, TimeUnit.SECONDS)
            .debounce(2, TimeUnit.SECONDS)
            .subscribe(new Consumer<Long>() {
                @Override
                public void accept(Long aLong) throws Exception {
                    Log.d(TAG + "debounce", String.valueOf(aLong));
                }
            });
}

输出如下：

**subscribeOn / ObserverOn **
subscribeOn : 发送事件的线程
observerOn：接收事件的线程
线程调度器：
* Schedulers.io(): 代表io操作的线程，通常用于网络，读写文件等io密集型的操作
* Schedulers.compucation(): 代表CPU计算密集型的操作，例如需要大量计算的操作
* Schedulers.newThread(): 代表一个常规的新线程
* AndroidSchedulers。mainThread(): 代表Android的主线程

    public static void subscribeOn_observerOn() {

        Observable
                .create(new ObservableOnSubscribe<String>() {
                    @Override
                    public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                        emitter.onNext("事件");
                        Log.d(TAG + "subscribeOn_ObserverOn", "发送事件:" + Thread.currentThread().getName());
                    }
                })
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        Log.d(TAG + "subscribeOn_ObserverOn", "接收事件:   " + Thread.currentThread().getName());
                    }
                });

    }

输出如下：

Rxjava讲解（2）

组合操作符

条件操作符

功能操作符

猜你喜欢

热点阅读