RxJava开发精要4 – Observables过滤

• 原文出自《RxJava Essentials》
• 原文做者 : Ivan Morgillo
• 译文出自 : 开发技术前线 www.devtf.cn
• 转载声明: 本译文已受权开发者头条享有独家转载权，未经容许，不得转载!
• 译者 : yuxingxin
• 项目地址 : RxJava-Essentials-CN

过滤Observables

在上一章中，咱们学习了使用RxJava建立一个Android工程以及如何建立一个可观测的列表来填充RecyclerView。咱们如今知道了如何从头、从列表、从一个已存在的传统Java函数来建立Observable。

这一章中，咱们将研究可观测序列的本质：过滤。咱们将学到如何从发射的Observable中选取咱们想要的值，如何获取有限个数的值，如何处理溢出的场景，以及更多的有用的技巧。

过滤序列

RxJava让咱们使用filter()方法来过滤咱们观测序列中不想要的值，在上一章中，咱们在几个例子中使用了已安装的应用列表，可是咱们只想展现以字母C开头的已安装的应用该怎么办呢？在这个新的例子中，咱们将使用一样的列表，可是咱们会过滤它，经过把合适的谓词传给filter()函数来获得咱们想要的值。

上一章中loadList()函数能够改为这样：

 

 

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

private void loadList(List<AppInfo> apps) {     mRecyclerView.setVisibility(View.VISIBLE);     Observable.from(apps)             .filter((appInfo) ->             appInfo.getName().startsWith("C"))             .subscribe(new Observable<AppInfo>() {                   @Override                 public void onCompleted() {                     mSwipeRefreshLayout.setRefreshing(false);                 }                   @Override                 public void onError(Throwable e) {                     Toast.makeText(getActivity(), "Something went wrong!", Toast.LENGTH_SHORT).show();                     mSwipeRefreshLayout.setRefreshing(false);                 }                   @Override                 public void onNext(AppInfo appInfo) {                     mAddedApps.add(appInfo);                     mAdapter.addApplication(mAddedApps.size() - 1,appInfo);                 }             }); }

 

咱们从上一章中的loadList()函数中添加下面一行：

 

 

