RxJava开发精要5 - Observables变换

发表于 5年以前 | 总阅读数：3633 次

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

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

*map家族

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

Map

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

考虑我们已安装的应用列表。我们怎么才能够显示同样的列表，但是所有的名字都是小写。

我们的loadList()函数可以改成这样：

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发射的每一项数据都应用一个函数，它将函数的结果填充回可观测序列，等待和下一次发射的数据一起使用。

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

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);
            }
        });

我们得到的结果是：

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

我们也可以创建一个新版本的loadList()函数用来比较每个安装应用的名字从而创建一个名字长度递增的列表。

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()函数中引入了一个新的元素：

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将所有的联系起来，像通常一样然后订阅它：

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