RxJava操作符系列传送门

• RxJava操作符源码

https://github.com/xiehui999/fuseProgram

• RxJava操作符系列一
• RxJava操作符系列二

前言

在之前的文章，我们介绍了一些Observable的创建以及数据转换的操作符，其中的一些数据转换的操作符理解还是有一定的难度的，但是相信如果敲一遍代码并且修改各种参数的值，去观察执行的日志，相信还是很容易的理解的。在官网，每个操作符都给出了图例，如果你对文字的理解不够清楚明白，也可以去参考图示帮助自己理解。在这篇文章中，我们将介绍一些常见的过滤操作符，在RxJava中过滤操作符也是比较好理解的，好了，让我们一起继续开启学习之旅吧。

Filter

该操作符接收一个Func1参数，我们可以在其中通过运用你自己的判断条件去判断我们要过滤的数据，当数据通过判断条件后返回true表示发射该项数据，否则就不发射，这样就过滤出了我们想要的数据。如下，我们过滤出不能被2整除的数

1. Integer[] ints = {1, 2, 3, 4, 5, 6, 7, 8, 9}; 
2.  
3.        Observable observable = Observable.from(ints).filter(new Func1<Integer, Boolean>() { 
4.  
5.            @Override 
6.  
7.            public Boolean call(Integer integer) { 
8.  
9.                return integer % 2 != 0;//返回true，就不会过滤掉，会发射数据，过滤掉返回false的值 
10.  
11.            } 
12.  
13.        }); 
14.  
15.        Action1 action1 = new Action1<Integer>() { 
16.  
17.            @Override 
18.  
19.            public void call(Integer i) { 
20.  
21.                Log.e(TAG, "call: "+i ); 
22.  
23.            } 
24.  
25.        }; 
26.  
27.        observable.subscribe(action1);  

输出日志信息

1. call: 1 
2.  
3. call: 3 
4.  
5. call: 5 
6.  
7. call: 7 
8.  
9. call: 9  

ofType

该操作符是filter操作符的一个特殊形式。它过滤一个Observable只返回指定类型的数据，例如当数据源有字符串和int型数据时，我们想要过滤出字符串就可以使用这个操作符，如下示例代码

1. Observable.just(0, "one", 6, 4, "two", 8, "three", 1, "four", 0) 
2.  
3. .ofType(String.class) 
4.  
5. .subscribe(new Subscriber<String>() { 
6.  
7. @Override 
8.  
9. public void onCompleted() { 
10.  
11. Log.e(TAG, "onCompleted:ofType "); 
12.  
13. } 
14.  
15.   
16.  
17. @Override 
18.  
19. public void onError(Throwable e) { 
20.  
21. Log.e(TAG, "onError:ofType "); 
22.  
23. } 
24.  
25.   
26.  
27. @Override 
28.  
29. public void onNext(String string) { 
30.  
31. Log.e(TAG, "onNext:ofType " + string); 
32.  
33. } 
34.  
35. });  

输出日志信息

1. onNext:ofType one 
2.  
3. onNext:ofType two 
4.  
5. onNext:ofType three 
6.  
7. onNext:ofType four 
8.  
9. onCompleted:ofType  

当然除了过滤基本类型的数据，也可以过滤自定义类型数据。

First

如果我们只对Observable发射的第一项数据，或者满足某个条件的第一项数据感兴趣，则可以使用First操作符。

1. Observable.just(10, 11, 12, 13).first().subscribe(new Action1() { 
2.  
3.             @Override 
4.  
5.             public void call(Integer integer) { 
6.  
7.                 Log.e(TAG, integer+""); 
8.  
9.             } 
10.  
11.         });  

上面日志只打印一个值10，当然我们也可以给first传一个参数Fun1，指定一个条件如下

1. Observable.just(10, 11, 12, 13).first(new Func1<Integer, Boolean>() { 
2.  
3.             @Override 
4.  
5.             public Boolean call(Integer integer) { 
6.  
7.                 return integer > 12; 
8.  
9.             } 
10.  
11.         }).subscribe(new Action1<Integer>() { 
12.  
13.             @Override 
14.  
15.             public void call(Integer integer) { 
16.  
17.                 Log.e(TAG, integer+""); 
18.  
19.             } 
20.  
21.         });  

此时输出的信息就是满足integer > 12的第一项数据13。

firstOrDefault

该操作符是first操作符的变形。主要是在没有发射任何数据时发射一个你在参数中指定的默认值。如下，它有有两个重载方法。

1. Observable.just(11,12,13).firstOrDefault(10).subscribe(new Action1<Object>() { 
2.  
3.             @Override 
4.  
5.             public void call(Object o) { 
6.  
7.                 Log.e(TAG, o.toString()); 
8.  
9.             } 
10.  
11.         });  

