上一个完成后如何使用一个 rxjava observable？

Question

我有两个 RxJava Observables，我从第一个 observable 中获取一个数组列表，然后使用它从另一个 observable 中获取数据。

Observable<KarobarTvVod> observable1 = youtubeDataHelper.getTVData();
    observable1.subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .unsubscribeOn(Schedulers.io())
            .subscribe(new Subscriber<KarobarTvVod>() {
                @Override
                public void onCompleted() {

                }

                @Override
                public void onError(Throwable e) {
                    e.printStackTrace();
                }

                @Override
                public void onNext(KarobarTvVod karobarTvVod) {
                    Log.d(TAG, "onNext: size" + karobarTvVod.getEtag());
                    tvObjArrayList = new ArrayList<TVObj>();
                    for (int i = 0; i < karobarTvVod.getItems().size(); i++) {
                        TVObj tvObj = new TVObj();
                        tvObj.setVideoDate(karobarTvVod.getItems().get(i).getSnippet().getPublishedAt());
                        tvObj.setVideoIcon(karobarTvVod.getItems().get(i).getSnippet().getThumbnails().getHigh().getUrl());
                        tvObj.setVideoTitle(karobarTvVod.getItems().get(i).getSnippet().getTitle());
                        tvObj.setVideoID(karobarTvVod.getItems().get(i).getId().getVideoId());
                        tvObjArrayList.add(tvObj);
                    }


                }
            });



    Observable<YoutubeViews> observable2 = youtubeDataHelper.getTVDataViews(tvObjArrayList);
    observable2.subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .unsubscribeOn(Schedulers.io())
            .subscribe(new Subscriber<YoutubeViews>() {
                @Override
                public void onCompleted() {

                }

                @Override
                public void onError(Throwable e) {
                    Log.d(TAG, "onError: in 2nd obs");
                    e.printStackTrace();
                }

                @Override
                public void onNext(YoutubeViews youtubeViews) {
                    Log.d(TAG, "onNext: views" + youtubeViews.getEtag());
                    viewsList = new ArrayList<String>();
                    for (int i = 0; i < youtubeViews.getItems().size(); i++) {

                        viewsList.add(youtubeViews.getItems().get(i).getStatistics().getViewCount());
                    }
                    tvView.displayList(tvObjArrayList, viewsList);
                }
            });

这只是示例代码，当 tvObjArrayList 从第一个 Observable 填充到第二个 Observable 时，我需要传递它，这样做的最佳做法是什么？而且我在第一个 Observable 中使用了 for-loop，有没有更好的方法来使用 rxjava 来实现它？谢谢

Answer 1

您应该使用flatMap 运算符。没有比这更容易的了。

Observable<KarobarTvVod> observable1 = youtubeDataHelper.getTVData();
observable1.subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .unsubscribeOn(Schedulers.io())
        .flatMap(new Func1<KarobarTvVod, Observable<YoutubeViews>>() {
            @Override
            public Observable<YoutubeViews> call(KarobarTvVod karobarTvVod) {
                Log.d(TAG, "onNext: size" + karobarTvVod.getEtag());
                tvObjArrayList = new ArrayList<TVObj>();
                for (int i = 0; i < karobarTvVod.getItems().size(); i++) {
                    TVObj tvObj = new TVObj();
                    tvObj.setVideoDate(karobarTvVod.getItems().get(i).getSnippet().getPublishedAt());
                    tvObj.setVideoIcon(karobarTvVod.getItems().get(i).getSnippet().getThumbnails().getHigh().getUrl());
                    tvObj.setVideoTitle(karobarTvVod.getItems().get(i).getSnippet().getTitle());
                    tvObj.setVideoID(karobarTvVod.getItems().get(i).getId().getVideoId());
                    tvObjArrayList.add(tvObj);
                }
                return youtubeDataHelper.getTVDataViews(tvObjArrayList);
            }
        }).subscribe(new Subscriber<YoutubeViews>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "onError: in 1st or 2nd obs");
                e.printStackTrace();
            }

            @Override
            public void onNext(YoutubeViews youtubeViews) {
                Log.d(TAG, "onNext: views" + youtubeViews.getEtag());
                viewsList = new ArrayList<String>();
                for (int i = 0; i < youtubeViews.getItems().size(); i++) {

                    viewsList.add(youtubeViews.getItems().get(i).getStatistics().getViewCount());
                }
                tvView.displayList(tvObjArrayList, viewsList);
            }
        });

