RxJava开发精要6 - 组合Observables

时间  2019-11-12
标签 rxjava 开发 组合 observables 栏目 Java 繁體版
原文   原文链接

上一章中。咱们学到怎样转换可观測序列。咱们也看到了map(),scan(),groupBY(),以及不少其它实用的函数的实际样例，它们帮助咱们操做Observable来建立咱们想要的Observable。java

本章中，咱们将研究组合函数并学习怎样同一时候处理多个Observables来建立咱们想要的Observable。git

Merge

在异步的世界经常会建立这种场景。咱们有多个来源但是仅仅想有一个结果：多输入，单输出。RxJava的merge()方法将帮助你把两个甚至不少其它的Observables合并到他们发射的数据里。下图给出了把两个序列合并在一个终于发射的Observable。github

正如你看到的那样，发射的数据被交叉合并到一个Observable里面。markdown

注意假设你同步的合并Observable，它们将链接在一块儿并且不会交叉。架构

像一般同样。咱们用咱们的App和已安装的App列表来建立了一个“真实世界”的样例。咱们还需要第二个Observable。咱们可以建立一个单独的应用列表而后逆序。固然没有实际的意义，仅仅是为了这个样例。第二个列表，咱们的loadList()函数像如下这样：并发

private void loadList(List<AppInfo> apps) {
    mRecyclerView.setVisibility(View.VISIBLE);
    List reversedApps = Lists.reverse(apps);
    Observable<AppInfo> observableApps =Observable.from(apps);
    Observable<AppInfo> observableReversedApps =Observable.from(reversedApps);
    Observable<AppInfo> mergedObserbable = Observable.merge(observableApps,observableReversedApps);

    mergedObserbable.subscribe(new Observer<AppInfo>(){
        @Override
        public void onCompleted() {
            mSwipeRefreshLayout.setRefreshing(false);
            Toast.makeText(getActivity(), "Here is the list!", Toast.LENGTH_LONG).show();
        }

        @Override
        public void onError(Throwable e) {
            Toast.makeText(getActivity(), "One of the two Observable threw an error!", Toast.LENGTH_SHORT).show();
            mSwipeRefreshLayout.setRefreshing(false);
        }

        @Override
        public void onNext(AppInfoappInfo) {
            mAddedApps.add(appInfo);
            mAdapter.addApplication(mAddedApps.size() - 1, appInfo);
        } 
    });
}

咱们建立了Observable和observableApps数据以及新的observableReversedApps逆序列表。使用Observable.merge()，咱们可以建立新的ObservableMergedObservable在单个可观測序列中发射源Observables发出的所有数据。app

正如你能看到的,每个方法签名都是同样的，所以咱们的观察者无需在乎不论什么不一样就可以复用代码。结果例如如下：异步

注意错误时的toast消息。你可以以为每个Observable抛出的错误将会打断合并。假设你需要避免这种状况。RxJava提供了mergeDelayError()，它能从一个Observable中继续发射数据即使是当中有一个抛出了错误。当所有的Observables都完毕时，mergeDelayError()将会发射onError()。例如如下图所看到的：ide

ZIP

咱们在处理多源时可能会带来这样一种场景：多从个Observables接收数据，处理它们。而后将它们合并成一个新的可观測序列来使用。RxJava有一个特殊的方法可以完毕：zip()合并两个或者多个Observables发射出的数据项，依据指定的函数Func*变换它们。并发射一个新值。下图展现了zip()方法怎样处理发射的“numbers”和“letters”而后将它们合并一个新的数据项：函数

对于“真实世界”的样例来讲，咱们将使用已安装的应用列表和一个新的动态的Observable来让样例变得有点有趣味。

Observable<Long> tictoc = Observable.interval(1, TimeUnit.SECONDS);

tictocObservable变量使用interval()函数每秒生成一个Long类型的数据：简单且高效。正如以前所说的，咱们需要一个Func对象。

因为它需要传两个參数，因此是Func2:

private AppInfo updateTitle(AppInfoappInfo, Long time) {
    appInfo.setName(time + " " + appInfo.getName());
    return appInfo;
}

现在咱们的loadList()函数变成这样：

private void loadList(List<AppInfo> apps) {
    mRecyclerView.setVisibility(View.VISIBLE);
    Observable<AppInfo> observableApp = Observable.from(apps);

    Observable<Long> tictoc = Observable.interval(1, TimeUnit.SECONDS);

    Observable.zip(observableApp, tictoc,
    (AppInfo appInfo, Long time) -> updateTitle(appInfo, time))
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new Observer<AppInfo>() {
        @Override
        public void onCompleted() {
            Toast.makeText(getActivity(), "Here is the list!", Toast.LENGTH_LONG).show();
        }

        @Override
        public void onError(Throwable e) {
            mSwipeRefreshLayout.setRefreshing(false);
            Toast.makeText(getActivity(), "Something went wrong!", Toast.LENGTH_SHORT).show();
        }

        @Override
        public void onNext(AppInfoappInfo) {
            if (mSwipeRefreshLayout.isRefreshing()) {
                mSwipeRefreshLayout.setRefreshing(false);
            } 
            mAddedApps.add(appInfo);
            int position = mAddedApps.size() - 1;
            mAdapter.addApplication(position, appInfo);
            mRecyclerView.smoothScrollToPosition(position);
        }
    });
}