如果写成上面的代码，这个执行会和first效果一样。因为没有发射数据的时候才用到默认值，那么我们将上面代码更改如下，使用empty创建一个不发射任何数据但是正常终止的Observable。

1. Observable.empty().firstOrDefault(10).subscribe(new Action1<Object>() { 
2.  
3.             @Override 
4.  
5.             public void call(Object o) { 
6.  
7.                 Log.e(TAG, o.toString()); 
8.  
9.             } 
10.  
11.         });  

发现此时输出了数据10.该操作符还提供了两个参数的重载方法firstOrDefault(T defaultValue, Func1 super T, Boolean> predicate)。我们可以增加一个条件。如下示例

1. Observable.just(10，13,16).firstOrDefault(15, new Func1<Integer, Boolean>() { 
2.  
3.             @Override 
4.  
5.             public Boolean call(Integer integer) { 
6.  
7.                 return integer>20; 
8.  
9.             } 
10.  
11.         }).subscribe(new Action1<Integer>() { 
12.  
13.             @Override 
14.  
15.             public void call(Integer integer) { 
16.  
17.                 Log.e(TAG, ""+integer); 
18.  
19.             } 
20.  
21.         });  

此时数据源10,13,16都不满足大于20,则此时将输出默认值15，如果我们将数据源数据增加一个值22.那么此时将不再输出默认值，而是输出22。

takeFirst

该操作符与first操作符的区别就是如果原始Observable没有发射任何满足条件的数据，first会抛出一个NoSuchElementException直接执行onError()，而takeFist会返回一个空的Observable(不调用onNext()但是会调用onCompleted)

如下面下面示例代码

1. Observable.just(10,11).filter(new Func1<Integer, Boolean>() { 
2.  
3.             @Override 
4.  
5.             public Boolean call(Integer integer) { 
6.  
7.                 return integer>20; 
8.  
9.             } 
10.  
11.         }).first().subscribe(new Subscriber<Object>() { 
12.  
13.   
14.  
15.             @Override 
16.  
17.             public void onCompleted() { 
18.  
19.                 Log.e(TAG, "onCompleted: "); 
20.  
21.             } 
22.  
23.   
24.  
25.             @Override 
26.  
27.             public void onError(Throwable e) { 
28.  
29.                 Log.e(TAG, "onError: "+e.toString()); 
30.  
31.             } 
32.  
33.   
34.  
35.             @Override 
36.  
37.             public void onNext(Object o) { 
38.  
39.                 Log.e(TAG, "onNext: "+o.toString()); 
40.  
41.             } 
42.  
43.         });  

执行后输出的信息如下

1. onError: java.util.NoSuchElementException: Sequence contains no elements 

若此时用takeFirst

1. Observable.just(10,11).takeFirst(new Func1<Integer, Boolean>() { 
2.  
3.             @Override 
4.  
5.             public Boolean call(Integer integer) { 
6.  
7.                 Log.e(TAG, "call: takeFirst" ); 
8.  
9.                 return integer>30; 
10.  
11.             } 
12.  
13.         }).subscribe(new Subscriber<Object>() { 
14.  
15.   
16.  
17.             @Override 
18.  
19.             public void onCompleted() { 
20.  
21.                 Log.e(TAG, "onCompleted: "); 
22.  
23.             } 
24.  
25.   
26.  
27.             @Override 
28.  
29.             public void onError(Throwable e) { 
30.  
31.                 Log.e(TAG, "onError: "+e.toString()); 
32.  
33.             } 
34.  
35.   
36.  
37.             @Override 
38.  
39.             public void onNext(Object o) { 
40.  
41.                 Log.e(TAG, "onNext: "+o.toString()); 
42.  
43.             } 
44.  
45.         });  

发现此时不会出现异常，而是执行了onCompleted()。

single

如果原始Observable在完成之前不是正好发射一次数据，它会抛出一个NoSuchElementException，白话可以理解为发送数据是一项的话输出此项的值，若是多个数据则抛出异常执行onError()方法。

如下代码

1. Observable.just(10, 11, 12, 13).single().subscribe(new Subscriber<Integer>() { 
2.  
3.             @Override 
4.  
5.             public void onCompleted() { 
6.  
7.                  Log.e(TAG, "onCompleted"); 
8.  
9.             } 
10.  
11.   
12.  
13.             @Override 
14.  
15.             public void onError(Throwable e) { 
16.  
17.                  Log.e(TAG, "onError"+e.toString()); 
18.  
19.             } 
20.  
21.   
22.  
23.             @Override 
24.  
25.             public void onNext(Integer integer) { 
26.  
27.                  Log.e(TAG,  integer); 
28.  
29.             } 
30.  
31.         });  

输出信息

1. onError: java.util.NoSuchElementException: Sequence contains no elements 

如果将上述代码做下简单更改

