RxJava 每 x 秒使用 onComplete 调用一次服务答案

【问题标题】：RxJava call service every x seconds with onCompleteRxJava 每 x 秒使用 onComplete 调用一次服务
【发布时间】：2017-07-23 07:11:39
【问题描述】：

我需要每隔 X 秒进行一次相同的网络调用，所以我使用了下面的代码

Observable.interval(X, TimeUnit.SECONDS)
                .flatMap(n -> myCall())
                .flatMap(response -> Observable.from(response.getList())
                .doOnNext(object -> createNewObject())
                .doOnCompleted(() -> myStuff())
                .subscribe()

但问题是我从来没有调用过 doOnCompleted。有没有办法每隔 X 秒继续调用我的服务，但在每次调用后都有一个 onComplete ？

【问题讨论】：

不可能。你为什么想要这样的东西？
我的服务返回给我一个对象列表，所以我想在我的对象列表上的迭代完成时更新我的数据。
在这种情况下，您应该使用 .doOnNext(() -> myStuff()) 而不是 .doOnCompleted(() -> myStuff())
我想用我的最终对象列表调用 myStuff 以避免多次执行相同的处理
最终名单是什么？你正在做无限的间隔。请描述你的场景要求，你想要达到什么目标？

标签： android rx-java retrofit2

【解决方案1】：

Observable.interval(X, TimeUnit.SECONDS)
            .flatMap(n -> myCall())
            .doOnNext(() -> myStuff())
            .subscribe()

对于网络调用，请使用 subscribeOn 和 observeOn，并执行 doOnError 动作 doOnError(() -> showNoInternetConnection())

doOnCompleted 从不要求间隔！！！

最终代码：

Observable.interval(X, TimeUnit.SECONDS)
            .flatMap(n -> myCall())
            .doOnNext(() -> myStuff())
            .doOnError(() -> showNoInternetConnection())
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe()

或者您可以使用订阅此婴儿车。

Observable.interval(X, TimeUnit.SECONDS)
            .flatMap(n -> myCall())
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(this::myStuff(), this::showNoInternetConnection())

【讨论】：

不幸的是，这并没有回答我的问题。有没有办法每 x 秒调用一次我的服务（因为 doOnCompleted 从未调用过，所以没有间隔）？
为什么需要 onComplete？您想对网络调用的每个结果执行某些操作，因此只需使用 doOnNext （或订阅）并按照此处的建议执行操作。
如果我想使用 onNext 我不会问这个问题

【解决方案2】：

很遗憾，没有。 Complete 事件by definition 表示流已关闭，不再有信号。

【讨论】：

好的，所以 Rx 完全不可能？

【解决方案3】：

您可以将流操作符移动到第一个 flatMap 中，在那里您将收到 onComplete 调用。

Observable.interval(X, TimeUnit.SECONDS)
            .flatMap(n -> myCall().flatMap(response -> Observable.from(response.getList())
                .doOnNext(object -> createNewObject())
                .doOnCompleted(() -> myStuff())
            )

【讨论】：

谢谢！这正是我想要的！