如何防止可观察对象将计时器迭代排队

Question

我正在使用以下观察值来执行常规任务。Observable在类首次加载到内存中时启动，然后定期执行我的代码。因为它是一个单一的可观察对象，所以从我的测试看，它肯定是一个假设，即代码永远不会第二次启动并并行处理，以防它运行的时间超过间隔。

private static Subscription subscription = Observable.timer(0, 1000, TimeUnit.MILLISECONDS)
        .flatMap(new Func1<Long, Observable<String>>() {
            @Override public Observable<String> call(Long aLong) {

                // some code

                return Observable.just(null);
            }
        }).subscribeOn(Schedulers.newThread()).observeOn(Schedulers.newThread())
        .subscribe();

但这也有一个缺点，即rxjava累积延迟发射，并在延迟迭代完成后以快速顺序启动它们。示例:如果定时器被编程为每1000毫秒迭代一次，并且迭代n花费5000毫秒，则迭代n+1、n+2、n+3等依次启动，但不遵循定时器间隔。

并不是那么糟糕，但真正的问题是当Android睡眠几个小时后会发生什么。因为一旦设备唤醒，rxjava就会快速启动所有未命中的迭代，这会对性能造成相当大的影响。

我如何告诉rxjava忘记遗漏的迭代？如果迭代花费的时间更长，我希望定时器在该迭代结束时启动，或者我希望丢弃错过的迭代并在到期时开始下一次迭代。我试着使用sample()和其他过滤器，但不知何故没有给我想要的效果，或者我不知道如何正确地应用它们。

注意，我不想为每个迭代创建一个新的可观察对象(我可以使用zip )，因为我想确保代码不是从多个线程执行的。

Answer 1

当你的Android设备处于睡眠模式时，不要要求Rx跳过某些事件。但是阻止Rx！

当您订阅一个流时，您需要一个处理程序来取消对该流的订阅。

Subscription subscription = Observable.timer(1, SECONDS).subscribe();

在activity的OnPause()方法中，可以通过对订阅调用unsubscribe()方法来停止流。

@Override
public void onPause() {
    subscription.unsubscribe();
}

在活动的onResume()方法中，您可以再次订阅流

@Override
public void onResume() {
   subscription = Observable.timer(1, SECONDS).subscribe();
}