1. Observable.just(10, 11, 12, 13).filter(new Func1<Integer, Boolean>() { 
2.  
3.             @Override 
4.  
5.             public Boolean call(Integer integer) { 
6.  
7.                 return integer > 12; 
8.  
9.             } 
10.  
11.         }).subscribe(new Subscriber<Integer>() { 
12.  
13.             @Override 
14.  
15.             public void onCompleted() { 
16.  
17.                  Log.e(TAG, "onCompleted"); 
18.  
19.             } 
20.  
21.   
22.  
23.             @Override 
24.  
25.             public void onError(Throwable e) { 
26.  
27.                  Log.e(TAG, "onError"+e.toString()); 
28.  
29.             } 
30.  
31.   
32.  
33.             @Override 
34.  
35.             public void onNext(Integer integer) { 
36.  
37.                  Log.e(TAG,  integer); 
38.  
39.             } 
40.  
41.         });  

此时会输出数据13,因为此时通过filter后就只有一条数据。single也有singleOrDefault(T)和singleOrDefault(T,Func1)两个变体，具体可以自己代码测试区别。

Last

该操作符与first意义相反，若我们只对Observable发射的最后一项数据，或者满足某个条件的最后一项数据感兴趣时使用该操作符。

示例代码

1. Observable.just(10, 11, 12, 13).last().subscribe(new Action1<Integer>() { 
2.  
3.             @Override 
4.  
5.             public void call(Integer integer) { 
6.  
7.                Log.e(TAG, "call: "+integer); 
8.  
9.             } 
10.  
11.         });  

执行后输出13.它有一个重载方法可以指定条件，获取满足条件的最后一项数据的。将上面代码修改如下

1. Observable.just(10, 11, 12, 13).last(new Func1<Integer, Boolean>() { 
2.  
3.             @Override 
4.  
5.             public Boolean call(Integer integer) { 
6.  
7.                 return integer < 12; 
8.  
9.             } 
10.  
11.         }).subscribe(new Action1<Integer>() { 
12.  
13.             @Override 
14.  
15.             public void call(Integer integer) { 
16.  
17.                 Log.e(TAG, "call: "+integer); 
18.  
19.             } 
20.  
21.         });  

此时最终输出数据就是11.该操作符和first一样也有几种变体，如lastOrDefault，TakeLast，具体效果可自己测试。

Skip

该操作符是跳过之前的前几项数据，然后再发射数据。

1. Observable.range(1, 10).skip(6).subscribe(new Action1<Integer>() { 
2.  
3.             @Override 
4.  
5.             public void call(Integer integer) { 
6.  
7.                Log.e(TAG, "call: "+integer ); 
8.  
9.             } 
10.  
11.         });  

输出日志信息

1. call: 7 
2.  
3. call: 8 
4.  
5. call: 9 
6.  
7. call: 10  

skip还有两个重载方法.skip(long time, TimeUnit unit)默认是在computation调度器上执行，如果要有更新UI操作需要通过observeOn方法指定为AndroidSchedulers.mainThread()，当然还有一个重载方法skip(long time, TimeUnit unit, Scheduler scheduler)可以指定调度器。注意的一点是这两个重载方法的第一个参数不是跳过的数据数量，指的是时间。

1. Observable.interval(500, TimeUnit.MILLISECONDS) 
2.  
3.                 .skip(2, TimeUnit.SECONDS) 
4.  
5.                 .observeOn(AndroidSchedulers.mainThread()) 
6.  
7.                 .subscribe(new Subscriber<Long>() { 
8.  
9.                     @Override 
10.  
11.                     public void onCompleted() { 
12.  
13.   
14.  
15.                     } 
16.  
17.   
18.  
19.                     @Override 
20.  
21.                     public void onError(Throwable e) { 
22.  
23.   
24.  
25.                     } 
26.  
27.   
28.  
29.                     @Override 
30.  
31.                     public void onNext(Long aLong) { 
32.  
33.                         tv.append("\n" + aLong); 
34.  
35.                         if (aLong > 10) { 
36.  
37.                             this.unsubscribe(); 
38.  
39.                         } 
40.  
41.                     } 
42.  
43.                 });  

如上代码，通过interval每隔500毫秒产生一个数据，通过skip设定跳过时间为2秒。并且当数据大于10时解除订阅。

skipLast

正好和skip 相反，忽略最后产生的n个数据项

1. Observable.range(1, 10).skipLast(6).subscribe(new Action1<Integer>() { 
2.  
3.             @Override 
4.  
5.             public void call(Integer integer) { 
6.  
7.                Log.e(TAG, "call: "+integer ); 
8.  
9.             } 
10.  
11.         });  

输出日志信息

1. call: 1 
2.  
3. call: 2 
4.  
5. call: 3 
6.  
7. call: 4  

接下文 

本文作者：佚名

来源：51CTO