RxJava开发精要5 – Observables变换

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

在上一章中，咱们探索了RxJava通用过滤方法。咱们学习了如何使用filter()方法过滤咱们不须要的值，如何使用take()获得发射元素的子集，如何使用distinct()函数来去除重复的。咱们学习了如何使用timeout()，sample()，以及debounce()来利用时间。

这一章中，咱们将学习如何变换可观测序列来建立一个更好知足咱们需求的序列。

*map家族

RxJava提供了几个mapping函数：map(),flatMap(),concatMap(),flatMapIterable()以及switchMap().全部这些函数都做用于一个可观测序列，而后变换它发射的值，最后用一种新的形式返回它们。让咱们用“真实世界”合适的例子一个个的学习下。

Map

RxJava的map函数接收一个指定的Func对象而后将它应用到每个由Observable发射的值上。下图展现了如何将一个乘法函数应用到每一个发出的值上以此建立一个新的Observable来发射转换的数据。

考虑咱们已安装的应用列表。咱们怎么才可以显示一样的列表，可是全部的名字都是小写。

咱们的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 27 28 29 30 31 32 33

private void loadList(List<AppInfo> apps) {     mRecyclerView.setVisibility(View.VISIBLE);     Observable.from(apps)         .map(new Func1<AppInfo,AppInfo>(){             @Override             public Appinfo call(AppInfo appInfo){                 String currentName = appInfo.getName();                 String lowerCaseName = currentName.toLowerCase();                 appInfo.setName(lowerCaseName);                 return appInfo;             }         })         .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);             }         }); }

 

正如你看到的，像往常同样建立咱们发射的Observable，咱们加一个map调用，咱们能够建立一个简单的函数来更新AppInfo对象并提供一个名字小写的新版本给观察者。

FlatMap

在复杂的场景中，咱们有一个这样的Observable：它发射一个数据序列，这些数据自己也能够发射Observable。RxJava的flatMap()函数提供一种铺平序列的方式，而后合并这些Observables发射的数据，最后将合并后的结果做为最终的Observable。

当咱们在处理可能有大量的Observables时，重要是记住任何一个Observables发生错误的状况，flatMap()函数将会触发它本身的onError()函数并放弃整个链。

重要的一点是关于合并部分：它容许交叉。正如上图所示，这意味着flatMap()函数在最后的Observable中不可以保证源Observables确切的发射顺序。

ConcatMap

RxJava的concatMap()函数解决了flatMap()的交叉问题，提供了一种可以把发射的值连续在一块儿的铺平函数，而不是合并它们，以下图所示：

FlatMapIterable

做为*map家族的一员，flatMapInterable()和flatMap()很像。仅有的本质不一样是它将源数据两两结成对，而后生成Iterable而不是原始数据和生成的Observables。

SwitchMap

以下图所示，switchMap()和flatMap()很像，除了一点：当原始Observable发射一个新的数据（Observable）时，它将取消订阅并中止监视以前那个数据的Observable产生的Observable，并开始监视当前这一个。

Scan

RxJava的scan()函数能够看作是一个累加器函数。scan()函数对原始Observable发射的每一项数据都应用一个函数，它将函数的结果填充回可观测序列，等待和下一次发射的数据一块儿使用。

做为一个通用的例子，给出一个累加器：

 

 

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

Observable.just(1,2,3,4,5)         .scan((sum,item) -> sum + item)         .subscribe(new Subscriber<Integer>() {             @Override             public void onCompleted() {                 Log.d("RXJAVA", "Sequence completed.");             }               @Override             public void onError(Throwable e) {                 Log.e("RXJAVA", "Something went south!");             }               @Override             public void onNext(Integer item) {                 Log.d("RXJAVA", "item is: " + item);             }         });

 

咱们获得的结果是：

 

 

1 2 3 4 5 6 7

RXJAVA: item is: 1 RXJAVA: item is: 3 RXJAVA: item is: 6 RXJAVA: item is: 10 RXJAVA: item is: 15 RXJAVA: Sequence completed.

 

咱们也能够建立一个新版本的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 27 28 29 30 31 32

private void loadList(List<AppInfo> apps) {     mRecyclerView.setVisibility(View.VISIBLE);     Observable.from(apps)             .scan((appInfo,appInfo2) -> {                 if(appInfo.getName().length > appInfo2.getName().length()){                     return appInfo;                 } else {                     return appInfo2;                 }             })             .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);                 }             }); }

 

结果以下：

有一个scan()函数的变体，它用初始值做为第一个发射的值，方法特征就像：scan(R,Func2)，就像下图中的例子这样：

GroupBy

拿第一个例子开始，咱们安装的应用程序列表按照字母表的顺序排序。然而，若是如今咱们想按照最近更新日期来排序咱们的App时该怎么办？RxJava提供了一个有用的函数从列表中按照指定的规则：groupBy()来分组元素。下图中的例子展现了groupBy()如何将发射的值根据他们的形状来进行分组。

这个函数将源Observable变换成一个发射Observables的新的Observable。它们中的每个新的Observable都发射一组指定的数据。

为了建立一个分组了的已安装应用列表，咱们在loadList()函数中引入了一个新的元素：

 

 

1 2 3 4 5 6 7 8 9

Observable<GroupedObservable<String,AppInfo>> groupedItems = Observable.from(apps)     .groupBy(new Func1<AppInfo,String>(){         @Override         public String call(AppInfo appInfo){             SimpleDateFormat formatter = new SimpleDateFormat("MM/yyyy");             return formatter.format(new Date(appInfo.getLastUpdateTime()));         }     });

 

如今咱们建立了一个新的Observable，groupedItems，将会发射一个带有GroupedObservable的序列。GroupedObservable是一个特殊的Observable，它源自一个分组的key。在这个例子中，key就是String，表明的意思是Month/Year格式化的最近更新日期。

这一点，咱们已经建立了几个发射AppInfo数据的Observable，用来填充咱们的列表。咱们想保留字母排序和分组排序。咱们将建立一个新的Observable将全部的联系起来，像一般同样而后订阅它：

 

 

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

Observable.concat(groupedItems)     .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()函数完成了，结果是：

Buffer

RxJava中的buffer()函数将源Observable变换一个新的Observable，这个新的Observable每次发射一组列表值而不是一个一个发射。

上图中展现了buffer()如何将count做为一个参数来指定有多少数据项被包在发射的列表中。实际上，buffer()函数有几种变体。其中有一个时容许你指定一个skip值：此后每当收到skip项数据，用count项数据就填充缓存。以下图所示：

buffer()带一个timespan的参数，会建立一个每隔timespan时间段就会发射一个列表的Observable。

Window

RxJava的window()函数和buffer()很像，可是它发射的时Observable而不是列表。下图展现了window()如何缓存3个数据项并把它们做为一个新的Observable发射出去。

这些Observables中的每个都发射原始Observable数据的一个子集，数量由count指定,最后发射一个onCompleted()结束。正如buffer()同样,window()也有一个skip变体,以下图所示：

Cast

RxJava的cast()函数是本章中最后一个操做符。它是map()操做符的特殊版本。它将源Observable中的每一项数据都转换为新的类型，把它变成了不一样的Class。

总结

这一章中，咱们学习了RxJava时如何控制和转换可观测序列。用咱们如今所学的知识，咱们能够建立、过滤、转换咱们所想要的任何种类的可观测序列。

下一章，咱们将学习如何组合Observable，合并它们，链接它们，再或者打包它们。