正如你看到的那样。zip()函数有三个參数：两个Observables和一个Func2。

细致一看会发现observeOn()函数。

它将在下一章中解说：现在咱们可以小试一下。

结果例如如下：

Join

前面两个方法。zip()和merge()方法做用在发射数据的范畴内，在决定怎样操做值以前有些场景咱们需要考虑时间的。RxJava的join()函数基于时间窗体将两个Observables发射的数据结合在一块儿。

为了正确的理解上一张图。咱们解释下join()需要的參数：

  • 第二个Observable和源Observable结合。

  • Func1參数：在指定的由时间窗体定义时间间隔内，源Observable发射的数据和从第二个Observable发射的数据相互配合返回的Observable。
  • Func1參数：在指定的由时间窗体定义时间间隔内，第二个Observable发射的数据和从源Observable发射的数据相互配合返回的Observable。
  • Func2參数：定义已发射的数据怎样与新发射的数据项相结合。

  • 例如如下练习的样例。咱们可以改动loadList()函数像如下这样：
private void loadList(List<AppInfo> apps) {
    mRecyclerView.setVisibility(View.VISIBLE);

    Observable<AppInfo> appsSequence =
    Observable.interval(1000, TimeUnit.MILLISECONDS)
                .map(position -> {
                    return apps.get(position.intValue());
                });

    Observable<Long> tictoc = Observable.interval(1000,TimeUnit.MILLISECONDS);

    appsSequence.join(
        tictoc, 
        appInfo -> Observable.timer(2,TimeUnit.SECONDS),
        time -> Observable.timer(0, TimeUnit.SECONDS),
        this::updateTitle)
        .observeOn(AndroidSchedulers.mainThread())
        .take(10)
        .subscribe(new Observer<AppInfo>() {
            @Override
            public void onCompleted() {
                Toast.makeText(getActivity(), "Here is the list!", Toast.LENGTH_LONG).show();
            }

            @Override
            public void onError(Throwable e) {
                mSwipeRefreshLayout.setRefreshing(false); 
                Toast.makeText(getActivity(), "Something went wrong!", Toast.LENGTH_SHORT).show();
            }

            @Override
            public void onNext(AppInfoappInfo) {
                if (mSwipeRefreshLayout.isRefreshing()) {
                    mSwipeRefreshLayout.setRefreshing(false);
                } 
                mAddedApps.add(appInfo);
                int position = mAddedApps.size() - 1;
                mAdapter.addApplication(position, appInfo);
                mRecyclerView.smoothScrollToPosition(position);
            } 
        });
}

咱们有一个新的对象appsSequence，它是一个每秒从咱们已安装的app列表发射app数据的可观測序列。

tictoc这个Observable数据每秒仅仅发射一个新的Long型整数。为了合并它们，咱们需要指定两个Func1变量：

appInfo -> Observable.timer(2, TimeUnit.SECONDS)

time -> Observable.timer(0, TimeUnit.SECONDS)

上面描写叙述了两个时间窗体。

如下一行描写叙述咱们怎样使用Func2将两个发射的数据结合在一块儿。

this::updateTitle

结果例如如下：

它看起来有点乱，但是注意app的名字和咱们指定的时间窗体，咱们可以看到：一旦第二个数据发射了咱们就会将它与源数据结合，但咱们用同一个源数据有2秒钟。这就是为何标题反复数字累加的缘由。

值得一提的是，为了简单起见，也有一个join()操做符做用于字符串而后简单的和发射的字符串链接成终于的字符串。

combineLatest

RxJava的combineLatest()函数有点像zip()函数的特殊形式。

正如咱们已经学习的，zip()做用于近期未打包的两个Observables。相反。combineLatest()做用于近期发射的数据项：假设Observable1发射了A并且Observable2发射了B和C，combineLatest()将会分组处理AB和AC，例如如下图所看到的：

combineLatest()函数接受二到九个Observable做为參数，假设有需要的话或者单个Observables列表做为參数。

从以前的样例中把loadList()函数借用过来，咱们可以改动一下来用于combineLatest()实现“真实世界”这个样例：

