RxJava2详细攻略（四）

5 过滤操作符

5.1 filter()

方法预览

public final Observable<T> filter(Predicate<? super T> predicate)

有什么用？

通过一定的逻辑来过滤被观察者发送的事件，如果返回true则会发送事件，否则不会发送。

怎么用？

Observable.just(1, 2, 3)
    .filter(new Predicate<Integer>(){
        @Override
        public boolean test(Integer integer) throws Exception{
            return integer < 2;
        }
    })
    .subscribe(new Observer<Integer>(){
        @Override
    public void onSubscribe(Disposable d) {
        Log.d(TAG, "==================onSubscribe ");
    }

    @Override
    public void onNext(Integer integer) {
        i += integer;
        Log.d(TAG, "==================onNext " + integer);
    }

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

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

以上代码只有小于2的事件才会发送，来看看打印结果：

05-24 22:57:32.562 12776-12776/com.example.louder.rxjavademo D/chan: ==================onSubscribe 
==================onNext 1
==================onComplete 

5.2 ofType()

方法预览

public final<U> Observable<U> ofType(final Class <U> clazz)

有什么用？

可以过滤不符合该类型事件

怎么用？

Observable.just(1, 2, 3, "chan", "zhide")
.ofType(Integer.class)
.subscribe(new Observer<Integer>(){
    @Override
    public void onSubscribe(Disposable d){
        Log.d(TAG, "==onSubscribe");
    }
    
    @Override
    public void onNext(Integer integer){
        i += integer;
        Log.d(TAG, "==onNext" + integer);
    }
    
    @Override
    public void onError(Throwable e){
        Log.d(TAG, "==onError");
    }
    
    @Override
    public void onComplete(){
        Log.d(TAG, "==onComplete");
    }
});

打印结果：

05-24 23:04:24.752 13229-13229/? D/chan: ==================onSubscribe 
==================onNext 1
==================onNext 2
==================onNext 3
05-24 23:04:24.753 13229-13229/? D/chan: ==================onComplete 

5.3 skip()

方法预览

public final Observable<T> skip(long count)
......

有什么用？

跳过正序某些事件，count代表跳过事件的数量。

怎么用？

Observable.just(1, 2, 3)
.skip(2)
.subscribe(new Observer<Integer>(){
    @Override
    public void onSubscribe(Disposable d){
        Log.d(TAG, "==onSubscribe");
    }
    
    @Override
    public void onNext(Integer integer){
        i += integer;
        Log.d(TAG, "==onNext" + integer);
    }
    
    @Override
    public void onError(Throwable e){
        Log.d(TAG, "==onError");
    }
    
    @Override
    public void onComplete(){
        Log.d(TAG, "==onComplete");
    }
});

打印结果：

05-24 23:13:50.448 13831-13831/? D/chan: ==================onSubscribe 
05-24 23:13:50.449 13831-13831/? D/chan: ==================onNext 3
==================onComplete 

skipLast()作用也是跳过某些事件，不过它用来跳过正序的后面的事件，这里就不再讲解。

5.4 distinct()

方法预览

public final Observable<T> distinct()

有什么用？

过滤事件序列中的重复事件。

怎么用？

Observable.just(1, 2, 3, 3, 2, 1)
.distinct()
.subscribe(new Observer < Integer > () {
    @Override
    public void onSubscribe(Disposable d) {
        Log.d(TAG, "==================onSubscribe ");
    }

    @Override
    public void onNext(Integer integer) {
        i += integer;
        Log.d(TAG, "==================onNext " + integer);
    }

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

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

打印结果：

05-24 23:19:44.334 14206-14206/com.example.louder.rxjavademo D/chan: ==================onSubscribe 
==================onNext 1
==================onNext 2
==================onNext 3
==================onComplete 

5.5 distinctUntilChanged()

方法预览

public final Observable<T> distinctUntilChanged()

有什么用？

过滤连续重复的事件

怎么用？

Observable.just(1, 2, 3, 3, 2, 1)
.distinctUntilChanged()
.subscribe(new Observer<Integer>(){
     @Override
    public void onSubscribe(Disposable d) {
        Log.d(TAG, "==================onSubscribe ");
    }

    @Override
    public void onNext(Integer integer) {
        i += integer;
        Log.d(TAG, "==================onNext " + integer);
    }

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

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

因为事件序列中连续出现两次3，所以第二次3并不会发出。

5.6 take()

方法预览

public final Observable<T> take(long count)
......

有什么用？

控制观察者接收的事件的数量

怎么用？

Observable.just(1, 2, 3, 4, 5)
.take(3)
.subscribe(new Observer<Integer>(){
    @Override
    public void onSubscribe(Disposable d){
        Log.d(TAG, "==onSubscribe");
    }
    
    @Override
    public void onNext(Integer integer){
        i += integer;
        Log.d(TAG, "==onNext" + integer);
    }
    
    @Override
    public void onError(Throwable e){
        Log.d(TAG, "==onError");
    }
    
    @Override
    public void onComplete(){
        Log.d(TAG, "==onComplete");
    }
});

打印结果：

05-24 23:28:32.899 14704-14704/? D/chan: ==================onSubscribe 
==================onNext 1
==================onNext 2
==================onNext 3
==================onComplete 

takeLast（）的作用就是控制观察者只能接受事件序列后面几件事情，这里就不再讲解。

5.7 debounce()

方法预览

public final Observable<T> debounce(long timeout, TimeUnit unit)
......

有什么用？

如果两个事件发送的时间间隔小于设定的时间间隔则前一个事件就不会发送给观察者。

怎么用？

Observable.create(new ObservableOnSubscribe<Integer>(){
    @Override
    public void subscribe(ObservableEmitter<Integer> e) throws Exception{
        e.onNext(1);
        Thread.sleep(900);
        e.onNext(2);
    }
})
.debounce(1, TimeUnit.SECONDS)
.subscribe(new Observer<Integer>(){
    @Override
    public void onSubscribe(Disposable d){
        Log.d(TAG, "==onSubscribe");
    }
    
    @Override
    public void onNext(Integer integer){
        Log.d(TAG, "==onNext" + integer);
    }
     @Override
    public void onError(Throwable e) {
        Log.d(TAG, "===================onError ");
    }

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

打印结果：

05-25 20:39:10.512 17441-17441/com.example.rxjavademo D/chan: ===================onSubscribe 
05-25 20:39:12.413 17441-17478/com.example.rxjavademo D/chan: ===================onNext 2

可以看到事件1并没有发送出去，现在将间隔时间改为1000，看看打印结果：

05-25 20:42:10.874 18196-18196/com.example.rxjavademo D/chan: ===================onSubscribe 
05-25 20:42:11.875 18196-18245/com.example.rxjavademo D/chan: ===================onNext 1
05-25 20:42:12.875 18196-18245/com.example.rxjavademo D/chan: ===================onNext 2

5.8 firstElement() && lastElement()

方法预览

public final Maybe<T> firstElement()
public final Maybe<T> lastElement()

有什么用？

firstElement()取事件序列的第一个元素，lastElement取最后一个元素。

怎么用？

Observable.just(1, 2, 3, 4)
.firstElement()
.subscribe(new Consumer<Integer> (){
    @Override
    public void accept(Integer integer) throws Exception{
        Log.d(TAG, "==firstElement" + integer);
    }
});
Observable.just(1, 2, 3, 4)
.lastElement()
.subscribe(new Consumer < Integer > () {
    @Override
    public void accept(Integer integer) throws Exception {
        Log.d(TAG, "====================lastElement " + integer);
    }
});

打印结果：

05-25 20:47:22.189 19909-19909/? D/chan: ====================firstElement 1
====================lastElement 4

5.9 elementAt() & elementAtOrError()

方法预览

public final Maybe<T> elementAt(long index)
public final Single<T> elementAtOrError(long index)

有什么用？
elementAt()可以指定取出事件序列中事件，但是输入的index超出事件序列的总数的话就不会出现任何结果。这种情况下，你想发出异常信息的话就用elementAtOrError（）。

Observable.just(1, 2, 3, 4)
.elementAt(0)
.subscribe(new Consumer<Integer> (){
    @Override
    public void accept(Integer integer) throws Exception{
        Log.d(TAG, "==accept" + integer);
    }
});

打印结果：

05-25 20:56:22.266 23346-23346/com.example.rxjavademo D/chan: ====================accept 1

将elementAt（）的值改为5，这时是没有打印结果的，因为没有满足条件的元素。

替换elementAt()为elementAtError()，代码如下：

Observable.just(1, 2, 3, 4)
.elementAtOrError(5)
.subscribe(new Consumer<Integer>(){
    @Override
    public void accept(Integer integer) throws Exception{
        Log.d(TAG, "==accept" + integer);
    }
});

打印结果：

io.reactivex.exceptions.OnErrorNotImplementedException
at io.reactivex.internal.functions.Functions$OnErrorMissingConsumer.accept(Functions.java: 704)
at io.reactivex.internal.functions.Functions$OnErrorMissingConsumer.accept(Functions.java: 701)
at io.reactivex.internal.observers.ConsumerSingleObserver.onError(ConsumerSingleObserver.java: 47)
at io.reactivex.internal.operators.observable.ObservableElementAtSingle$ElementAtObserver.onComplete(ObservableElementAtSingle.java: 117)
at io.reactivex.internal.operators.observable.ObservableFromArray$FromArrayDisposable.run(ObservableFromArray.java: 110)
at io.reactivex.internal.operators.observable.ObservableFromArray.subscribeActual(ObservableFromArray.java: 36)
at io.reactivex.Observable.subscribe(Observable.java: 10903)
at io.reactivex.internal.operators.observable.ObservableElementAtSingle.subscribeActual(ObservableElementAtSingle.java: 37)
at io.reactivex.Single.subscribe(Single.java: 2707)
at io.reactivex.Single.subscribe(Single.java: 2693)
at io.reactivex.Single.subscribe(Single.java: 2664)
at com.example.rxjavademo.MainActivity.onCreate(MainActivity.java: 103)
at android.app.Activity.performCreate(Activity.java: 6942)
at android.app.Instrumentation.callActivityOnCreate(Instrumentation.java: 1126)
at android.app.ActivityThread.performLaunchActivity(ActivityThread.java: 2880)
at android.app.ActivityThread.handleLaunchActivity(ActivityThread.java: 2988)
at android.app.ActivityThread. - wrap14(ActivityThread.java)
at android.app.ActivityThread$H.handleMessage(ActivityThread.java: 1631)
at android.os.Handler.dispatchMessage(Handler.java: 102)
at android.os.Looper.loop(Looper.java: 154)
at android.app.ActivityThread.main(ActivityThread.java: 6682)
at java.lang.reflect.Method.invoke(Native Method)
at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java: 1520)
at com.android.internal.os.ZygoteInit.main(ZygoteInit.java: 1410)
Caused by: java.util.NoSuchElementException
at io.reactivex.internal.operators.observable.ObservableElementAtSingle$ElementAtObserver.onComplete(ObservableElementAtSingle.java: 117) 
at io.reactivex.internal.operators.observable.ObservableFromArray$FromArrayDisposable.run(ObservableFromArray.java: 110) 
at io.reactivex.internal.operators.observable.ObservableFromArray.subscribeActual(ObservableFromArray.java: 36) 
at io.reactivex.Observable.subscribe(Observable.java: 10903) 
at io.reactivex.internal.operators.observable.ObservableElementAtSingle.subscribeActual(ObservableElementAtSingle.java: 37) 
at io.reactivex.Single.subscribe(Single.java: 2707) 
at io.reactivex.Single.subscribe(Single.java: 2693) 
at io.reactivex.Single.subscribe(Single.java: 2664) 
at com.example.rxjavademo.MainActivity.onCreate(MainActivity.java: 103) 
at android.app.Activity.performCreate(Activity.java: 6942) 
at android.app.Instrumentation.callActivityOnCreate(Instrumentation.java: 1126) 
at android.app.ActivityThread.performLaunchActivity(ActivityThread.java: 2880) 
at android.app.ActivityThread.handleLaunchActivity(ActivityThread.java: 2988) 
at android.app.ActivityThread. - wrap14(ActivityThread.java) 
at android.app.ActivityThread$H.handleMessage(ActivityThread.java: 1631) 
at android.os.Handler.dispatchMessage(Handler.java: 102) 
at android.os.Looper.loop(Looper.java: 154) 
at android.app.ActivityThread.main(ActivityThread.java: 6682) 
at java.lang.reflect.Method.invoke(Native Method) 
at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java: 1520) 
at com.android.internal.os.ZygoteInit.main(ZygoteInit.java: 1410) 

这时候会抛出NoSuchElementException异常。

6 条件操作符

6.1 all()

方法预览

public final Observable<T> ambWith(ObservableSource<? extends T> other)

有什么用？

判断事件序列是否全部满足两个事件，如果都满足则返回true，反之则返回flase。

怎么用？

Observable.just(1, 2, 3, 4)
.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, "==aBoolean" + aBoolean);
    }
});

打印结果：

05-26 09:39:51.644 1482-1482/com.example.rxjavademo D/chan: ==================aBoolean true

6.2 takeWhile()

方法预览

public final Observable<T> takeWhile(Predicate<? super T> predicate)

有什么用？

可以设置条件，当某个数据满足条件时就会发送该数据，反之则不发送。

怎么用？

Observable.just(1, 2, 3, 4)
.takeWhile(new Predicate<Integer>(){
    @Override
    public boolean test(Integer integer) throws Exception{
        return integer < 3;
    }
})
.subscribe(new Comsumer<Integer>(){
    @Override
    public void accept(Integer integer) throws Exception{
        Log.d(TAG, "==integer" + integer);
    }
});

打印结果：

05-26 09:43:14.634 3648-3648/com.example.rxjavademo D/chan: ========================integer 1
========================integer 2

6.3 skipWhile()

方法预览

public final Observable<T> skipWhile(Predicate<? super T> predicate)

有什么用？

可以设置条件，当某个数据满足条件时不发送该数据，反之则发送。

怎么用？

Observable.just(1, 2, 3, 4)
.skipWhile(new Predicate<Integer> (){
    @Overide
    public boolean test(Integer i nteger) throws Exception{
        return integer < 3;
    }
})
.subscribe(new Consumer<Integer> (){
    @Override
    public void accept(new Consumer<Integer>(){
        @Override
        public void accept(Integer integer) throws Exception{
            Log.d(TAG, "==integer" + integer);
        }
    });

6.4 takeUntil()

方法预览

public final Observable<T> takeUntil(Predicate<? super T> stopPredicate)

有什么用？

可以设置条件，当事件满足此条件时，下一次的事件就不会被发送了。

怎么用？

Observable.just(1, 2, 3, 4, 5, 6)
.takeUntil(new Predicate<Integer> (){
    @Override
    public boolean test(Integer integer) throws Exception{
        return integer > 3;
    }
})
.subscribe(new Consumer<Integer>(){
    @Override
    public void accept(Integer integer) throws Exception{
        Log.d(TAG, "==integer" + integer);
    }
});

打印结果：

05-26 09:55:12.918 7933-7933/com.example.rxjavademo D/chan: ========================integer 1
========================integer 2
05-26 09:55:12.919 7933-7933/com.example.rxjavademo D/chan: ========================integer 3
========================integer 4

6.5 skipUntil()

方法预览

public final <U> Observable<T> skipUntil(ObservableSource<U> other)

有什么用？

当skipUntil（）中的Observable发送事件了，原来的Observable才会发送事件给观察者。

怎么用？

Observable.intervalRange(1, 5, 0, 1, TimeUnit.SECONDS)
.skipUntil(Observable.intervalRange(6, 5, 3, 1, TimeUnit.SECONDS))
.subscribe(new Observer<Long>(){
    @Override
    public void onSubscribe(Disposable d){
        Log.d(TAG, "==onSubscribe");
    }
    
    @Override
    public void onNext(Long along){
     Log.d(TAG, "==onNext" + aLong);   
    }
    
    @Override
    public void onError(Throwable e){
        Log.d(TAG, "==onError");
    }
    
    @Override
    public void onComplete(){
        Log.d(TAG, "==onComplete");
    }
});

打印结果：

05-26 10:08:50.574 13023-13023/com.example.rxjavademo D/chan: ========================onSubscribe 
05-26 10:08:53.576 13023-13054/com.example.rxjavademo D/chan: ========================onNext 4
05-26 10:08:54.576 13023-13054/com.example.rxjavademo D/chan: ========================onNext 5
========================onComplete

从结果可以看出，skipUntil()里的Observable并不会发送事件给观察者。

6.6 suquenceEqual()

方法预览

public static <T> Single<Boolean> sequenceEqual(ObservableSource<? extends T> source1, ObservableSource<? extends T> source2)
......

有什么用？

判断两个Observable发送的事件是否相同。

怎么用？

Observable.sequenceEqual(Observable.just(1, 2, 3),
Observable.just(1, 2, 3))
.subscribe(new Consumer<Boolean>(){
    @Override
    public void accept(Boolean aBoolean) throws Exception{
        Log.d(TAG, "==onNext" + aBoolean);
    }
});

打印结果：

05-26 10:11:45.975 14157-14157/? D/chan: ========================onNext true

6.7 contains()

方法预览

public final Single<Boolean> contains(final Object element)

有什么用？

判断事件序列中是否含有某个元素，如果有则返回true，如果没有则返回false。

怎么用？

Observable.just(1, 2, 3)
.contains(3)
.subscribe(new Consumer<Boolean>(){
    @Override
    public void accept(Boolean aBoolean) throws Exception{
        Log.d(TAG, "==onNext" + aBoolean);
    }
});

打印结果：

05-26 10:14:23.522 15085-15085/com.example.rxjavademo D/chan: ========================onNext true

6.8 isEmpty()

方法预览

public final Single<Boolean> isEmpty();

有什么用？

判断事件序列是否为空。

怎么用？

Observable.create(new ObservableOnSubscribe<Integer>(){
    @Override
    public void subscribe(ObservableEmitter<Integer> e) throws Exception{
        e.onComplete();
    }
})
.isEmpty()
.subscribe(new Consumer<Boolean>(){
    @Override
    public void accept(Boolean aBoolean) throws Exception{
        Log.d(TAG, "==onNext" + aBoolean);
    }
});

打印结果：

05-26 10:17:16.725 16109-16109/com.example.rxjavademo D/chan: ========================onNext true

6.9 amb()

方法预览

public static <T> Observable<T> amb(Iterable<? extends ObservableSource<? extends T>> sources)

有什么用？

amb()要传入一个Observable集合，但是只会发送最先发送事件的Observable中的事件，其余Observable将会被丢弃。

怎么用？

ArrayList<Observable<Long>> list = new ArrayList<>();

list.add(Observable.intervalRange(1, 5, 2, 1, TimeUnit.SECONDS));
list.add(Observable.intervalRange(6, 5, 0, 1, TimeUnit.SECOND));

Observable.amb(list)
.subscribe(new Consumer<Long>(){
    @Override
    public void accept(Long aLong) throws Exception{
        Log.d(TAG, "==aLong" + aLong);
    }
});

打印结果：

05-26 10:21:29.580 17185-17219/com.example.rxjavademo D/chan: ========================aLong 6
05-26 10:21:30.580 17185-17219/com.example.rxjavademo D/chan: ========================aLong 7
05-26 10:21:31.579 17185-17219/com.example.rxjavademo D/chan: ========================aLong 8
05-26 10:21:32.579 17185-17219/com.example.rxjavademo D/chan: ========================aLong 9
05-26 10:21:33.579 17185-17219/com.example.rxjavademo D/chan: ========================aLong 10

6.10 defaultIfEmpty()

方法预览

public final Observable<T> defaultIfEmpty(T defaultItem)

有什么用？

如果观察者只发送一个onComplete（）事件，则可以利用这个方法发送一个值。

怎么用？

Observable.create(new ObservableOnSubscribe<Integer> (){
    @Override
    public void subscribe(ObservableEmitter <integer> e) throws Exception{
        e.onComplete();
    }
})
.defaultIfEmpty(666)
.subscribe(new Consumer<Integer integer> throws Exception){
    Log.d(TAG, "==onNext" + integer);
    }
});

打印结果：

05-26 10:26:56.376 19249-19249/com.example.rxjavademo D/chan: ========================onNext 666

Over！