RxJava 2：重新订阅（或取消并再次订阅）链中的第二个无限可观察对象答案

【问题标题】：RxJava 2: Resubscribe (or Cancel and subscribe again) to second infinite observable in chainRxJava 2：重新订阅（或取消并再次订阅）链中的第二个无限可观察对象
【发布时间】：2018-03-11 04:00:20
【问题描述】：

我有两个无限的 observables（getLastNDaysItemsInfinite 和 listenToServerUpdates），它们被链接成一个。第一个发出第二个应该听的项目。

            repository
            .getLastNDaysItemsInfinite(4)
            .flatMap(items ->
                    Observable
                            .fromIterable(items)
                            .map(Item::getId)
                            .toList()
                            .flatMapObservable(ids ->
                                    repository
                                            .listenToServerUpdates(ids)
                                            .onErrorResumeNext(throwable -> {
                                                Log.w(TAG, "Error occurred: ", throwable);
                                                return Observable.empty();
                                            }))
            );

listenToServerUpdates observable 打开套接字连接并在我们取消订阅时关闭它。因此，我需要实现当 getLastNDaysItemsInfinite(4) 发出新的项目列表时，listenToServerUpdates 将关闭其套接字连接并打开新的行为。但现在它只是为来自 getLastNDaysItemsInfinite 的每个新项目包创建新的可观察的 listenToServerUpdates 并因此创建新的套接字连接。

当 getLastNDaysItemsInfinite 发出新项目时，我如何重新订阅或取消并再次订阅运行可观察的 listenToServerUpdates ？

提前致谢！

【问题讨论】：

标签： android sockets observable subscription rx-java2

【解决方案1】：

您可以使用switchMap 代替flatMap。

这样

 repository
            .getLastNDaysItemsInfinite(4)
            .switchMap(items ->
                Observable
                        .fromIterable(items)....

因此，每次来自getLastNDaysItemsInfinite 的新列表时，先前创建的Observables 都会取消订阅。 Here 是文档

【讨论】：