private void loadList(List<AppInfo> apps) {
    mRecyclerView.setVisibility(View.VISIBLE);
    Observable<AppInfo> appsSequence = Observable.interval(1000, TimeUnit.MILLISECONDS)
              .map(position ->apps.get(position.intValue()));
    Observable<Long> tictoc = Observable.interval(1500, TimeUnit.MILLISECONDS);
    Observable.combineLatest(appsSequence, tictoc,
               this::updateTitle)
       .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Observer<AppInfo>() {

        @Override
        public void onCompleted() {
            Toast.makeText(getActivity(), "Here is the list!", Toast.LENGTH_LONG).show();
        }

        @Override
        public void onError(Throwable e) {
            mSwipeRefreshLayout.setRefreshing(false);
            Toast.makeText(getActivity(), "Something went wrong!", Toast.LENGTH_SHORT).show();
        }

        @Override
        public void onNext(AppInfoappInfo) {
            if (mSwipeRefreshLayout.isRefreshing()) {
                mSwipeRefreshLayout.setRefreshing(false);
            } 
            mAddedApps.add(appInfo);
            int position = mAddedApps.size() - 1;
            mAdapter.addApplication(position, appInfo);
            mRecyclerView.smoothScrollToPosition(position);
        } 
    });
}

这咱们使用了两个Observables：一个是每秒钟从咱们已安装的应用列表发射一个App数据，第二个是每隔1.5秒发射一个Long型整数。

咱们将他们结合起来并运行updateTitle()函数，结果例如如下：

正如你看到的。因为不一样的时间间隔，AppInfo对象如咱们所预料的那样有时候会反复。

And,Then和When

在未来另外一些zip()知足不了的场景。

如复杂的架构，或者是仅仅为了我的爱好。你可以使用And/Then/When解决方式。

它们在RxJava的joins包下，使用Pattern和Plan做为中介，将发射的数据集合并到一块儿。

咱们的loadList()函数将会被改动从这样：

private void loadList(List<AppInfo> apps) {

    mRecyclerView.setVisibility(View.VISIBLE);

    Observable<AppInfo> observableApp = Observable.from(apps);

    Observable<Long> tictoc = Observable.interval(1, TimeUnit.SECONDS);

    Pattern2<AppInfo, Long> pattern = JoinObservable.from(observableApp).and(tictoc); 

    Plan0<AppInfo> plan = pattern.then(this::updateTitle);

    JoinObservable
        .when(plan)
        .toObservable()
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Observer<AppInfo>() {

            @Override
            public void onCompleted() {
                Toast.makeText(getActivity(), "Here is the list!", Toast.LENGTH_LONG).show();
            }

            @Override
            public void onError(Throwable e) {
                mSwipeRefreshLayout.setRefreshing(false); 
                Toast.makeText(getActivity(), "Something went wrong!", Toast.LENGTH_SHORT).show();
            }

            @Override
            public void onNext(AppInfoappInfo) {
                if (mSwipeRefreshLayout.isRefreshing()) { 
                mSwipeRefreshLayout.setRefreshing(false);
                } 
                mAddedApps.add(appInfo);
                int position = mAddedApps.size() - 1;
                mAdapter.addApplication(position, appInfo); mRecyclerView.smoothScrollToPosition(position);
            } 
        });
}

和一般同样。咱们有两个发射的序列。observableApp，发射咱们安装的应用列表数据。tictoc每秒发射一个Long型整数。现在咱们用and()链接源Observable和第二个Observable。

JoinObservable.from(observableApp).and(tictoc);

这里建立一个pattern对象，使用这个对象咱们可以建立一个Plan对象:”咱们有两个发射数据的Observables,then()是作什么的？”

pattern.then(this::updateTitle);

现在咱们有了一个Plan对象并且当plan发生时咱们可以决定接下来发生的事情。

.when(plan).toObservable()

这时候，咱们可以订阅新的Observable，正如咱们老是作的那样。

Switch

有这样一个复杂的场景就是在一个subscribe-unsubscribe的序列里咱们可以从一个Observable本身主动取消订阅来订阅一个新的Observable。

RxJava的switch()，正如定义的，将一个发射多个Observables的Observable转换成另外一个单独的Observable。后者发射那些Observables近期发射的数据项。

给出一个发射多个Observables序列的源Observable，switch()订阅到源Observable而后開始发射由第一个发射的Observable发射的同样的数据。当源Observable发射一个新的Observable时，switch()立刻取消订阅前一个发射数据的Observable（所以打断了从它那里发射的数据流）而后订阅一个新的Observable，并開始发射它的数据。

StartWith

咱们已经学到怎样链接多个Observables并追加指定的值到一个发射序列里。

RxJava的startWith()是concat()的相应部分。正如concat()向发射数据的Observable追加数据那样，在Observable開始发射他们的数据以前。 startWith()经过传递一个參数来先发射一个数据序列。

总结

这章中。咱们学习了怎样将两个或者不少其它个Observable结合来建立一个新的可观測序列。

咱们将可以merge Observable。join Observables 。zip Observables 并在几种状况下把他们结合在一块儿。

下一章。咱们将介绍调度器，它将很是easy的帮助咱们建立主线程以及提升咱们应用程序的性能。咱们也将学习怎样正确的运行长任务或者I/O任务来得到更好的性能。

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程