1 2	.fliter((appInfo -> appInfo.getName().startsWith("C"))

 

建立Observable完之后，咱们从发出的每一个元素中过滤掉开头字母不是C的。为了让这里更清楚一些，咱们用Java 7的语法来实现：

 

 

1 2 3 4 5 6 7

.filter(new Func1<AppInfo,Boolean>(){     @Override     public Boolean call(AppInfo appInfo){         return appInfo.getName().startsWith("C");     } })

 

咱们传一个新的Func1对象给filter()函数，即只有一个参数的函数。Func1有一个AppInfo对象来做为它的参数类型而且返回Boolean对象。只要条件符合filter()函数就会返回true。此时，值会发射出去而且全部的观察者都会接收到。

正如你想的那样，从一个咱们获得的可观测序列中建立一个咱们须要的序列filter()是很好用的。咱们不须要知道可观测序列的源或者为何发射这么多不一样的数据。咱们只是想要这些元素的子集来建立一个能够在应用中使用的新序列。这种思想促进了咱们编码中的分离性与抽象性。

filter()函数最经常使用的用法之一时过滤null对象：

 

 

1 2 3 4 5 6 7

.filter(new Func1<AppInfo,Boolean>(){     @Override     public Boolean call(AppInfo appInfo){         return appInfo != null;     } })

 

这看起来简单，对于简单的事情有许多模板代码，可是它帮咱们免去了在onNext()函数调用中再去检测null值，让咱们把注意力集中在应用业务逻辑上。

下图展现了过滤出的C字母开头的已安装的应用列表。

获取咱们须要的数据

当咱们不须要整个序列时，而是只想取开头或结尾的几个元素，咱们能够用take()或takeLast()。

Take

若是咱们只想要一个可观测序列中的前三个元素那将会怎么样，发射它们，而后让Observable完成吗？take()函数用整数N来做为一个参数，从原始的序列中发射前N个元素，而后完成：

 

 

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

private void loadList(List<AppInfo> apps) {     mRecyclerView.setVisibility(View.VISIBLE);     Observable.from(apps)             .take(3)             .subscribe(new Observable<AppInfo>() {                   @Override                 public void onCompleted() {                     mSwipeRefreshLayout.setRefreshing(false);                 }                   @Override                 public void onError(Throwable e) {                     Toast.makeText(getActivity(), "Something went wrong!", Toast.LENGTH_SHORT).show();                     mSwipeRefreshLayout.setRefreshing(false);                 }                   @Override                 public void onNext(AppInfo appInfo) {                     mAddedApps.add(appInfo);                     mAdapter.addApplication(mAddedApps.size() - 1,appInfo);                 }             }); }

 

下图中展现了发射数字的一个可观测序列。咱们对这个可观测序列应用take(2)函数，而后咱们建立一个只发射可观测源的第一个和第二个数据的新序列。

TakeLast

若是咱们想要最后N个元素，咱们只需使用takeLast()函数：

 

 

1 2 3 4

Observable.from(apps)         .takeLast(3)         .subscribe(https://github.com/yuxingxin/RxJava-Essentials-CN/raw/master.);

 

正如听起来那样不值一提，重点注意takeLast()函数因为用一组有限的发射数的本质使得它仅可用于完成的序列。

下图中展现了如何从可观测源中发射最后一个元素来建立一个新的序列：

下图中展现了咱们在已安装的应用列表使用take()和takeLast()函数后发生的结果：

有且仅有一次

一个可观测序列会在出错时重复发射或者被设计成重复发射。distinct()和distinctUntilChanged()函数能够方便的让咱们处理这种重复问题。

Distinct

若是咱们想对一个指定的值仅处理一次该怎么办？咱们能够对咱们的序列使用distinct()函数去掉重复的。就像takeLast()同样，distinct()做用于一个完整的序列，而后获得重复的过滤项，它须要记录每个发射的值。若是你在处理一大堆序列或者大的数据记得关注内存使用状况。

下图展现了如何在一个发射1和2两次的可观测源上建立一个无重的序列：

为了建立咱们例子中序列，咱们将使用咱们至今已经学到的几个方法：
* take()：它有一小组的可识别的数据项。
* repeat()：建立一个有重复的大的序列。

而后，咱们将应用distinct()函数来去除重复。

注意

咱们用程序实现一个重复的序列，而后过滤出它们。这听起来时难以想象的，可是为了实现这个例子来使用咱们至今为止已学习到的东西则是个不错的练习。

 

 

1 2 3 4

Observable<AppInfo> fullOfDuplicates = Observable.from(apps)     .take(3)     .repeat(3);

 

fullOfDuplicates变量里把咱们已安装应用的前三个重复了3次：有9个而且许多重复的。而后，咱们使用distinct():

 

 

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22

fullOfDuplicates.distinct()             .subscribe(new Observable<AppInfo>() {                   @Override                 public void onCompleted() {                     mSwipeRefreshLayout.setRefreshing(false);                 }                   @Override                 public void onError(Throwable e) {                     Toast.makeText(getActivity(), "Something went wrong!", Toast.LENGTH_SHORT).show();                     mSwipeRefreshLayout.setRefreshing(false);                 }                   @Override                 public void onNext(AppInfo appInfo) {                     mAddedApps.add(appInfo);                     mAdapter.addApplication(mAddedApps.size() - 1,appInfo);                 }             }); }

 

结果，很明显，咱们获得：

DistinctUntilsChanged

若是在一个可观测序列发射一个不一样于以前的一个新值时让咱们获得通知这时候该怎么作？咱们猜测一下咱们观测的温度传感器，每秒发射的室内温度：

 

 

1 2

21°https://github.com/yuxingxin/RxJava-Essentials-CN/raw/master.21°https://github.com/yuxingxin/RxJava-Essentials-CN/raw/master.21°https://github.com/yuxingxin/RxJava-Essentials-CN/raw/master.21°https://github.com/yuxingxin/RxJava-Essentials-CN/raw/master.22°https://github.com/yuxingxin/RxJava-Essentials-CN/raw/master.

 

每次咱们得到一个新值，咱们都会更新当前正在显示的温度。咱们出于系统资源保护并不想在每次值同样时更新数据。咱们想忽略掉重复的值而且在温度确实改变时才想获得通知。ditinctUntilChanged()过滤函数能作到这一点。它能轻易的忽略掉全部的重复而且只发射出新的值。

下图用图形化的方式展现了咱们如何将distinctUntilChanged()函数应用在一个存在的序列上来建立一个新的不重复发射元素的序列。

First and last

下图展现了如何从一个从可观测源序列中建立只发射第一个元素的序列。

first()方法和last()方法很容易弄明白。它们从Observable中只发射第一个元素或者最后一个元素。这两个均可以传Func1做为参数，：一个能够肯定咱们感兴趣的第一个或者最后一个的谓词：

下图展现了last()应用在一个完成的序列上来建立一个仅仅发射最后一个元素的新的Observable。

与first()和last()类似的变量有：firstOrDefault()和lastOrDefault().这两个函数当可观测序列完成时再也不发射任何值时用得上。在这种场景下，若是Observable再也不发射任何值时咱们能够指定发射一个默认的值

Skip and SkipLast

下图中展现了如何使用skip(2)来建立一个不发射前两个元素而是发射它后面的那些数据的序列。

skip()和skipLast()函数与take()和takeLast()相对应。它们用整数N做参数，从本质上来讲，它们不让Observable发射前N个或者后N个值。若是咱们知道一个序列以没有太多用的“可控”元素开头或结尾时咱们可使用它。

下图与前一个场景相对应：咱们建立一个新的序列，它会跳事后面两个元素从源序列中发射剩下的其余元素。

ElementAt

若是咱们只想要可观测序列发射的第五个元素该怎么办？elementAt()函数仅从一个序列中发射第n个元素而后就完成了。

若是咱们想查找第五个元素可是可观测序列只有三个元素可供发射时该怎么办？咱们可使用elementAtOrDefault()。下图展现了如何经过使用elementAt(2)从一个序列中选择第三个元素以及如何建立一个只发射指定元素的新的Observable。

Sampling

让咱们再回到那个温度传感器。它每秒都会发射当前室内的温度。说实话，咱们并不认为温度会变化这么快，咱们可使用一个小的发射间隔。在Observable后面加一个sample()，咱们将建立一个新的可观测序列，它将在一个指定的时间间隔里由Observable发射最近一次的数值：

 

 

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21

Observable<Integer> sensor = [...]   sensor.sample(30,TimeUnit.SECONDS)     .subscribe(new Observable<Integer>() {           @Override         public void onCompleted() {           }           @Override         public void onError(Throwable e) {           }           @Override         public void onNext(Integer currentTemperature) {             updateDisplay(currentTemperature)         }     });

 

例子中Observable将会观测温度Observable而后每隔30秒就会发射最后一个温度值。很明显，sample()支持所有的时间单位：秒，毫秒，天，分等等。

下图中展现了一个间隔发射字母的Observable如何采样一个发射数字的Observable。Observable的结果将会发射每一个已发射字母的最后一组数据：1，4，5.

若是咱们想让它定时发射第一个元素而不是最近的一个元素，咱们可使用throttleFirst()。

Timeout

假设咱们工做的是一个时效性的环境，咱们温度传感器每秒都在发射一个温度值。咱们想让它每隔两秒至少发射一个，咱们可使用timeout()函数来监听源可观测序列,就是在咱们设定的时间间隔内若是没有获得一个值则发射一个错误。咱们能够认为timeout()为一个Observable的限时的副本。若是在指定的时间间隔内Observable不发射值的话，它监听的原始的Observable时就会触发onError()函数。

 

 

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20

Subscription subscription = getCurrentTemperature()     .timeout(2,TimeUnit.SECONDS)     .subscribe(new Observable<Integer>() {           @Override         public void onCompleted() {           }           @Override         public void onError(Throwable e) {             Log.d("RXJAVA","You should go check the sensor, dude");         }           @Override         public void onNext(Integer currentTemperature) {             updateDisplay(currentTemperature)         }     });

 

和sample()同样，timeout()使用TimeUnit对象来指定时间间隔。

下图中展现了一旦Observable超过了限时就会触发onError()函数：由于超时后它才到达，因此最后一个元素将不会发射出去。

Debounce

debounce()函数过滤掉由Observable发射的速率过快的数据；若是在一个指定的时间间隔过去了仍旧没有发射一个，那么它将发射最后的那个。

就像sample()和timeout()函数同样，debounce()使用TimeUnit对象指定时间间隔。

下图展现了多久从Observable发射一次新的数据，debounce()函数开启一个内部定时器，若是在这个时间间隔内没有新的数据发射，则新的Observable发射出最后一个数据：

总结

这一章中，咱们学习了如何过滤一个可观测序列。咱们如今可使用filter()，skip()，和sample()来建立咱们想要的Observable。

下一章中，咱们将学习如何转换一个序列，将函数应用到每一个元素，给它们分组和扫描来建立咱们所须要的能完成目标的特定Observable。