如何停止可观察的间隔答案

【问题标题】：How to stop interval from observable如何停止可观察的间隔
【发布时间】：2020-01-29 10:12:24
【问题描述】：

我是 rxjava 的新手，我想每 2 秒执行一次轮询任务 50 次，如果任务中满足某些条件，它也可能终止，我正在尝试使用 Observable.interval 但我发现没有办法终止它除了抛出异常，还有其他运营商可以满足我的目标吗？顺便说一句，此功能作为 API 提供可观察对象，因此我无法通过取消订阅来控制订阅者和终止。

Observable.interval(timeout, interval, TimeUnit.SECONDS)
.flatmap(task - > task)

【问题讨论】：

标签： rx-java

【解决方案1】：

我猜Observable.takeUntil(stopPredicate) 或Observable.takeWhile(predicate) 可以帮助你：

Observable.interval(timeout, interval, TimeUnit.SECONDS) 
.takeWhile(val -> val < 42)

此处 observable 将在第 42 次尝试时终止

【讨论】：

什么是超时，这是执行的最大时间还是发出第一个值的初始延迟？

【解决方案2】：

您可以使用takeUntil 停止Observable.interval，如下所示：

Observable.interval(0, 1, TimeUnit.SECONDS)
            .observeOn(AndroidSchedulers.mainThread())
            .takeUntil(new Predicate<Long>() {
                @Override
                public boolean test(Long aLong) throws Exception {
                    return aLong == 10;
                }
            })
            .subscribe(new Consumer<Long>() {
                @Override
                public void accept(Long aLong) throws Exception {
                    Log.d(TAG, "Tick: " + aLong);
                }
            });

在本例中，10 秒后，observable 将停止。

【讨论】：

我们可以给这个停止事件添加回调吗？即它是onComplete吗？我想我正在寻找类似 doOnUnsbscribe() 或 doOnComplete() 的东西

【解决方案3】：

下面的代码将产生以下输出：

Starting interval..
Next tick is: 1
Next tick is: 2
Next tick is: 3
Next tick is: 4
Next tick is: 5
Interval complete

如果您取消注释此行：“// if(t==3) dispose();”它只会打印以下内容：

Starting interval..
Next tick is: 1
Next tick is: 2
Next tick is: 3

代码：

         Observable.interval(1, TimeUnit.SECONDS).take(5).subscribeWith(new DisposableObserver<Long>() {
        @Override public void onStart() {                
            // do whatever might be needed here to run once before
            // the interval begins
            System.out.println("Starting interval..");
        }
        @Override public void onNext(Long t) {
            System.out.println("Next tick is: "+t);
            // if( t==3 ) dispose();
        }
        @Override public void onError(Throwable t) {
            t.printStackTrace();
        }
        @Override public void onComplete() {
            // will be called once when all 5 ticks are completed
            // will NOT be called if dispose() has been called
               System.out.println("Interval complete");
        }
    });

【讨论】：