Answer 2

你可以使用运算符

toList().flatMap()

对于 Observable A，在 flatMap 函数中，为 Observable B 工作。

例如：

observableA
    .toList()
    .flatMap(observableB.subscribe())
    .subscribe()

Answer 3

我假设getTVData 和getTVDataViews 各自发出一个项目，然后调用onComplete。如果它是真的，那么下面的例子就有效。没有loop，只是纯粹的rx :)

//getTVData should emit one item and then call obComplete
//otherwise toList() will wait forever
service.getTVData()
        .flatMap(karobarTvVod -> Observable.from(karobarTvVod.getItems()))
        .map(item -> {
            TVObj tvObj = new TVObj();
            //set other fields
            //by the way, I recommend you to use immutable objects
            return tvObj;
        })
        .toList()
        //here we have List<TVObj>
        .flatMap(
                objs -> {
                    //getTVDataViews should emit one item and then call onComplete
                    //otherwise toList will wait forever
                    return service.getTVDataViews(objs)
                            .flatMap(youtubeViews -> Observable.from(youtubeViews.getItems()))
                            .map(integer -> integer.toString())
                            //after that we will have List<String>
                            .toList();
                },
                //a function that combines one item emitted by each of the source and collection Observables 
                // and returns an item to be emitted by the resulting Observable
                new Func2<List<TVObj>,List<String>,Pair<List<TVObj>,List<String>>>() {
                    @Override
                    public Pair<List<TVObj>, List<String>> call(List<TVObj> objs, List<String> strings) {
                        return new Pair(objs, strings);
                    }
                })
        .subscribe(pair -> tvView.displayList(pair.first, pair.second));

附言。虽然这种方法更简洁，但我相信 loop 用于创建 list 的项目更有效。

Answer 4

你需要在第一个的 onComplete 中订阅第二个 observable

    Observable<KarobarTvVod> observable1 = youtubeDataHelper.getTVData();
    Observable<YoutubeViews> observable2 = youtubeDataHelper.getTVDataViews(tvObjArrayList);

    observable1.subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .unsubscribeOn(Schedulers.io())
        .subscribe(new Subscriber<KarobarTvVod>() {
            @Override
            public void onCompleted() {
                 observable2.subscribeOn(Schedulers.io())
           .observeOn(AndroidSchedulers.mainThread())
           .unsubscribeOn(Schedulers.io())
           .subscribe(new Subscriber<YoutubeViews>() {
                   @Override
                   public void onCompleted() {

                   }

                   @Override
                   public void onError(Throwable e) {
                       Log.d(TAG, "onError: in 2nd obs");
                       e.printStackTrace();
                   }

                   @Override
                   public void onNext(YoutubeViews youtubeViews) {
                       Log.d(TAG, "onNext: views" + youtubeViews.getEtag());
                       viewsList = new ArrayList<String>();
                       for (int i = 0; i < youtubeViews.getItems().size(); i++) {

                           viewsList.add(youtubeViews.getItems().get(i).getStatistics().getViewCount       ());
                       }
                       tvView.displayList(tvObjArrayList, viewsList);
                   }
               });
            }

            @Override
            public void onError(Throwable e) {
                e.printStackTrace();
            }

            @Override
            public void onNext(KarobarTvVod karobarTvVod) {
                Log.d(TAG, "onNext: size" + karobarTvVod.getEtag());
                tvObjArrayList = new ArrayList<TVObj>();
                for (int i = 0; i < karobarTvVod.getItems().size(); i++) {
                    TVObj tvObj = new TVObj();
                    tvObj.setVideoDate(karobarTvVod.getItems().get(i).getSnippet().getPublishedAt());
                    tvObj.setVideoIcon(karobarTvVod.getItems().get(i).getSnippet().getThumbnails().getHigh().getUrl());
                    tvObj.setVideoTitle(karobarTvVod.getItems().get(i).getSnippet().getTitle());
                    tvObj.setVideoID(karobarTvVod.getItems().get(i).getId().getVideoId());
                    tvObjArrayList.add(tvObj);
                }


            }
        });

当然，为了使这段代码更具可读性，我会在 onComplete 方法上为 observable2.subsriberOn 使用 Consumer 函数